Overlaps

Overlapping filters on one connection each receive a message once, told apart by subscription identifiers; one filter mounted twice shares it.

Two filters overlap when one topic can match both, such as devices/+/telemetry and devices/dev42/#. They are two subscriptions to the server, and each receives a message once. The server sends a packet per matching subscription, and the crate tells the packets apart by MQTT 5 subscription identifiers. The crate attaches an identifier only to a filter that meets another one on the same connection. A share group next to a plain subscription on the same filter overlaps the same way.

The same filter mounted twice, or two members of one share group, is one subscription to the server. The crate hands its deliveries to the mounts in turn, and the filter stays subscribed while any of them is open. A shutdown leaves it in the session.

A server announces whether it offers subscription identifiers when the connection opens. On a server without them, a filter that meets another one on the same connection does not open, and the error names both filters. Two brokers with client ids of their own keep such filters on separate connections.

Example

The fleet's readings and everything dev42 sends: a reading of dev42 reaches both handlers, once each. Each retry topic matches its own filter only, so a retry copy returns to the subscription that asked for it.

Macros
Manual

examples/mqtt_subscriptions_overlaps.rs
/// Records the readings of the whole fleet.
#[subscriber(MqttFilter::new("devices/+/telemetry"))]
async fn record(telemetry: &Telemetry) -> HandlerOutcome {
    tracing::info!(device = %telemetry.device, telemetry.temperature, "reading recorded");
    HandlerOutcome::ack()
}

/// Watches everything dev42 sends, its readings included.
#[subscriber(MqttFilter::new("devices/dev42/#"))]
async fn watch(telemetry: &Telemetry) -> HandlerOutcome {
    tracing::info!(device = %telemetry.device, "dev42 watched");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "telemetry-svc"),
        |b| {
            // Each retry topic matches its own filter and not the other one, so a copy comes
            // back to the subscription that asked for it alone.
            b.include(record)
                .out_retry(Publish::default())
                .to("devices/retry/telemetry");
            b.include(watch)
                .out_retry(Publish::default())
                .to("devices/dev42/retry");
        },
    )
}

examples/manual/mqtt_subscriptions_overlaps.rs
fn app() -> impl App {
    RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "telemetry-svc"),
        |b| {
            // Each retry topic matches its own filter and not the other one, so a copy comes
            // back to the subscription that asked for it alone.
            b.include(
                subscriber("devices/+/telemetry", Record)
                    .build()
                    .map_source(|_| MqttFilter::new("devices/+/telemetry")),
            )
            .out_retry(Publish::default())
            .to("devices/retry/telemetry");
            b.include(
                subscriber("devices/dev42/#", Watch)
                    .build()
                    .map_source(|_| MqttFilter::new("devices/dev42/#")),
            )
            .out_retry(Publish::default())
            .to("devices/dev42/retry");
        },
    )
}

Example​

Example