Skip to main content

Device state: retained messages and the last will

This guide keeps the current state of every device on the server, so a reader that starts later sees it at once, and announces whether the service itself is running.

MQTT keeps no history, with one exception: a retained message. The server keeps the last retained message of each topic and sends it to every new subscription whose filter matches. The last will is a message the server publishes for a client whose connection ends without a clean disconnect.

The example is a gateway monitor. Gateways send heartbeats with their battery level. The service keeps the state of each gateway, ok or low-battery, retained on gateways/<id>/state, and dashboards read it from there.

The state of each gateway​

src/main.rs
/// The state of one gateway, as plain text a dashboard shows as it is. The type carries its own
/// bytes, so no codec runs on them, and the `{gateway}` level becomes a setter.
#[derive(Outgoing, Serialized)]
#[outgoing(name = "gateways/{gateway}/state")]
struct GatewayState(Vec<u8>);

#[derive(OutSlot)]
#[publishes(GatewayState)]
struct States;

/// Turns each heartbeat into the gateway's state.
#[subscriber(MqttFilter::new("gateways/+/heartbeat").qos(Qos::AtLeastOnce))]
async fn track(heartbeat: &Heartbeat, Out(states): Out<impl Publisher, States>) -> HandlerOutcome {
    let state = if heartbeat.battery_percent < 20 {
        "low-battery"
    } else {
        "ok"
    };
    let announced = states
        .message(&GatewayState(state.as_bytes().to_vec()))
        .to()
        .gateway(&heartbeat.gateway)
        .publish()
        .await;
    if let Err(err) = announced {
        tracing::warn!(heartbeat.gateway, %err, "state not announced");
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

/// An empty retained payload deletes the retained message, so a removed gateway leaves no state.
#[subscriber("fleet/decommissioned")]
async fn forget(gone: &Decommissioned, Out(states): Out<impl Publisher, States>) -> HandlerOutcome {
    let cleared = states
        .message(&GatewayState(Vec::new()))
        .to()
        .gateway(&gone.gateway)
        .publish()
        .await;
    if let Err(err) = cleared {
        tracing::warn!(gone.gateway, %err, "state not cleared");
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

A state is plain text a dashboard shows as it is, so GatewayState derives Serialized and carries its own bytes, and no codec runs on them. Its topic is a name template: {gateway} becomes a setter, so .to().gateway(..) addresses one gateway's topic. track publishes the state of the gateway each heartbeat comes from.

forget publishes an empty payload to the state topic of a gateway that is taken out of service. An empty retained message deletes the retained message of its topic, so the server stops handing that state to new subscribers.

Retained publishes​

src/main.rs
fn service(broker: MqttBroker) -> RustStream {
    RustStream::new(AppInfo::new("gateway-monitor", "0.1.0")).with_broker(broker, |b| {
        // Runs once the broker is connected, and replaces the will's `offline` of an earlier run.
        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
            },
        );
        // Every state goes out retained, so a dashboard that starts later reads it at once.
        let retained = Publish::default().qos(Qos::AtLeastOnce).retain(true);
        b.include(track)
            .out(States, retained)
            .out_retry(Publish::default())
            .to("gateways/retry/heartbeat")
            .build();
        b.include(forget).out(States, retained).build();
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    service(broker("mqtt://localhost:1883", "gateway-monitor"))
}

retain(true) on the publish policy makes every publish through the slot retained. Both handlers share the slot States and the policy, so a state and its deletion travel the same way. One message can also be retained on its own: the retain step on a publish overrides the policy for that call.

A shared subscription does not receive retained messages when it subscribes. A dashboard that reads the states subscribes outside any share group.

Online and offline​

src/main.rs
/// The service's own status, which dashboards read to tell whether it is running.
#[derive(Outgoing, Serialized)]
#[outgoing(name = "services/gateway-monitor/status")]
struct ServiceStatus(Vec<u8>);

/// If the connection ends without a clean disconnect, the server publishes `offline` on the
/// status topic, retained.
fn broker(url: &str, client_id: &str) -> MqttBroker {
    MqttBroker::new(url, client_id).last_will(
        "services/gateway-monitor/status",
        "offline",
        Qos::AtLeastOnce,
        true,
    )
}

last_will takes the topic, the payload, the quality of service and the retain flag of the will. The broker hands the will to the server when it connects. If the connection ends without a DISCONNECT packet, because the process crashed or the network failed, the server publishes the will: here offline, retained on services/gateway-monitor/status. A clean shutdown sends DISCONNECT, and the server discards the will.

The other half is the after_startup hook in the app above. It runs once the broker is connected and publishes online, retained, to the same topic. So the status topic always holds the latest of the two, and a dashboard that subscribes reads it at once.

Testing the state​

The in-process tests check what the service publishes: the state of each gateway, the empty payload of a removed one, and online at startup. The in-process server keeps retained messages as Mosquitto does. The will is published by a server for a connection that ends abruptly, so one test runs against Mosquitto:

MQTT_TEST_URL=mqtt://localhost:1883 cargo test -- --test-threads=1

It runs the service, stops it cleanly, and starts a dashboard afterwards: the dashboard receives online and the gateway's state, both retained. Then it ends a run of the service without a DISCONNECT, and the dashboard receives offline. At the end it removes the test's gateway, which clears its state on the server.