ruststream-rumqttc 0.7.1

ruststream-rumqttc connects a RustStream service to an MQTT 5 broker over rumqttc. Handlers, routing, codecs and middleware come from the framework; this crate is the transport.

Features

• A connection task the crate owns: one event loop per broker, per-subscription streams, reconnects with backoff and resubscribes when the session is gone.
• QoS-aware acknowledgement for QoS 1 and 2; QoS 0 reports that it has none.
• Shared subscriptions for competing consumers.
• Overlapping filters on one connection, each receiving a message once (MQTT 5 subscription identifiers); the same filter opened twice is one wire subscription.
• Topics and filters: MqttTopic for a topic, MqttFilter for the + and # wildcards.
• Retry caps and dead-letter topics declared where the handler is mounted.
• Batches assembled on the client.
• Headers as user properties, with content-type, reply-to and correlation-id as the native MQTT 5 properties.
• Byte payloads without a codec: #[derive(Serialized)] out, #[derive(Deserialized)] in.
• Sessions, wills and retained messages, QoS and retain per message (publisher.message(&state).retain(true).publish()), and TLS with client certificates.
• AsyncAPI with the specification's mqtt binding, behind the asyncapi feature.
• Tests run the production app (feature testing): TestApp::start(app()) connects MqttBroker in process, with no server; TestApp::start_live(app()) runs the same test against a real broker.

Install

[dependencies]
ruststream = { version = "0.7", features = ["macros", "json"] }
ruststream-rumqttc = "0.7"
serde = { version = "1", features = ["derive"] }

[dev-dependencies]
ruststream-rumqttc = { version = "0.7", features = ["testing"] }

Write a service

use std::time::Duration;

use ruststream_rumqttc::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, Deserialize, Outgoing, Serialize)]
struct Telemetry {
    device: String,
    temperature: f64,
}

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "alerts")]
struct Alert {
    device: String,
}

#[subscriber(MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce).shared("workers"))]
async fn handle(telemetry: &Telemetry, Out(alerts): Out<impl Publisher>) -> HandlerOutcome {
    if telemetry.temperature <= 30.0 {
        return HandlerOutcome::ack();
    }
    let alert = Alert {
        device: telemetry.device.clone(),
    };
    if alerts.message(&alert).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    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")
            .keep_alive(Duration::from_secs(30))
            .clean_start(false)
            .session_expiry(Duration::from_hours(1)),
        |b| {
            b.include(handle)
                .out(DefaultSlot, Publish::default().qos(Qos::AtLeastOnce))
                .out_retry(Publish::default())
                .to("devices/retry/telemetry")
                .build();
        },
    )
}

#[ruststream::app] generates main, so the binary understands run and asyncapi gen.

Test it

TestApp runs the service's own app with MqttBroker in process, with no server. TestApp::start_live(app()) runs the same test against a real broker.

use ruststream::testing::TestApp;

let tb = TestApp::start(app()).await?;

tb.broker::<MqttBroker>()
    .message(&Telemetry {
        device: "dev42".to_owned(),
        temperature: 31.5,
    })
    .to("devices/dev42/telemetry")
    .publish()
    .await?;

tb.broker::<MqttBroker>()
    .published::<Alert>("alerts")
    .assert_called_once()
    .with(&Alert {
        device: "dev42".to_owned(),
    });
tb.shutdown().await?;

Documentation

• Getting started - installation and a first service against a local Mosquitto.
• Guides - topics and share groups, delivery guarantees, device state, publishing commands, retries, request and response, testing and a production server.
• Components - one page per part of the crate: the broker, subscriptions, settlement, the context, publishing, AsyncAPI and testing.
• Architecture - how the crate sits between rumqttc and the runtime.
• Benchmarks - what the crate costs over the raw client.
• API reference - every item on docs.rs.
• The framework: https://powersemmi.github.io/ruststream/latest/

Minimum supported Rust version

The MSRV is 1.95, edition 2024.

Contributing

See CONTRIBUTING.md.

just check          # fmt, clippy, feature checks
just test           # in-process tests, no server
just test-brokers   # live integration + conformance against mosquitto

License

Licensed under the Apache-2.0 license.