Reconnect

The connection task reconnects with a backoff from 100 ms to 5 s and resubscribes; keep_alive sets how soon a dead connection is noticed.

A connection that fails after startup is dialed again after 100 ms. Each failed attempt doubles the wait, up to 5 seconds, and a successful connect resets it. A service notices nothing but the gap: its handlers keep running, and its subscriptions stay open. When the server reports that it no longer has the session, the task subscribes every filter again. With a persistent session the server still has it, together with what arrived during the gap.

A refusal the connection will not outgrow ends the task instead: bad credentials, an unacceptable client id, a TLS handshake the certificates cannot satisfy. Every subscription then returns that error, as MqttError::Receive.

keep_alive is how long the connection may stay silent. When nothing else is sent for that long, the client pings the server. The server closes a connection it has heard nothing from for one and a half keep-alive intervals, and publishes its last will. The client notices a server that stopped answering the same way, and reconnects. The default is 60 seconds, and the client's minimum is 5: a shorter one fails startup, in a TestApp test as on a server.

Example

A silent connection is noticed within about 15 seconds. The session outlives the gap by an hour, so what arrives while the service reconnects is kept for it.

Macros
Manual

examples/mqtt_broker_reconnect.rs
/// A connection silent for 15 seconds is probed, and one that stays silent is dropped and dialed
/// again. The session outlives the gap by an hour, so what arrives meanwhile is kept.
fn broker(url: &str, client_id: &str) -> MqttBroker {
    MqttBroker::new(url, client_id)
        .keep_alive(Duration::from_secs(15))
        .clean_start(false)
        .session_expiry(Duration::from_hours(1))
}

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

examples/manual/mqtt_broker_reconnect.rs
/// A connection silent for 15 seconds is probed, and one that stays silent is dropped and dialed
/// again. The session outlives the gap by an hour, so what arrives meanwhile is kept.
fn broker(url: &str, client_id: &str) -> MqttBroker {
    MqttBroker::new(url, client_id)
        .keep_alive(Duration::from_secs(15))
        .clean_start(false)
        .session_expiry(Duration::from_hours(1))
}

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

Example​

Example