stem-mqtt-client 0.3.3

An MQTT 3.1.1 / MQTT 5.0 client implemented in Rust with UniFFI bindings for Kotlin, Swift, and Python.
Documentation

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.

CI License: MIT MSRV

Crate Version Downloads Docs
stem-mqtt-client crates.io downloads docs.rs
stem-mqtt-broker crates.io downloads docs.rs

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

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

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

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

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

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

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

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 — see Publishing for what goes where.

Language Install Import
Rust cargo add stem-mqtt-client stem-mqtt-broker use mqtt_client::…; use mqtt_broker::…;
Python pip install stem-mqtt-client stem-mqtt-broker import mqtt_client, mqtt_broker
Node.js npm i stem-mqtt stem-mqtt/client · stem-mqtt/broker
Kotlin (JVM/Android) io.github.sorinirimies.stemmqtt:stem-mqtt-kotlin (Maven Central) import uniffi.mqtt_client.*
Java (JDK 22+) io.github.sorinirimies.stemmqtt:stem-mqtt-java (Maven Central) import uniffi.mqtt_client.*;
C# dotnet add package StemMqtt using uniffi.mqtt_client;
Go go get github.com/sorinirimies/stem-mqtt-go …/stem-mqtt-go/mqtt_client
Dart dart pub add stem_mqtt package:stem_mqtt/mqtt_client.dart
Haskell cabal install stem-mqtt import UniFFI.MqttClient
Swift StemMqttSwift-<version>.zip from the GitHub Release import MqttClient

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/:

Language Guide Notes
Rust rust.md tokio; the quick start above
Python python.md asyncio; wheels bundle the native library
Node.js node.md ESM, server-side; Promises
Kotlin kotlin.md coroutines; JVM and Android
Java java.md JDK 22+ (FFM); CompletableFuture
C# csharp.md async/await; .NET 8+
Go go.md cgo; install the native libraries separately
Dart dart.md pull-style API; needs cargo to build
Haskell haskell.md pull-style API; needs cargo to build
Swift swift.md async/await; XCFrameworks for Apple platforms

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 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
Option Meaning
--bind, --port TCP listener (default 0.0.0.0:1883)
--ws-port also accept MQTT-over-WebSocket (browser clients)
--allow-anonymous accept clients without credentials
--max-clients connection limit (0 = unlimited)
--log-level error … trace (default info)

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

