Shutdown

Shutdown sends a clean DISCONNECT and waits up to 5 s for what is queued to leave; a publisher used afterwards reports MqttError::NotConnected.

The runtime stops reading first and lets the handlers in flight finish. How long they get is the app's shutdown_timeout, with no bound by default. Their publishes and acknowledgements join the client's queue as they settle. Each subscription leaves its filter on the server as it closes, so a persistent session keeps the app's subscriptions, and the server queues what arrives until the service returns.

Then the broker sends a DISCONNECT behind everything already queued. It waits up to 5 seconds for the connection task to write the queue out, and logs a warning when they run out. The server discards the last will of a connection that ends this way.

A publisher taken before the shutdown keeps pointing at the closed connection. A publish through it returns MqttError::NotConnected and never succeeds against the dead connection. An acknowledgement after the shutdown is refused too.

Example

The handlers get ten seconds to finish, and their alerts leave before the broker disconnects.

Macros
Manual

examples/mqtt_broker_shutdown.rs
/// At shutdown the handlers get ten seconds to finish; their alerts and acknowledgements leave
/// before the broker disconnects.
fn service(broker: MqttBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("telemetry", "0.1.0"))
        .shutdown_timeout(Duration::from_secs(10))
        .with_broker(broker, |b| {
            b.include(check)
                .out(DefaultSlot, Publish::default())
                .build();
        })
}

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

examples/manual/mqtt_broker_shutdown.rs
/// At shutdown the handlers get ten seconds to finish; their alerts and acknowledgements leave
/// before the broker disconnects.
fn service(broker: MqttBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("telemetry", "0.1.0"))
        .shutdown_timeout(Duration::from_secs(10))
        .with_broker(broker, |b| {
            b.include(subscriber("broker-shutdown/telemetry", Check).build())
                .out(DefaultSlot, Publish::default())
                .build();
        })
}

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

Example​

Example