Subscriptions

A topic in #[subscriber] subscribes to that one topic at QoS 1; a descriptor carries wildcards, the quality of service and a share group.

#[subscriber("devices/dev42/telemetry")] subscribes to the topic devices/dev42/telemetry. The bare name is an MqttTopic with its defaults: Qos::AtLeastOnce and no share group. A name with a + or # wildcard is a filter, not a topic, so the service does not start, and the error names MqttFilter, the descriptor that takes it. Every other setting belongs to a descriptor: MqttTopic for one topic, MqttFilter for a filter with wildcards. Each descriptor carries its quality of service and its share group. A descriptor goes where the name goes, as the attribute's first argument. The manual form passes the name to subscriber(name, body), and a descriptor joins the chain through map_source. The subscription is open once the server has acknowledged it, before the service takes its first message.

Example

The readings of the device dev42, on one topic.

Macros

examples/mqtt_subscriptions_name.rs

/// Subscribes to the one topic dev42 publishes its readings to.
#[subscriber("devices/dev42/telemetry")]
async fn record(telemetry: &Telemetry) -> HandlerOutcome {
    tracing::info!(device = %telemetry.device, telemetry.temperature, "reading recorded");
    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| {
            b.include(record);
        },
    )
}

Manual

examples/manual/mqtt_subscriptions_name.rs

/// Subscribes to the one topic dev42 publishes its readings to.
struct Record;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "telemetry-svc"),
        |b| {
            b.include(subscriber("devices/dev42/telemetry", Record).build());
        },
    )
}