MqttBroker

MqttBroker connects a service to an MQTT 5 server; new only records the URL and the client id, and the runtime connects it at startup.

MqttBroker::new takes the server's URL, mqtt://host:port or mqtts://host:port, and the client id. A URL without a port takes 1883, or 8883 for mqtts://. It does no I/O, so the whole service fits the synchronous #[ruststream::app] builder. The settings that follow it are builder steps too: credentials and TLS, the session, the last will, the limits and the keep-alive. Each page gives the default a setting has when the service leaves it unset.

The client id is the session's identity on the server. A second connection under the same id takes the session over, and the server closes the first one. So two instances of one service connect under two ids, and share their work through a share group.

At startup the runtime connects the broker and receives a ConnectedMqttBroker. The crate runs one task per connection: it drives the client's event loop, hands each delivery to its subscription, and reconnects when the connection is lost. Every subscription and every publisher of the service shares that connection. At shutdown the runtime closes it with a clean DISCONNECT.

#[subscriber("devices/dev42/telemetry")] reads one topic at QoS 1. A MqttFilter reads many topics through the + and # wildcards, and the other descriptors are on the Subscriptions pages. The crate speaks MQTT 5 only: headers ride its user properties, and share groups exist only there.

The crate has two features, both off by default. testing adds the in-process mode a TestApp test connects, and asyncapi adds the MQTT bindings of the generated document. The codec is the service's choice: json, msgpack or cbor on ruststream itself.

Example

The service reads the telemetry of every device in the share group workers. A filter names no topic a retry copy could go to, so the mount site names one. cargo run --example mqtt_service --features testing -- run starts it against a server on localhost:1883.

Macros
Manual

examples/mqtt_service.rs
use ruststream_rumqttc::prelude::*;

#[subscriber(MqttFilter::new("devices/+/telemetry").shared("workers"))]
async fn record(telemetry: &Telemetry) -> HandlerOutcome {
    tracing::info!(telemetry.device, telemetry.temperature, "reading recorded");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "telemetry-svc"),
        |b| {
            // A filter matches many topics and names none of them, so the mount site says where
            // a retry copy is published.
            b.include(record)
                .out_retry(Publish::default())
                .to("devices/retry/telemetry");
        },
    )
}

examples/manual/mqtt_service.rs
use ruststream_rumqttc::prelude::*;

struct Record;

impl Handle<Telemetry> for Record {
    fn handle(
        &self,
        telemetry: &Telemetry,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        tracing::info!(telemetry.device, telemetry.temperature, "reading recorded");
        ready(Ok(()))
    }
}

fn telemetry() -> MqttFilter {
    MqttFilter::new("devices/+/telemetry").shared("workers")
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "telemetry-svc"),
        |b| {
            let record = subscriber("devices/+/telemetry", Record).map_source(|_| telemetry());
            // A filter matches many topics and names none of them, so the mount site says where
            // a retry copy is published.
            b.include(record.build())
                .out_retry(Publish::default())
                .to("devices/retry/telemetry");
        },
    )
}

Example​

Example