[Meta] Implement Shared Subscriptions

[andsel]

Add shared subscription support allowing for load balanced consumers of a subscription

MQTT specification reference

4.8.2 Shared Subscriptions

Shared Subscription Topic Filter definition

A Shared Subscription can be associated with multiple subscribing MQTT Sessions. Like a Non‑shared Subscription, it has a Topic Filter and Subscription Options; however, a publication that matches its Topic Filter is only sent to one of its subscribing Sessions. Shared Subscriptions are useful where several consuming Clients share the processing of the publications in parallel.
A Shared Subscription is identified using a special style of Topic Filter. The format of this filter is:
$share/{ShareName}/{filter}

• $share is a literal string that marks the Topic Filter as being a Shared Subscription Topic Filter.
• {ShareName} is a character string that does not include /, + or #
• {filter} The remainder of the string has the same syntax and semantics as a Topic Filter in a non-shared subscription. Refer to section 4.7.

A Shared Subscription's Topic Filter MUST start with $share/ and MUST contain a ShareName that is at least one character long [MQTT-4.8.2-1].
The ShareName MUST NOT contain the characters "/", "+" or "#", but MUST be followed by a "/" character. This "/" character MUST be followed by a Topic Filter [MQTT-4.8.2-2] as described in section 4.7.

Non-normative comment

Shared Subscriptions are defined at the scope of the MQTT Server, rather than of a Session. A ShareName is included in the Shared Subscription's Topic Filter so that there can be more than one Shared Subscription on a Server that has the same {filter} component. Typically, applications use the ShareName to represent the group of subscribing Sessions that are sharing the subscription.

Examples:

• Shared subscriptions $share/consumer1/sport/tennis/+ and $share/consumer2/sport/tennis/+ are distinct shared subscriptions and so can be associated with different groups of Sessions. Both of them match the same topics as a non-shared subscription to sport/tennis/+.
  If a message were to be published that matches sport/tennis/+ then a copy would be sent to exactly one of the Sessions subscribed to $share/consumer1/sport/tennis/+ , a separate copy of the message would be sent to exactly one of the Sessions subscribed to $share/consumer2/sport/tennis/+ and further copies would be sent to any Clients with non-shared subscriptions to sport/tennis/+
• Shared subscription $share/consumer1//finance matches the same topics as a non-shared subscription to /finance.
  Note that $share/consumer1//finance and $share/consumer1/sport/tennis/+ are distinct shared subscriptions, even though they have the same ShareName. While they might be related in some way, no specific relationship between them is implied by them having the same ShareName.

A Shared Subscription is created by using a Shared Subscription Topic Filter in a SUBSCRIBE request. So long as only one Session subscribes to a particular Shared Subscription, the shared subscription behaves like a non-shared subscription, except that:

• The $share and {ShareName} portions of the Topic Filter are not taken into account when matching against publications.
• No Retained Messages are sent to the Session when it first subscribes. It will be sent other matching messages as they are published.

Once a Shared Subscription exists, it is possible for other Sessions to subscribe with the same Shared Subscription Topic Filter. The new Session is associated with the Shared Subscription as an additional subscriber. Retained messages are not sent to this new subscriber. Each subsequent Application Message that matches the Shared Subscription is now sent to one and only one of the Sessions that are subscribed to the Shared Subscription.

A Session can explicitly detach itself from a Shared Subscription by sending an UNSUBSCRIBE Packet that contains the full Shared Subscription Topic Filter. Sessions are also detached from the Shared Subscription when they terminate.

A Shared Subscription lasts for as long as it is associated with at least one Session (i.e. a Session that has issued a successful SUBSCRIBE request to its Topic Filter and that has not completed a corresponding UNSUBSCRIBE). A Shared Subscription survives when the Session that originally created it unsubscribes, unless there are no other Sessions left when this happens. A Shared Subscription ends, and any undelivered messages associated with it are deleted, when there are no longer any Sessions subscribed to it.

Notes on Shared Subscriptions:

