Retained

A retained message stays on its topic for every later subscriber of a matching filter; an empty retained payload clears it.

The server keeps the last retained message of each topic. A subscription that opens later receives it at once, so a dashboard that starts after a device announced its state still reads that state. A publisher retains every message with Publish::default().retain(true), and one message with the retain step. Publishing an empty payload retained removes the message the server kept for that topic. A shared subscription receives no retained messages. A scope's after_startup hook runs once the broker is connected, which is where a service announces its own status. A state is bytes on the wire rather than an encoded model, so its type carries its own bytes with #[derive(Serialized)], and no codec runs on them.

Example

Macros

examples/mqtt_retained.rs

/// A device state announcement. An MQTT state is bytes on the wire rather than an encoded
/// model, so the type carries its own bytes and no codec runs on them; the name template turns
/// the device id into a setter the call fills in.
#[derive(Outgoing, Serialized)]
#[outgoing(name = "devices/{device}/state")]
struct DeviceState(Vec<u8>);

/// The service's own status, announced once it is connected.
#[derive(Outgoing, Serialized)]
#[outgoing(name = "services/state-tracker/status")]
struct ServiceStatus(Vec<u8>);

Manual

examples/manual/mqtt_retained.rs

/// A device state announcement. An MQTT state is bytes on the wire rather than an encoded
/// model, so the type carries its own bytes and no codec runs on them; the name template turns
/// the device id into a setter the call fills in.
#[derive(Outgoing, Serialized)]
#[outgoing(name = "devices/{device}/state")]
struct DeviceState(Vec<u8>);

/// The service's own status, announced once it is connected.
#[derive(Outgoing, Serialized)]
#[outgoing(name = "services/state-tracker/status")]
struct ServiceStatus(Vec<u8>);

The service announces its status retained at startup, and each device state through its slot:

Macros

examples/mqtt_retained.rs

#[ruststream::app]
fn app() -> impl App {
    service(MqttBroker::new("mqtt://localhost:1883", "state-tracker"))
}

fn service(broker: MqttBroker) -> RustStream {
    RustStream::new(AppInfo::new("state-tracker", "0.1.0")).with_broker(broker, |b| {
        // Every publish of this policy is retained, so a dashboard that starts later still
        // reads the status.
        b.after_startup(
            Publish::default().qos(Qos::AtLeastOnce).retain(true),
            async move |publisher| -> Result<(), PublishError<MqttError>> {
                publisher
                    .message(&ServiceStatus(b"online".to_vec()))
                    .publish()
                    .await
            },
        );
        // The mount site fixes what every publish through the slot takes; the handler's step is
        // what one message changes.
        b.include(announce_state)
            .out(States, Publish::default().qos(Qos::AtLeastOnce))
            .out_retry(Publish::default())
            .to("devices/retry/telemetry")
            .build();
    })
}

Manual

examples/manual/mqtt_retained.rs

fn app() -> impl App {
    service(MqttBroker::new("mqtt://localhost:1883", "state-tracker"))
}

fn service(broker: MqttBroker) -> RustStream {
    RustStream::new(AppInfo::new("state-tracker", "0.1.0")).with_broker(broker, |b| {
        // Every publish of this policy is retained, so a dashboard that starts later still
        // reads the status.
        b.after_startup(
            Publish::default().qos(Qos::AtLeastOnce).retain(true),
            async move |publisher| -> Result<(), PublishError<MqttError>> {
                publisher
                    .message(&ServiceStatus(b"online".to_vec()))
                    .publish()
                    .await
            },
        );
        // The mount site fixes what every publish through the slot takes; the handler's step is
        // what one message changes.
        b.include(
            subscriber("devices/+/telemetry", AnnounceState)
                .map_source(|_| fleet())
                .build(),
        )
        .out(States, Publish::default().qos(Qos::AtLeastOnce))
        .out_retry(Publish::default())
        .to("devices/retry/telemetry")
        .build();
    })
}