Getting started

This guide installs ruststream-rumqttc, starts a local MQTT server and runs a first service against it.

Installation

Cargo.toml

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

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

ruststream is the framework: macros gives the attributes, json the codec, and logging prints the service's log to the terminal. ruststream-rumqttc is the MQTT 5 broker. The broker enables no codec of its own, so the service picks json, msgpack or cbor on ruststream.

The broker crate has two features, both off by default:

Feature	What it gives
testing	the in-process MQTT server behind TestApp
asyncapi	the MQTT bindings in the generated AsyncAPI document

System requirements

The rumqttc client speaks TLS through rustls, whose cryptography comes from aws-lc-rs. aws-lc-rs compiles C code, so the build needs a C compiler. On Debian and Ubuntu, build-essential provides it. Other platforms are listed in the aws-lc-rs requirements.

A local MQTT server

One Eclipse Mosquitto container is enough to develop against. It reads its listeners from a configuration file:

mosquitto.conf

listener 1883
allow_anonymous true

compose.yaml

services:
  mosquitto:
    image: eclipse-mosquitto:2
    ports:
      - "1883:1883"
    volumes:
      - ./mosquitto.conf:/mosquitto/config/mosquitto.conf:ro

docker compose up -d

The server listens on localhost:1883 and accepts clients without a password.

The first service

Macros

src/main.rs

/// The coldest temperature a sensor can report; anything below it is a broken sensor.
const ABSOLUTE_ZERO: f64 = -273.15;

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

#[subscriber(MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce))]
async fn record(telemetry: &Telemetry) -> HandlerOutcome {
    if telemetry.temperature < ABSOLUTE_ZERO {
        tracing::warn!(telemetry.device, "impossible reading dropped");
        return HandlerOutcome::drop();
    }
    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| {
            b.include(record)
                .out_retry(Publish::default())
                .to("devices/retry/telemetry");
        },
    )
}

Manual

src/main.rs

/// The coldest temperature a sensor can report; anything below it is a broken sensor.
const ABSOLUTE_ZERO: f64 = -273.15;

#[derive(Debug, PartialEq, Deserialize, JsonSchema)]
#[schemars(crate = "ruststream::schemars")]
struct Telemetry {
    device: String,
    temperature: f64,
}

struct Record;

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

fn app() -> impl App {
    RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "telemetry-svc"),
        |b| {
            let fleet = MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce);
            b.include(
                subscriber("devices/+/telemetry", Record)
                    .map_source(|_| fleet)
                    .build(),
            )
            .out_retry(Publish::default())
            .to("devices/retry/telemetry");
        },
    )
}

fn main() -> ExitCode {
    run_main(app)
}

record reads the JSON telemetry of every device. It acknowledges a reading and drops one below absolute zero. MqttFilter subscribes to a topic filter: + stands for one level of the topic, here the device's id. Qos::AtLeastOnce makes each reading an acknowledged delivery: the acknowledgement leaves when record returns. A filter matches many topics, so the mount site names the topic a delayed retry goes to; every subscription over a filter names one, and the service refuses to start without it.

MqttBroker::new takes the server's URL and the client id, and only records them. The connection opens when the service starts. The client id names the service's session on the server, so each running process has its own. #[ruststream::app] turns app into main; in the manual form, run_main does the same.

Run the service:

cargo run -- run

Publish a reading from another terminal with the client that ships in the image:

docker compose exec mosquitto mosquitto_pub -V mqttv5 -q 1 \
  -t devices/dev42/telemetry -m '{"device":"dev42","temperature":21.5}'

The service logs reading recorded with the device and its temperature. Ctrl+C stops it.

The first test

The example file also holds two tests. Each starts the same app with TestApp::start, publishes a reading to devices/dev42/telemetry and checks how record settled it. MqttBroker connects to an in-process MQTT server there, so the tests need no Mosquitto:

cargo test

Testing an MQTT service walks through the tests and what the in-process server models.

Coming from rumqttc

A service on rumqttc drives the event loop itself and acknowledges each message by hand. Here is where each part goes in RustStream:

rumqttc	RustStream
MqttOptions and its setters	the MqttBroker builder: keep_alive, credentials, clean_start, last_will
an EventLoop::poll loop	the broker's own connection task, which reconnects and resubscribes
AsyncClient::subscribe	a #[subscriber] handler on an MqttTopic or an MqttFilter
deserializing publish.payload	the handler's parameter type, decoded by the codec
set_manual_acks(true) and AsyncClient::ack	the handler's outcome (Delivery guarantees)
AsyncClient::publish	a reply or an Out parameter, sent through a publish policy (Publishing commands)