• If there's more than one Session subscribed to the Shared Subscription, the Server implementation is free to choose, on a message by message basis, which Session to use and what criteria it uses to make this selection.
• Different subscribing Clients are permitted to ask for different Requested QoS levels in their SUBSCRIBE packets. The Server decides which Maximum QoS to grant to each Client, and it is permitted to grant different Maximum QoS levels to different subscribers. When sending an Application Message to a Client, the Server MUST respect the granted QoS for the Client's subscription [MQTT-4.8.2-3], in the same that it does when sending a message to a ‑Subscriber.
• If the Server is in the process of sending a QoS 2 message to its chosen subscribing Client and the connection to the Client breaks before delivery is complete, the Server MUST complete the delivery of the message to that Client when it reconnects [MQTT-4.8.2-4] as described in section 4.3.3. If the Client's Session terminates before the Client reconnects, the Server MUST NOT send the Application Message to any other subscribed Client [MQTT-4.8.2-5].
• If the Server is in the process of sending a QoS 1 message to its chosen subscribing Client and the connection to that Client breaks before the Server has received an acknowledgement from the Client, the Server MAY wait for the Client to reconnect and retransmit the message to that Client. If the Client'sSession terminates before the Client reconnects, the Server SHOULD send the Application Message to another Client that is subscribed to the same Shared Subscription. It MAY attempt to send the message to another Client as soon as it loses its connection to the first Client.
• If a Client responds with a PUBACK or PUBREC containing a Reason Code of 0x80 or greater to a PUBLISH packet from the Server, the Server MUST discard the Application Message and not attempt to send it to any other Subscriber [MQTT-4.8.2-6].
• A Client is permitted to submit a second SUBSCRIBE request to a Shared Subscription on a Session that's already subscribed to that Shared Subscription. For example, it might do this to change the Requested QoS for its subscription or because it was uncertain that the previous subscribe completed before the previous connection was closed. This does not increase the number of times that the Session is associated with the Shared Subscription, so the Session will leave the Shared Subscription on its first UNSUBSCRIBE.
• Each Shared Subscription is independent from any other. It is possible to have two Shared Subscriptions with overlapping filters. In such cases a message that matches both Shared Subscriptions will be processed separately by both of them. If a Client has a Shared Subscription and a Non‑shared Subscription and a message matches both of them, the Client will receive a copy of the message by virtue of it having the Non‑shared Subscription. A second copy of the message will be delivered to one of the subscribers to the Shared Subscription, and this could result in a second copy being sent to this Client.

Task list decomposition

• Parse and validate Shared Subscription Topic Filter respecting the rules [MQTT-4.8.2-1] [MQTT-4.8.2-1] Validate shared subscription and closed connection if not valid #795
• Update the SUBSCRIBE message flow processing to understand when a shared subscription is involved and create a SharedSubscription object instance on the SubscriptionsDirectory. This SharedSubscription should contains the list of clients that as subscribed to the shared topic. This implementation has to satisfy the requirement:
  • the Server MUST respect the granted QoS for the Client's subscription [MQTT-4.8.2-3]
  • same shared subscription name can be used with different TopicFilters, they are effectively different subscriptions.
  • QoS2 - the Server MUST complete the delivery of the message to that Client when it reconnects [MQTT-4.8.2-4], if the Client's Session terminates before the Client reconnects, the Server MUST NOT send the Application Message to any other subscribed Client [MQTT-4.8.2-5] (this is already managed by QoS2 delivering flow).
  • If a Client responds with a PUBACK or PUBREC containing a Reason Code of 0x80 or greater to a PUBLISH packet from the Server, the Server MUST discard the Application Message and not attempt to send it to any other Subscriber [MQTT-4.8.2-6] PR: Initial implementation of shared subscriptions #796
• Update the UNSUBSCRIBE and the termination of a session so that all of its shared subscriptions are properly cleaned. A Session can explicitly detach itself from a Shared Subscription by sending an UNSUBSCRIBE Packet that contains the full Shared Subscription Topic Filter. Sessions are also detached from the Shared Subscription when they terminate. PR: Unsubscribe shared subscriptions #799
• Verify if shared subscriptions need to survive broker restarts, and if the answer is yes, provide the implementation to store and reload. The answer is YES, because the clients are removed from shared subscriptions only when they DISCONNECT explicitly and session is clean start, or UNSUBSCRIBE or the session expires, so a broker restart have to re-establish all the shared subscriptions, eventually will be removed if the clients doesn't reconnect within the session expiry period Persistence for shared subscriptions #802
• Update the CONNACK response's Shared Subscription Available property accordingly CONNACK set shared subscription available property to true #800
• Check SUBACK/DISCONNECT payload doesn't contain Shared Subscriptions not supported reason code

[andsel]

Completed as all tasks has been implemented