Skip to main content

Demultiplexing

One connection carries every subscription: subscription identifiers tell overlapping filters apart, and subscriptions on one filter take turns.

Two PUBLISH packets for one topic, each naming the subscription identifier of the filter it was sent for, handed to that filter's subscriptions in turnFilters on the connectionPUBLISHdevices/dev42/telemetryidentifier 17devices/+/telemetryidentifier 17takes it(its turn)takes thenext onePUBLISHdevices/dev42/telemetryidentifier 42devices/#identifier 42one subscriptiontakes every oneA packet no open subscription claims waits, up to 1024,for the subscription that opens later
The diagram as text
  1. Filters on the connection: devices/+/telemetry identifier 17; takes it (its turn); takes the next one; devices/# identifier 42; one subscription takes every one
  2. PUBLISH devices/dev42/telemetry identifier 17 → devices/+/telemetry identifier 17
  3. devices/+/telemetry identifier 17 → takes it (its turn)
  4. devices/+/telemetry identifier 17 → takes the next one: next
  5. PUBLISH devices/dev42/telemetry identifier 42 → devices/# identifier 42
  6. devices/# identifier 42 → one subscription takes every one
  7. A packet no open subscription claims waits, up to 1024, for the subscription that opens later

The server subscribes a filter once per connection. Two handlers on the same filter, or the two members of a share group in one service, are one subscription on the server and two subscriptions in the service. The crate hands the filter's deliveries to them in turn. The filter stays subscribed while any of them is open, and a shutdown leaves it in the session. It is subscribed at the highest quality of service any of them asks for.

Filters that differ but can match one topic, devices/+/telemetry next to devices/# or a share group next to a plain subscription on its filter, are separate subscriptions on the server. For a message that matches both, Mosquitto sends a packet for each. MQTT 5 lets a subscription carry an identifier, and the server names it in every packet it sends for that subscription. The crate gives an identifier only to a filter that overlaps another one on the connection, and routes each packet by the identifier it names.

A filter that meets another one only after it was subscribed is subscribed again, with an identifier, and asks the server for no retained messages that time, so a retained message is not delivered twice. The identifier is derived from the filter itself. A restarted service that resumes a persistent session therefore gives each filter the identifier the server still holds.

Subscription identifiers are optional for a server, and its CONNACK says whether it offers them. On a server that does not, opening a filter that overlaps another one on the same connection fails with an error naming both, before anything is subscribed. Two brokers with client ids of their own keep such filters apart. When a server sends one packet for two filters, the delivery to the first carries the acknowledgement, and the other's ack reports AckError::Unsupported.

A packet no open subscription claims is held, because a resumed session sends its backlog before the service has opened the subscriptions it belongs to. Up to 1024 packets wait, and the oldest is dropped with a warning past that.

Lanes​

One subscription's dispatch: one stream of deliveries spread over two lanes in turnThe subscriptiona topic or a filter, on the shared connectionMqttSubscriber streamone delivery at a timeworkers(2)the lanes take the deliveries in turnLane 1runs its deliveries in orderLane 2runs its deliveries in orderby_key: one partition-key value, one lane;without the header, the next lane in turnAcross processes: a share grouphands each message to one member
The diagram as text
  1. The subscription a topic or a filter, on the shared connection → MqttSubscriber stream one delivery at a time: delivered
  2. MqttSubscriber stream one delivery at a time → workers(2) the lanes take the deliveries in turn: one at a time
  3. workers(2) the lanes take the deliveries in turn → Lane 1 runs its deliveries in order
  4. workers(2) the lanes take the deliveries in turn → Lane 2 runs its deliveries in order
  5. by_key: one partition-key value, one lane; without the header, the next lane in turn
  6. Across processes: a share group hands each message to one member

Without workers, one loop runs the handler for each delivery of a subscription in turn. The order in which the server sent them is kept, and nothing runs side by side.

workers(n) runs n lanes, and a lane runs its deliveries in order. The lanes take the deliveries in turn, so deliveries of one topic run side by side on different lanes. Under workers(n, by_key) the partition-key header the publisher set picks the lane: the deliveries of one key run in order, and a delivery without the header takes the next lane in turn.

Across processes, a share group divides the messages: the server hands each one to one member. The Receive Maximum bounds how many QoS 1 and QoS 2 deliveries a connection holds unacknowledged at once, across all its subscriptions.

A batch handler takes up to the size the mount site names. A PUBLISH carries one message, so the crate assembles the batch on the client. A batch closes when it holds that size or 20 milliseconds after its first delivery.