stem-mqtt

An MQTT 3.1.1 / 5.0 client and broker written in Rust — async (tokio), TLS, WebSocket, QoS 0/1/2, shared subscriptions, enhanced authentication — usable as a Rust library, as a standalone broker binary, and from Kotlin, Swift, Python, Go, C#, Java, Dart, Node.js and Haskell through generated UniFFI bindings. Client and broker ship in every language, one generator per language, each runtime-tested against a real broker in CI.

Crate	Version	Downloads	Docs
stem-mqtt-client
stem-mqtt-broker

• Quick start · Installation · Features · Running the broker · Other languages

Quick start (Rust)

cargo add stem-mqtt-client stem-mqtt-broker anyhow
cargo add tokio --features full

An embedded broker and a client talking to each other:

use mqtt_broker::{MqttBroker, MqttBrokerConfig};
use mqtt_client::{ConnectOptions, MqttClient, MqttVersion, QoS};

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Broker on a free local port.
    let broker = MqttBroker::new(MqttBrokerConfig::new("127.0.0.1", 0));
    broker.start().await?;
    let port = broker.bound_port().expect("bound");

    // Client (MQTT 5; use MqttVersion::V311 for 3.1.1).
    let client = MqttClient::new(ConnectOptions::new("127.0.0.1", port, "my-client-id", MqttVersion::V5));
    client.connect().await?;
    client.subscribe("stem/demo".into(), QoS::AtLeastOnce).await?;
    client.publish("stem/demo".into(), b"hello".to_vec(), QoS::AtLeastOnce, false).await?;

    client.disconnect().await?;
    broker.stop().await?;
    Ok(())
}

Just want a broker? cargo install stem-mqtt-broker && mqtt-broker --port 1883 (see Running the broker).

Runnable examples (the mqtt-client ones need a broker already running):

cargo run -p stem-mqtt-broker --example simple_broker
cargo run -p stem-mqtt-client --example pub_sub
cargo run -p stem-mqtt-client --example will_and_retain
cargo run -p stem-mqtt-client --example mqtt_versions          # MQTT 3.1.1 + 5.0 on the same broker
cargo run -p stem-mqtt-client --example topics                 # topic hierarchies, `+` / `#` wildcards
cargo run -p stem-mqtt-client --example long_lived_connection  # persistent connection, keep-alive
cargo run -p stem-mqtt-client --example shared_subscriptions   # $share/<group>/… round-robin work queue
cargo run -p stem-mqtt-broker --example auth_broker            # then, in another terminal:
cargo run -p stem-mqtt-client --example auth_client            # one login rejected, one accepted

Full crate docs: crates/mqtt-client, crates/mqtt-broker.

Preview

Publish / subscribe

The mqtt-broker CLI, then the pub_sub example against it: a full publish/subscribe round trip.

MQTT 3.1.1 and 5.0 side by side

Two clients on different protocol versions share one broker, each publishing and both receiving both messages — the version is negotiated per connection, not per broker.

Topics and wildcards

An exact topic, a + single-level and a # multi-level filter watch the same topic tree; four publishes show which filter catches which message.

Shared subscriptions

Three workers join $share/workers/jobs/# and split nine jobs round-robin, three each, while an ordinary subscriber on the same topics still sees all nine.

Retained messages and Last Will

A subscriber that arrives after a retained publish still receives it; when a client vanishes without DISCONNECT, the broker publishes its will.

Authentication

A pluggable auth provider rejects two wrong logins and accepts the right one; the broker's own log then shows every decision and connection.

Long-lived connection

One connection held open across several keep-alive intervals with a heartbeat every two seconds — the shape of a background service or IoT device.

Recorded with VHS: just vhs-tape <name> or just vhs-all (tapes in examples/vhs; each tape runs the example of the same name).

Installation

Every published package is prefixed stem-mqtt- (mqtt-client / mqtt-broker are taken by unrelated projects on crates.io and PyPI); import names in code are unchanged. A package is available once a release has published it to that registry.

Every language gets the same two components — a client and a broker.

Detailed guides — install, requirements, a complete example, client and broker API, authentication, TLS, language-specific notes, building from source and troubleshooting — live in docs/languages/:

Callbacks are available everywhere except Dart and Haskell, which use the pull-style API.

Ruby isn't supported: UniFFI has no official Ruby backend and no production-ready generator exists (packaging/ruby).

Features

