MqttFilter

MqttFilter subscribes with the + and # wildcards; each delivery carries its own topic, and the mount site names where retry copies go.

MqttFilter::new takes the topic filter. + matches exactly one level of a topic, and # matches every level that remains, so it may only come last. A filter that starts with a wildcard does not match a topic that starts with $, such as the server's own $SYS topics. qos sets the quality of service, and shared puts the subscription in a share group. The subscription's name is the filter. A handler reads the topic each message was published to through DeliveryTopic. A filter with # before its last level, or one with a control character, stops the service at startup.

A filter is no topic a publisher can use, so the subscription has no topic of its own for a retry_after copy. The mount site names one with .out_retry(policy).to(topic), or names one per delivery with a transform. A registration that names neither does not start, and the error names the subscription and MqttFilter. A destination the filter matches brings the copy back to this subscription.

Example

The readings of every device, read at QoS 1; a retry copy goes to devices/retry/telemetry.

Macros
Manual

examples/mqtt_subscriptions_filter.rs
/// Collects the readings of every device: `+` stands for the device's own level of the topic.
#[subscriber(MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce))]
async fn collect(telemetry: &Telemetry) -> HandlerOutcome {
    tracing::info!(device = %telemetry.device, telemetry.temperature, "reading collected");
    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| {
            // A filter is no topic to publish to, so the mount site names where a retry copy
            // goes: a topic the filter matches, so the copy comes back here.
            b.include(collect)
                .out_retry(Publish::default())
                .to("devices/retry/telemetry");
        },
    )
}

examples/manual/mqtt_subscriptions_filter.rs
/// Collects the readings of every device: `+` stands for the device's own level of the topic.
struct Collect;

impl Handle<Telemetry> for Collect {
    fn handle(
        &self,
        telemetry: &Telemetry,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(device = %telemetry.device, telemetry.temperature, "reading collected");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "telemetry-svc"),
        |b| {
            let fleet = MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce);
            // A filter is no topic to publish to, so the mount site names where a retry copy
            // goes: a topic the filter matches, so the copy comes back here.
            b.include(
                subscriber("devices/+/telemetry", Collect)
                    .build()
                    .map_source(|_| fleet),
            )
            .out_retry(Publish::default())
            .to("devices/retry/telemetry");
        },
    )
}

Example​

Example