Path Description
crates/mqtt-client Async, tokio-based MQTT client; also hosts the wire-protocol codec (mqtt_client::protocol) shared with the broker.
crates/mqtt-broker Standalone broker (library + mqtt-broker CLI), reusing the client's codec.
demo/ Browser client (web/) and a Rust + Topcoat dashboard (dashboard/).
tests/bindings One end-to-end smoke test per language, run by scripts/test_bindings.nu.
packaging/ Per-language package templates, Dockerfile and Kubernetes manifests.
scripts/ Nushell scripts for bindings, packaging, publishing and releases (tested in scripts/tests).
fuzz/ cargo-fuzz targets for the packet decoder and topic matching.

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
Language Generator Runtime-tested Status
Kotlin UniFFI (built in) Gradle packaging test in CI stable
Swift UniFFI (built in) ✅ client + broker + all callbacks stable
Python UniFFI (built in) ✅ client + broker + all callbacks stable
Go uniffi-bindgen-go ✅ client + broker + all callbacks stable
C# uniffi-bindgen-cs ✅ client + broker + all callbacks stable
Java (JDK 22+) uniffi-bindgen-java ✅ client + broker + all callbacks stable
Dart uniffi-dart ✅ client + broker through the pull-style API ¹ experimental
Node.js uniffi-bindgen-node-js ✅ client + broker + all callbacks experimental
Haskell uniffi-bindgen-haskell + our patch (PR #3) ✅ client + broker through the pull-style API ¹ experimental

¹ 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 built-in allow_anonymous rule. ² 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 (Gitea .gitea/workflows/ci.yml, Linux runner; mirrored in .github/workflows/ci.yml) runs the same command per language (test-bindings-runtime), builds each package (test-packaging: dotnet pack, gradle build, npm pack, OCI bundle) and runs the Kotlin Gradle test; the release workflow runs the runtime test before packaging each language.

Publishing

nu scripts/publish_packages.nu stage <language>                       # bindings + native libs -> dist/<language>
nu scripts/publish_packages.nu publish <language> <version> --dry-run  # print the plan; --target github|gitea|public

Public registries

The Gitea release workflow publishes to the indexes people install from (--target public). Each job needs its own repository secret and skips quietly without it:

Language Registry Package Secret(s) Notes
Rust crates.io stem-mqtt-client, stem-mqtt-broker CRATES_IO_TOKEN
Python PyPI stem-mqtt-client, stem-mqtt-broker PYPI_API_TOKEN manylinux_2_28 wheels (zig)
Node.js npmjs.org stem-mqtt NPM_TOKEN granular token with bypass 2FA; Linux x86_64 prebuild today
C# NuGet.org StemMqtt NUGET_API_KEY
Kotlin / Java Maven Central io.github.<owner>.stemmqtt:stem-mqtt-kotlin / …-java MAVEN_CENTRAL_USERNAME, MAVEN_CENTRAL_PASSWORD, SIGNING_KEY, SIGNING_PASSWORD verify the io.github.<owner> namespace first
Haskell Hackage stem-mqtt HACKAGE_TOKEN source package, builds the bundled Rust with cargo
Dart pub.dev stem_mqtt — (OIDC) from GitHub Actions (publish-pubdev.yml); first release by hand
Go Go modules github.com/<owner>/stem-mqtt-go GO_MODULE_TOKEN pushes the module to its own repo and tags it
Swift — release asset (XCFrameworks) — needs macOS

The Haskell and Dart packages are exactly what their runtime tests exercise. The Node, C#, Java, Kotlin and Go packages bundle or expect native libraries only for the platforms the release runner builds (Linux x86_64 today); other platforms use the release assets or a source build. Maven Central and Hackage releases cannot be deleted — read the job logs of the first release.

Project registries

Besides the public indexes, every language also goes to GitHub Packages (Maven, npm, NuGet, and OCI bundles on ghcr.io for Go, Dart and Haskell) and/or your Gitea instance's own registries (--target gitea --base-url http://host:3000, secret PACKAGES_TOKEN with write:package). Kotlin's Android AAR and the Swift XCFrameworks need the GitHub workflow (NDK / macOS). A manual workflow (.github/workflows/publish-packages.yml) dry-runs or publishes a throw-away version such as 0.0.0-rc.1 without cutting a release. See packaging/README.md for the artifact layout.

Development

Every task below is also a just recipe — run just --list for the full set (bindings, packaging, changelog, version bumps, releases, multi-remote git push/pull, ...):

cargo build --workspace
cargo test  --workspace
cargo clippy --workspace --all-targets -- -D warnings
cargo fmt --all -- --check

Or run everything CI runs in one shot:

just check-all      # fmt + clippy + test + doc + nu script tests
./scripts/check.sh  # equivalent, no `just`/`nu` required

Bindings, packages and CI

just install-bindgens          # the pinned third-party generators (scripts/bindings/spec.nu)
just test-bindings             # runtime-test every language (or: just test-bindings go java)
just package-verify-all        # compile + pack every publishable package, no upload
just ci-bindings               # all of the above, strict — what the Gitea CI runs
just clean-bindings            # remove generated artifacts and scratch dirs
just check-bindgen-updates     # are the pinned generators behind their upstream? (weekly in CI)
just ci-image                  # CI image with every toolchain + generator pre-installed

CI runs on Gitea (.gitea/workflows/ci.yml, Linux runner; also mirrored in .github/workflows/ci.yml): fmt, clippy, test, doc, the nu tests, then one test-bindings-runtime job per language, the Kotlin Gradle test, and one test-packaging job per package. A weekly scheduled run catches upstream generator drift.

Nushell scripts

Deterministic release/CI plumbing (tag validation, version bump, changelog, crates.io publish, quality gate) lives in scripts/*.nu as tested Nushell scripts rather than ad-hoc shell in CI YAML — see scripts/tests/ (nu scripts/tests/run_all.nu, also reachable as just test-nu).

CI/CD

  • Gitea (.gitea/workflows/) — the Linux runner that does the real work: ci.yml (fmt, clippy, tests, docs, nu tests, a runtime test and a package build per language, toolchain setup retried once on network flakes), release.yml (on a vX.Y.Z tag: broker binaries for Linux and Windows, Gitea release, then crates.io and every registry above), deps-update.yml.
  • GitHub (.github/workflows/) — mirrors CI and adds what needs GitHub-hosted runners: macOS/iOS/Android builds, the Swift package, GitHub Packages, PyPI wheels for several platforms, and publish-pubdev.yml.

Release process

just release <version>   # preflight + bump + quality gate + tag + push

See the justfile for the multi-remote (GitHub + Gitea) workflow: just bump, just release-all, just sync-gitea, just migrate-gitea, …

License

MIT — see LICENSE.