• MQTT 3.1.1 and MQTT 5.0 — one codec, version negotiated per connection (MqttVersion::V311 / MqttVersion::V5).
• QoS 0, 1, 2 — at-most-once, at-least-once, and the full exactly-once four-part handshake (PUBLISH → PUBREC → PUBREL → PUBCOMP), with real redelivery (DUP=1 resend on timeout) on both the client and broker side — not just a single fire-and-hope attempt.
• Wire codec for all control packet types — CONNECT/CONNACK, PUBLISH and its ack chain, SUBSCRIBE/SUBACK, UNSUBSCRIBE/UNSUBACK, PING, DISCONNECT, and MQTT 5 AUTH — with the enhanced-authentication exchange built on top (see Authentication below).
• TLS (client + broker, mqtt-client::TlsOptions / mqtt-broker::BrokerTlsConfig) — pure-Rust rustls, custom CA support, and mutual TLS (mTLS) client-certificate verification. No OpenSSL/system- TLS dependency.
• Opt-in client auto-reconnect (ConnectOptions::auto_reconnect) with exponential backoff, replaying every topic the client was subscribed to once reconnected.
• Last Will and Testament, retained messages, keep-alive pings.
• MQTT-over-WebSocket (--ws-port) alongside raw TCP, so browser clients (which can't open raw TCP sockets) can connect directly — see demo/ for a full browser pub/sub demo webpage.
• Authentication — pluggable username/password (MqttAuthProvider) and MQTT 5 enhanced authentication (multi-round challenge/response: SCRAM, Kerberos, OAuth, …) via MqttEnhancedAuthProvider on the broker and MqttAuthHandler on the client.
• Shared subscriptions ($share/<group>/<filter>) — one delivery per group, round-robin — routed through a topic-trie subscription index instead of a per-publish scan of every session.
• Session expiry (session_expiry_secs) for persistent sessions that never return.
• Event observation (MqttBrokerEventListener) and, for runtimes that can't receive callbacks (Dart, Haskell), a pull-style API: enable_message_queue / next_message on the client, enable_event_queue / next_event on the broker.
• UniFFI bindings — both crates build as cdylib/staticlib (cargo rustc --lib --crate-type cdylib,staticlib) and ship a uniffi-bindgen binary for Kotlin, Swift and Python; Go, C#, Java, Dart, Node.js and Haskell are generated through pinned community generators — all of them runtime-tested in CI against a real broker.
• Hardened by default — packet-size limits, bounded per-client send queues, keep-alive timeout detection, credential-redacting Debug, foreign callbacks that can't crash I/O tasks, and a decoder that is fuzzed (randomised tests in every build + cargo-fuzz targets in fuzz/) — see CHANGELOG.md.

Running the broker

cargo install stem-mqtt-broker                 # or: cargo run -p stem-mqtt-broker --bin mqtt-broker --
mqtt-broker --bind 0.0.0.0 --port 1883 --ws-port 8083 --allow-anonymous

Pre-built Linux and Windows mqtt-broker binaries are attached to every release.

Demos

docker compose up --build        # broker (TCP + WS) + a browser client at http://localhost:8090
just demo-dashboard              # broker + the Rust/Topcoat web dashboard (multi-client, live feed) on :3000

demo/README.md covers the plain-binaries path; packaging/k8s has a one-Pod and a Deployment + Service Kubernetes demo.

Workspace layout

There is deliberately no separate "core" crate: mqtt-client::protocol is a pure, synchronous codec with no networking or async dependencies, so it is the shared foundation for both client and broker.

Generating foreign-language bindings

Everything is driven by Nushell scripts; the generator table (languages, pinned generator versions, required UniFFI release, registry) lives in one file, scripts/bindings/spec.nu.

nu scripts/generate_bindings.nu kotlin            # kotlin | swift | python (built into UniFFI)
nu scripts/install_bindgens.nu                    # install the pinned third-party generators
nu scripts/generate_bindings.nu go mqtt-broker    # go | csharp | java | dart | node | haskell
just bindings-all                                 # or: just bindings-third-party

¹ Dart: foreign callbacks cannot be invoked from Rust's own threads (the VM aborts). Haskell: callback interfaces are exposed only as opaque handles. So MqttMessageListener, MqttAuthProvider, MqttBrokerEventListener and the enhanced-auth callbacks can't be implemented there — use the pull-style API that exists for exactly this: enable_message_queue / next_message(timeout_ms) on the client and enable_event_queue / next_event(timeout_ms) on the broker. Their smoke tests receive messages and observe broker events that way. Authentication uses the pull-style auth queue (enable_auth_queue / next_auth_request / resolve_auth_request). ² Upstream's Haskell generator had three bugs that made every binding unusable: constructors never lowered their arguments (generated code didn't compile), flat-error variants dropped their message (decoder failed with "left N trailing bytes"), and records sharing a field name broke the public module. packaging/haskell/uniffi-bindgen-haskell.patch fixes all three (and is proposed upstream, see packaging/haskell); install_bindgens.nu builds the generator from the pinned revision with the patch applied.

One solution per language, client and broker everywhere. Each language has exactly one generator (no parallel hand-written addons), and each component is self-contained: the broker owns its own QoS enum instead of importing the client's, so no generator ever meets a cross-crate "external type" (the Node generator rejects them) and every language gets a standalone broker package.

UniFFI bindgens can only read metadata from the UniFFI release they were built for, and the third-party generators disagree (most target 0.31, Haskell 0.32). generate_bindings.nu therefore builds the library for such a generator in a scratch workspace under target/uniffi-<version>/ pinned to the right release; your workspace is never touched.

Do the bindings actually work?

nu scripts/test_bindings.nu            # every language (missing toolchains are skipped)
nu scripts/test_bindings.nu go java --strict
just test-bindings go

For each language this builds the native libraries, generates the bindings, stages a scratch project with the smoke test from tests/bindings/<language>/ and runs it against real sockets: start a broker, subscribe, publish QoS 1, receive it through a foreign callback, refuse a bad client through a foreign auth callback, observe connections through a foreign event listener, stop. Languages that can't do callbacks run a reduced scenario (see the header of each test). Verdicts: PASS, FAIL, SKIP (toolchain missing). CI runs the same command for every language on every push, and each package is built and installed the way a user would before it is released. Maintainer docs: development · publishing.

License

MIT — see LICENSE.