Skip to main content

Topics, wildcards and shared subscriptions

This guide reads MQTT topics one at a time and many at once, shares the work between replicas of a service and mounts filters that overlap.

A topic is a name of levels separated by /, such as devices/dev42/telemetry. The server hands a message to every subscription whose filter matches its topic at that moment. It keeps nothing for later, except one retained message per topic (Device state) and the queue of a persistent session (Delivery guarantees).

The example is a fleet service. It stores the telemetry of every device, traces every message of one device and schedules firmware rollouts.

A share group​

store reads the telemetry of every device. The service runs in several replicas, and each reading must be stored once.

src/main.rs
/// Every replica of the service joins the share group `ingest`, so each reading is stored once.
#[subscriber(MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce).shared("ingest"))]
async fn store(telemetry: &Telemetry, Ctx(topic): Ctx<DeliveryTopic>) -> HandlerOutcome {
    tracing::info!(%topic, telemetry.temperature, "reading stored");
    HandlerOutcome::ack()
}

MqttFilter reads every topic a filter matches. + matches exactly one level, so devices/+/telemetry reads the telemetry of each device and nothing deeper. shared("ingest") subscribes $share/ingest/devices/+/telemetry, and the server hands each message to one session of the group. Each replica connects with its own client id and joins the same group, so the replicas split the readings between them. A subscription outside the group, or in another group, still gets its own copy.

The handler never sees the $share/.. prefix, and it never sees the filter either. Ctx<DeliveryTopic> is the topic this reading arrived on, such as devices/dev42/telemetry, so the device's id is in reach.

Order per device​

The server hands each reading of a share group to any member, so the readings of one device may land on different replicas. Order by device holds inside one subscription: workers(n, by_key) runs the readings that share a key one after another, and different keys side by side. The key is the partition-key header, PARTITION_KEY_HEADER, which the device sets on each reading, such as its id. A subscription whose readings must stay in order per device reads outside a share group.

A multi-level wildcard​

trace logs every message of one device, whatever its kind.

src/main.rs
/// A payload taken as the bytes it is, because one device sends messages of several kinds.
#[derive(Deserialized)]
struct Raw<'a>(&'a [u8]);

/// Logs every message of the device under investigation, whatever its kind.
#[subscriber(MqttFilter::new("devices/dev42/#"))]
async fn trace(message: &Raw<'_>, Ctx(topic): Ctx<DeliveryTopic>) -> HandlerOutcome {
    tracing::debug!(%topic, bytes = message.0.len(), "dev42 traced");
    HandlerOutcome::ack()
}

# matches the rest of the topic, any number of levels, and it stands last in a filter. devices/dev42/# reads devices/dev42/telemetry, devices/dev42/telemetry/raw and every other topic of the device. The device sends messages of several kinds, so trace takes the payload as the bytes it is, and no codec runs on it.

A topic that starts with $, such as the server's own $SYS/.. statistics, is matched only by a filter that starts with $ too. So # alone never reads them.

One topic​

src/main.rs
#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "fleet/firmware")]
struct FirmwareRelease {
    version: String,
}

/// A bare string is one topic, read at the default quality of service.
#[subscriber("fleet/firmware")]
async fn schedule_rollout(release: &FirmwareRelease) -> HandlerOutcome {
    tracing::info!(release.version, "rollout scheduled");
    HandlerOutcome::ack()
}

A bare string in #[subscriber(..)] is one topic, read with the defaults of MqttTopic. MqttTopic::new("fleet/firmware").qos(..) writes the same subscription with its settings. MqttTopic takes no wildcard: one in its name fails the start with an error that names MqttFilter, before any I/O. Wildcards belong to subscriptions only, so a publish names one topic, and a publish to a topic with + or # returns an error.

The app​

src/main.rs
#[ruststream::app]
fn app() -> impl App<State = ()> {
    RustStream::new(AppInfo::new("fleet", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "fleet-svc"),
        |b| {
            // A filter names no topic of its own, so each one names where a retry copy goes:
            // a topic it matches and no other filter of the service does.
            b.include(store)
                .out_retry(Publish::default())
                .to("devices/retry/telemetry");
            b.include(trace)
                .out_retry(Publish::default())
                .to("devices/dev42/retry");
            b.include(schedule_rollout);
        },
    )
}

A filter matches many topics, so a delayed copy of its delivery has no single topic to go to. The mount site names one with out_retry(Publish::default()).to(..), and the service refuses to start without it. A copy reaches every subscription whose filter matches its topic, so each topic here is matched by its own filter and by no other. Retries and dead letters explains the copies.

Overlapping filters​

store and trace both match devices/dev42/telemetry, and they share one connection. Each of them runs once for a reading on that topic. The server sends a copy per matching subscription, and the broker tells the copies apart by the subscription identifier each one carries. Subscription identifiers are optional for a server. On a server that does not offer them, the service fails to start with an error that names both filters.

The same filter mounted twice in one service is one subscription for the server. Its handlers take turns on the deliveries instead of each getting a copy.

Testing the routing​

The tests in the example file run the app in process. They publish to devices/dev42/telemetry, devices/dev7/telemetry and devices/dev42/telemetry/raw, and check which handlers ran. The in-process server matches topics the way Mosquitto does, wildcards and overlaps included. Several replicas sharing a group need several connections to one server, so the split itself shows only against a real broker.