Startup

Startup waits up to 30 seconds for the server's CONNACK; a refusal the connection cannot outgrow fails it at once, naming the cause.

The runtime connects the broker first. The settings are checked before anything is dialed: a keep-alive under five seconds, an mqtts:// URL without tls_ca or a port that is not a number fail startup with MqttError::Invalid. Then the connection task dials the server. A server that is down or busy is dialed again with a backoff, until it answers or the 30 seconds run out. A timeout fails startup with MqttError::Connect, and the error carries the last reason a dial failed.

Some refusals give the same answer on every attempt: wrong credentials, a client id the server rejects, a peer that is not an MQTT server, and a TLS handshake the certificates cannot satisfy. Each fails startup at once with MqttError::Connect, naming the refusal. A server that refuses a client certificate after the handshake answers the way a broken connection does, so that refusal is retried and reported when the 30 seconds run out.

Once connected, the runtime opens every subscription, each one waiting for the server's SUBACK. A filter the server rejects fails startup with MqttError::Subscribe. The after_startup hooks run last, so a hook publishes through a connection whose subscriptions are already open.

A TestApp test in process checks the settings as the connect does, then opens the subscriptions and runs the hooks. The dial and the server's refusals are for a live test.

Example

Once its subscriptions are open, the service announces that it is online on its status topic. The announcement is retained, so a dashboard that subscribes later reads it.

Macros
Manual

examples/mqtt_broker_startup.rs
/// Once the connection is up and every subscription is open, the service announces itself on its
/// status topic, retained for whoever subscribes later.
fn service(broker: MqttBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(broker, |b| {
        b.include(record);
        b.after_startup(
            Publish::default().retain(true),
            async move |publisher| -> Result<(), PublishError<MqttError>> {
                publisher.message(&Status { online: true }).publish().await
            },
        );
    })
}

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

examples/manual/mqtt_broker_startup.rs
/// Once the connection is up and every subscription is open, the service announces itself on its
/// status topic, retained for whoever subscribes later.
fn service(broker: MqttBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(broker, |b| {
        b.include(subscriber("broker-startup/telemetry", Record).build());
        b.after_startup(
            Publish::default().retain(true),
            async move |publisher| -> Result<(), PublishError<MqttError>> {
                publisher.message(&Status { online: true }).publish().await
            },
        );
    })
}

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

Example​

Example