MQTT bindings

With the asyncapi feature the generated document speaks MQTT 5: the session the client opens, the QoS of each operation and the properties of each message.

The core builds the AsyncAPI document from the app, before anything connects. This crate adds the specification's mqtt binding, version 0.2.0, to the parts it knows about. A server reports where the client connects, the protocol version 5 and the session it opens. An operation reports the quality of service it runs at. A message reports the MQTT 5 properties it rides. A reply named per request reports where its address is read.

Everything comes from the service's own declarations: the broker, the descriptors, the publish policies and the transforms. The document is published and shared, so no binding carries a credential. The binary #[ruststream::app] writes prints the document with asyncapi gen.

Example

The dependency line:

ruststream-rumqttc = { version = "0.7", features = ["asyncapi"] }

The service registers its broker under the label mqtt, the key of its server in the document. cargo run --example mqtt_asyncapi --features asyncapi -- asyncapi gen prints the document.

Macros

examples/mqtt_asyncapi.rs

fn service(broker: MqttBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("telemetry", "1.0.0")).with_broker_labeled("mqtt", broker, |b| {
        b.include(collect)
            .out(Alerts, Publish::default().retain(true))
            .out_retry(Publish::default())
            .to("devices/retry/telemetry")
            .build();
        b.include(answer)
            .out_reply(Publish::default())
            .transform(ToResponseTopic);
    })
}

#[ruststream::app]
fn app() -> impl App<State = ()> {
    let password = env::var("MQTT_PASSWORD").unwrap_or_default();
    service(broker(&password))
}

Manual

examples/manual/mqtt_asyncapi.rs

fn service(broker: MqttBroker) -> impl App<State = ()> {
    RustStream::new(AppInfo::new("telemetry", "1.0.0")).with_broker_labeled("mqtt", broker, |b| {
        b.include(
            subscriber("devices/+/telemetry", Collect)
                .build()
                .map_source(|_| fleet()),
        )
        .out(Alerts, Publish::default().retain(true))
        .out_retry(Publish::default())
        .to("devices/retry/telemetry")
        .build();
        // A request without a response topic is answered on `devices/dev42/pong`.
        let answer = subscriber("devices/dev42/ping", Answer)
            .map_source(|_| MqttTopic::new("devices/dev42/ping"));
        b.include(answer.reply().to("devices/dev42/pong").build())
            .out_reply(Publish::default())
            .transform(ToResponseTopic);
    })
}

fn app() -> impl App<State = ()> {
    let password = env::var("MQTT_PASSWORD").unwrap_or_default();
    service(broker(&password))
}