Session

clean_start(false) with a session_expiry keeps the session on the server, so a service that loses its connection gets what arrived meanwhile.

An MQTT session belongs to a client id and lives on the server. It holds the client's subscriptions and the QoS 1 and 2 messages the client has not acknowledged yet. By default every connect starts a clean session, and the server forgets it when the connection ends.

clean_start(false) asks the server to resume the session the client id already has. session_expiry says how long the server keeps it after the connection ends. The duration counts in whole seconds, and anything longer than u32::MAX seconds keeps the session with no end. Without an expiry the server drops the session with the connection, whatever clean_start says.

While the session lives, the server queues every QoS 1 and 2 message that matches its subscriptions. When the service connects again under the same id, the server sends what it queued, then what arrives next. This covers a lost connection, a reconnect, a process that dies and a clean shutdown. A clean shutdown leaves the app's subscriptions in the session, so a service that restarts receives what was published while it was stopped.

A QoS 0 message is never queued. A delivery the handler left unacknowledged, such as one HandlerOutcome::retry() settled, comes back when the session resumes.

The session lives on the server, so a TestApp test in process starts a fresh one every time. What a session keeps is for a live test.

Example

The server keeps the service's session for a day. A reading published while the service is down reaches it when it comes back.

Macros
Manual

examples/mqtt_broker_session.rs
/// The server at `url` keeps this client's session for a day after the service disconnects.
fn broker(url: &str, client_id: &str) -> MqttBroker {
    MqttBroker::new(url, client_id)
        .clean_start(false)
        .session_expiry(Duration::from_hours(24))
}

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

examples/manual/mqtt_broker_session.rs
/// The server at `url` keeps this client's session for a day after the service disconnects.
fn broker(url: &str, client_id: &str) -> MqttBroker {
    MqttBroker::new(url, client_id)
        .clean_start(false)
        .session_expiry(Duration::from_hours(24))
}

fn app() -> impl App<State = ()> {
    service(broker("mqtt://localhost:1883", "telemetry-svc"))
}

Example​

Example