Skip to main content

Crate ruststream_rumqttc

Crate ruststream_rumqttc 

Source
Expand description

MQTT 5 broker for the RustStream messaging framework, backed by rumqttc: topic filters with wildcards, shared subscriptions, retained publishes, and headers that ride user properties.

A subscription is a topic or a topic filter the server delivers matching messages to, and a publish is one PUBLISH packet with a quality of service and a retain flag. The crate owns one task that drives the client’s event loop, hands each packet to the subscriptions whose filters match, and reconnects when the connection drops.

§A service

A handler is an async fn over the decoded payload, mounted on a broker inside the application object, and the attribute writes the main:

use std::time::Duration;

use ruststream_rumqttc::prelude::*;
use serde::Deserialize;

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

#[subscriber(MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce).shared("workers"))]
async fn handle(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")
            .keep_alive(Duration::from_secs(30))
            .clean_start(false)
            .session_expiry(Duration::from_hours(1)),
        |b| {
            // A filter names no topic a publisher can use, so the mount site says where a
            // deferred retry copy goes.
            b.include(handle)
                .out_retry(Publish::default())
                .to("devices/retry/telemetry");
        },
    )
}

cargo run -- run starts it. MqttBroker::new only records the URL and the client id; the runtime connects the broker at startup and disconnects it when the service stops.

§Where things are

The guides take an MQTT service from installation to production, in this order: getting started, topics, wildcards and shared subscriptions, delivery guarantees, device state, publishing commands, retries and dead letters, request and response, testing and connecting to production. The architecture pages show how the crate sits between rumqttc and the runtime, and the benchmarks what it costs.

Each topic below links to its component page on the site, which shows it in a service; the item’s own documentation keeps the detail.

§Broker

§Subscriptions

§Settlement

  • Acknowledgement: what an ack and a drop send on an MqttMessage at each quality of service.
  • retry: a delivery left unacknowledged until a persistent session resumes.
  • retry_after: a copy the runtime publishes after a delay, within the session.
  • Retries: max_attempts, counted on the framework’s retry-count header.
  • Dead letters: dead_letter, the topic a spent delivery goes to.
  • out_retry: where the retry copies of an MqttFilter registration go.

§Context

§Publishing

§Tooling

§Cargo features

Both are off by default and additive. testing adds the in-process mode TestApp::start connects the broker through, and asyncapi contributes the mqtt bindings to the generated document. The codecs, the macros and the harness itself are features of the core crate.

Modules§

prelude
The imports a routes file on MQTT writes every time, in one glob.

Structs§

ConnectedMqttBroker
The typed witness that connect succeeded: holds the live client directly, or the in-process transport when the test harness connected it.
DeliveryTopic
The key for MqttContext::topic.
MqttBroker
An MQTT 5 broker for the RustStream messaging framework.
MqttContext
The per-delivery context of this broker.
MqttFilter
A subscription descriptor for one MQTT topic filter, wildcards included.
MqttMessage
A message delivered by an MqttSubscriber.
MqttPublish
The publish policy for MqttPublisher: quality of service and the retain flag as pure declaration, paired with the connected broker by the runtime after connect.
MqttPublishOptions
The two arguments MQTT carries on every PUBLISH packet, as one message asked for them.
MqttPublisher
Publishes messages to MQTT topics through the shared connection.
MqttSubscriber
A subscription to one MQTT topic filter; yields MqttMessages one at a time, or in batches.
MqttTopic
A subscription descriptor for one MQTT topic.

Enums§

MqttError
Errors returned by the MQTT 5 broker.
Qos
Delivery quality of service for a subscription or a publish policy.

Constants§

PARTITION_KEY_HEADER
The header that carries a message’s partition key, which keyed worker lanes (workers(n, by_key)) order by.

Traits§

MqttPublishSteps
The per-message steps of an MQTT publish, on the publish builder itself.