# MQTT Typed Client
A **type-safe async MQTT client** built on top of rumqttc
**Automatic topic routing and subscription management** with **compile-time guarantees**
[](https://github.com/holovskyi/mqtt-typed-client/actions)
[](https://crates.io/crates/mqtt-typed-client)
[](https://docs.rs/mqtt-typed-client)
[](https://github.com/holovskyi/mqtt-typed-client/blob/main/LICENSE-MIT)
[](https://blog.rust-lang.org/2025/01/09/Rust-1.85.0.html)
## The problem
Raw MQTT topics are stringly-typed. You hand-build them with `format!()`, split
them with `split('/')`, and deserialize payloads by hand — and the compiler
can't help when you swap two segments or typo a prefix:
```rust,ignore
// rumqttc: easy to get wrong, fails silently at runtime
let topic = format!("sensors/{}/{}/data", location, device_id); // swapped order? compiler shrugs
let payload = serde_json::to_vec(&reading)?;
client.publish(topic, QoS::AtLeastOnce, false, payload).await?;
```
With `mqtt-typed-client` the topic *is* a type. One derive turns the pattern into
a checked API — wrong order, wrong parameter type, or a typo won't compile:
```rust,ignore
#[mqtt_topic("sensors/{location}/{device_id}/data")]
struct SensorTopic { location: String, device_id: u32, payload: SensorReading }
// generated, type-checked: device_id must be u32, order is fixed
client.sensor_topic().publish("kitchen", 42, &reading).await?;
```
The bigger win is on the receiving side. Every rumqttc app grows the same
hand-written dispatch loop:
```rust,ignore
// rumqttc: one loop, all routing by hand
while let Ok(event) = eventloop.poll().await {
if let Event::Incoming(Packet::Publish(p)) = event {
if p.topic.starts_with("sensors/") {
let parts: Vec<_> = p.topic.split('/').collect(); // parse, convert, dispatch...
} else if p.topic.starts_with("alerts/") {
// more of the same...
}
}
}
```
Here each topic type gets its own subscriber and messages route to it
automatically. The loop, the `starts_with`, and the `split('/')` are gone:
```rust,ignore
let mut sensors = client.sensor_topic().subscribe().await?;
let mut alerts = client.alert_topic().subscribe().await?;
tokio::select! {
msg = sensors.receive() => { /* msg.device_id is already a u32 */ }
msg = alerts.receive() => { /* typed alert */ }
}
```
## Key Features
- **Topics as types** — named parameters parsed/validated at compile time
- **Typed parameters** — `{device_id}` can be `u32`, `Uuid`, or your own enum, not just `String`
- **Automatic routing** — one broker stream fanned out to typed subscribers; the hand-written `poll()` + `match` + `starts_with(...)` dispatch loop goes away
- **Reconnect that keeps subscriptions** — automatic resubscribe on reconnect (happy path), graceful shutdown, LWT
- **MQTT 5** — connect with protocol 5 for typed publish properties, request/response fields, and per-message v5 metadata (see [MQTT 5](#mqtt-5) below)
**MSRV**: Rust 1.85.1 (driven by default `bincode` serializer; can be lowered with alternative serializers)
## Quick Start
Add the crate and the few deps the derive example needs:
```toml
[dependencies]
mqtt-typed-client = "0.4.0"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
bincode = "2"
```
> `serde` is needed for the `Serialize`/`Deserialize` derives and `bincode` for the
> default serializer's `Encode`/`Decode`. Switch serializers (e.g. `json`) and the
> derive requirements change accordingly.
```rust,no_run
use mqtt_typed_client::prelude::*;
use serde::{Deserialize, Serialize};
use bincode::{Encode, Decode};
#[derive(Serialize, Deserialize, Encode, Decode, Debug)]
enum SensorStatus {
Active,
Inactive,
Maintenance,
}
#[derive(Serialize, Deserialize, Encode, Decode, Debug)]
struct SensorReading {
temperature: f64,
status: SensorStatus, // enum field
location_note: String, // string field for variety
}
// Define typed topic with automatic parameter extraction
#[mqtt_topic("sensors/{location}/{device_id}/data")]
struct SensorTopic {
location: String, // String parameter
device_id: u32, // Numeric parameter - automatic conversion!
payload: SensorReading,
}
#[tokio::main]
async fn main() -> Result<()> {
// Connect to MQTT broker
let (client, connection) = MqttClient::::connect(
"mqtt://broker.hivemq.com:1883?client_id=demo_client"
).await?;
// Get typed client for this specific topic - method generated by macro
// Returns a typed client for publishing and subscribing to messages
// with automatic parameter handling for this topic pattern
let topic_client = client.sensor_topic();
// Subscribe to all matching topics: "sensors/+/+/data"
// Returns typed subscriber that automatically extracts and converts
// topic parameters into struct fields
let mut subscriber = topic_client.subscribe().await?;
let reading = SensorReading {
temperature: 22.5,
status: SensorStatus::Active,
location_note: "Kitchen sensor near window".to_string(),
};
// Publish with automatic type conversion to specific topic: "sensors/kitchen/42/data"
// Parameters are automatically converted to strings and inserted into topic pattern
let _ = topic_client.publish("kitchen", 42u32, &reading).await?;
// ^^^^^^^^ ^^^^^
// String u32 -> automatically converts to "42" in topic
// Receive with automatic parameter extraction and conversion
if let Some(ReceiveEvent::Message(msg)) = subscriber.receive().await {
println!("Device {} in location '{}' reported: temp={}°C, status={:?}",
msg.device_id, // u32 (converted from "42" in topic)
msg.location, // String (extracted from topic)
msg.payload.temperature, msg.payload.status);
}
connection.shutdown().await?;
Ok(())
}
```
## MQTT 5
Connect with protocol 5 to send MQTT 5 PUBLISH properties. MQTT 3.1.1 stays the
default; the v5 stack ships in the same crate with no MSRV or edition bump.
```rust,ignore
use mqtt_typed_client::{BincodeSerializer, MqttClient, PublishOptions};
use std::time::Duration;
// protocol=5 selects the v5 stack
let (client, connection) =
MqttClient::::connect("mqtt://broker:1883?protocol=5").await?;
// per-publish v5 properties via the builder
let opts = PublishOptions::builder()
.message_expiry(Duration::from_secs(60))
.content_type("application/json")
.response_topic("replies/abc") // request/response building block
.user_property("trace-id", "xyz")
.build();
client.sensor_topic().publish_with("kitchen", 42u32, &reading, opts).await?;
```
On a v4 connection a publish carrying any v5 property is rejected with a typed
`CapabilityError::RequiresV5`, never silently dropped. Inbound v5 properties are
exposed via `Mqtt5Meta` on the received message.
### Typed request/response (RPC)
Declare a reply type once and get a typed `call` / `serve` pair — the v5
`response_topic` + `correlation_data` plumbing is handled for you:
```rust,ignore
#[mqtt_topic("devices/{device_id}/rpc/get_temp", response = TempReply)]
struct GetTemp { device_id: String, payload: TempQuery }
// caller (v5 connection required)
let rpc = client.rpc().await?;
let reply: TempReply = rpc.get_temp().call("sensor-42", &query).await?;
// server: `serve` runs the loop and publishes the reply — the handler just
// returns it (a pull `responder()` / `receive()` API is also available)
client.get_temp().serve(|req: GetTemp| async move {
TempReply { value: read_sensor(&req.device_id).await }
}).await?;
```
Remote application errors are the reply type's own concern —
`response = Result` — distinct from `RequestError` (timeout /
transport / decode). Scale a service across N instances by running each with
`serve_shared("group", handler)` — MQTT 5 shared subscriptions load-balance
requests across the group (requires a v5 connection). See
`examples/011_rpc_request_response.rs` and `examples/015_rpc_serve_shared.rs`.
More v5 is tracked in [docs/ROADMAP_v5.md](https://github.com/holovskyi/mqtt-typed-client/blob/main/docs/ROADMAP_v5.md).
## Examples
See [examples/](https://github.com/holovskyi/mqtt-typed-client/tree/main/examples) - Complete usage examples with source code
- `000_hello_world.rs` - Basic publish/subscribe with macros
- `001_ping_pong.rs` - Multi-client communication
- `002_configuration.rs` - Advanced client configuration
- `003_hello_world_lwt.rs` - Last Will & Testament
- `004_hello_world_tls.rs` - TLS/SSL connections
- `005_hello_world_serializers.rs` - Custom serializers
- `006_retain_and_clear.rs` - Retained messages
- `007_custom_patterns.rs` - Custom topic patterns
- `008_modular_example.rs` - Modular application structure
- `009_message_metadata.rs` - Per-message metadata (QoS, retain, dup)
- `010_connection_state.rs` - Observing the connection lifecycle
- `011_rpc_request_response.rs` - Typed request/response (RPC) over MQTT 5
- `012_mqtt5_properties.rs` - MQTT 5 publish properties and reading v5 metadata
- `013_subscribe_options.rs` - MQTT 5 subscribe options (no-local, retain handling)
- `014_shared_subscriptions.rs` - MQTT 5 shared subscriptions (`$share`)
- `015_rpc_serve_shared.rs` - Scaled typed RPC service (`serve_shared`)
- `016_custom_payload_formats.rs` - Interop with bare text/compound/byte payloads (`TextSerializer`/`RawBytesSerializer`)
- `100_all_serializers_demo.rs` - All serializers side by side
- `102_multi_serializer_macro.rs` - Per-topic custom serializers
Run examples:
```bash
cargo run --example 000_hello_world
```
## Serialization Support
Multiple serialization formats are supported via feature flags:
- `bincode` - Binary serialization (default, most efficient)
- `json` - JSON serialization (default, human-readable)
- `messagepack` - MessagePack binary format
- `cbor` - CBOR binary format
- `postcard` - Embedded-friendly binary format
- `ron` - Rusty Object Notation
- `flexbuffers` - FlatBuffers FlexBuffers
- `protobuf` - Protocol Buffers (requires generated types)
Two more serializers are **always available** (they pull no extra dependency), so
they need no feature flag:
- `TextSerializer` - bare `Display`/`FromStr` payloads (numbers, tokens, compounds)
- `RawBytesSerializer` - opaque byte pass-through (`Vec` / `Bytes`)
Enable additional (feature-gated) serializers:
```toml
[dependencies]
mqtt-typed-client = { version = "0.4.0", features = ["messagepack", "cbor"] }
```
Custom serializers can be implemented by implementing the `MessageSerializer` trait.
**Choosing a serializer:**
- **Interop with an existing (non-Rust) system** → `TextSerializer` /
`RawBytesSerializer`. Payloads are *bare* values (`21.5`, `ON`, Homie `255,0,0`,
raw bytes), not framed structs — JSON would quote strings (`"21.5"`) and bincode
would length-prefix, so neither round-trips.
- **Performance / embedded / no_std** → `bincode`, `postcard`.
- **Human-readable** → `json`, `ron`.
> **Interop tip.** `TextSerializer` is a blanket over any `T: Display + FromStr` —
> the *same* machinery topic parameters already use — so `String`, numbers, `bool`,
> and your own comma-separated `Color`/`Coord` types all round-trip as bare text.
> See [`016_custom_payload_formats.rs`](examples/016_custom_payload_formats.rs) for
> a worked example (Homie color, bare `f64`, string token, raw-byte tunnel).
## Per-Topic Serializer Override
By default every topic uses the client's serializer. You can override it for a
specific topic type — handy for legacy formats or gradual migrations:
```rust,ignore
use mqtt_typed_client_macros::mqtt_topic;
// This topic always uses JSON, regardless of the client's default serializer.
#[mqtt_topic("legacy/devices/{id}/status", serializer = JsonSerializer)]
struct LegacyStatus {
id: u32,
payload: DeviceStatus,
}
```
The generated typed client works for custom-serializer topics too —
`client.legacy_status()...` publishes/subscribes as usual; the facade swaps in
the concrete serializer internally (via `clone_with_serializer`), so you keep the
ergonomic surface, not just the low-level API.
Limitation:
- Only a simple type path is accepted for `serializer = ...`. For a generic
serializer, declare a type alias first: `type MySer = MySerializer;` then
use `serializer = MySer`.
## Topic Pattern Matching
Supports MQTT wildcard patterns with named parameters:
- `{param}` - Named parameter (equivalent to `+` wildcard)
- `{param:#}` - Multi-level named parameter (equivalent to `#` wildcard)
```rust,ignore
use mqtt_typed_client_macros::mqtt_topic;
// Traditional MQTT wildcards
#[mqtt_topic("home/+/temperature")] // matches: home/kitchen/temperature
struct SimplePattern { payload: f64 }
// Named parameters (recommended)
#[mqtt_topic("home/{room}/temperature")] // matches: home/kitchen/temperature
struct NamedPattern {
room: String, // Automatically extracted: "kitchen"
payload: f64
}
// Multi-level parameters
#[mqtt_topic("logs/{service}/{path:#}")] // matches: logs/api/v1/users/create
struct LogPattern {
service: String, // "api"
path: String, // "v1/users/create"
payload: String // Changed from Data to String
}
```
## Runtime Pattern Override
The pattern in `#[mqtt_topic("...")]` is the *default*, but you can override it at
runtime — as long as the parameter set (same names and types) stays the same. Only the
literal segments and prefix may differ. Handy when the topic layout is decided at deploy
time rather than compile time: environment prefixes, multi-tenant tenancy, or legacy
formats. The override is validated when you call it (fast-fail on a parameter mismatch).
```rust,ignore
// declared once: #[mqtt_topic("greetings/{language}/{sender}")]
// Subscribe with an environment prefix from config:
let mut subscriber = client.greeting_topic()
.subscription()
.with_pattern("dev/greetings/{language}/{sender}")? // ✅ same {language}, {sender}
.subscribe()
.await?;
// Publish to a multi-tenant layout:
let publisher = client.greeting_topic()
.get_publisher_to("tenant_42/greetings/{language}/{sender}", "rust", "alice")?;
publisher.publish(&message).await?;
// Last Will with a custom pattern:
let lwt = GreetingTopic::last_will_to(
"dev/greetings/{language}/{sender}", "rust", "client", lwt_message,
)?;
// ❌ Rejected — different parameter names:
// .with_pattern("greetings/{room}/{device_id}")
```
Parameter *reordering* (same set, different order) is not accepted yet — tracked in
[#4](https://github.com/holovskyi/mqtt-typed-client/issues/4). See
[`examples/007_custom_patterns.rs`](https://github.com/holovskyi/mqtt-typed-client/blob/main/examples/007_custom_patterns.rs)
for a full runnable example.
## TLS and Transport
Transport security and extras are opt-in via feature flags:
| Feature | Effect |
| --- | --- |
| `tls-rustls` *(default)* | TLS via rustls with the aws-lc-rs provider |
| `tls-rustls-no-provider` | rustls without a bundled crypto provider — bring your own (e.g. ring) and avoid the aws-lc build |
| `tls-native` | Compile in the platform's native-tls (reachable via the backend escape hatch for now) |
| `websocket` | MQTT over WebSocket |
| `proxy` | Connect through an HTTP/HTTPS proxy |
(The 0.2 `rumqttc-*` feature names are gone as of 0.4, as 0.3 announced when it
deprecated them. Use the names above; `rumqttc-url` has no successor — URL
parsing is built in.)
## MQTT backend
The MQTT stack underneath is chosen at compile time by a `backend-*` feature.
Exactly one must be enabled — enabling none or both is a compile error with a
message saying so.
| Feature | Backend |
| --- | --- |
| `backend-rumqttc` *(default)* | upstream [`rumqttc`](https://crates.io/crates/rumqttc) — the supported choice |
| `backend-rumqttc-next` | the maintained [fork](https://github.com/thehouseisonfire/rumqtt) (`rumqttc-v4-next` / `rumqttc-v5-next`) — **experimental**, requires Rust 1.89 |
Since `backend-rumqttc` is a default feature, picking the other one means
turning defaults off:
```toml
mqtt-typed-client = { version = "0.4.0", default-features = false,
features = ["backend-rumqttc-next", "macros", "json"] }
```
The typed client API is the same either way — topics, publish, subscribe, RPC
and options name no backend type. The one exception is the deliberate escape
hatch: the semver-exempt `unstable-backend-api` surface has by definition the
shape of whichever backend is selected. (`QoS::to_rumqttc()` /
`to_rumqttc_v5()` / `From` used to be a second exception, visible
only under `backend-rumqttc`. They are no longer enabled by this crate under
any backend — see the CHANGELOG for the one-line migration.) Beyond
those, the fork's extra facilities (manual acks, publish tracking) are not yet
reachable through this crate's API, so there is no reason to switch unless you
are helping to exercise the second backend.
For custom TLS setups you can build the rustls config yourself. The crate
re-exports the backend's `rustls` (version-matched, so you don't add a
separate `rustls` dependency that could drift out of sync):
```rust,no_run
use mqtt_typed_client::rustls::{ClientConfig, RootCertStore};
# fn build_tls_config() -> ClientConfig {
let mut root_store = RootCertStore::empty();
// Add your trusted roots to `root_store` here (e.g. parsed from a PEM file).
ClientConfig::builder()
.with_root_certificates(root_store)
.with_no_client_auth()
# }
```
See [`examples/004_hello_world_tls.rs`](https://github.com/holovskyi/mqtt-typed-client/blob/main/examples/004_hello_world_tls.rs) for a complete TLS example,
including loading a CA certificate from a PEM file.
## Advanced Usage: Low-Level API
For cases where you need direct control without macros:
```rust,no_run
use mqtt_typed_client::prelude::*;
use serde::{Deserialize, Serialize};
use bincode::{Encode, Decode};
#[derive(Serialize, Deserialize, Encode, Decode, Debug)]
struct SensorData {
temperature: f64,
humidity: f64,
}
#[tokio::main]
async fn main() -> Result<()> {
let (client, connection) = MqttClient::::connect(
"mqtt://broker.hivemq.com:1883?client_id=demo_client"
).await?;
// Direct topic operations
let publisher = client.get_publisher::("sensors/temperature")?;
let mut subscriber = client.subscribe::("sensors/+").await?;
let data = SensorData { temperature: 23.5, humidity: 45.0 };
let _ = publisher.publish(&data).await?;
match subscriber.receive().await {
Some(ReceiveEvent::Message(msg)) => {
println!("Received from {} (qos {:?}): {:?}",
msg.topic.topic_path(), msg.meta.qos, msg.payload)
}
Some(ReceiveEvent::DecodeFailed(f)) => {
eprintln!("Deserialization error at {}: {:?}", f.topic.topic_path(), f.error)
}
Some(ReceiveEvent::Lagged { missed }) => {
eprintln!("Lagged: {} messages dropped", missed)
}
_ => {}
}
connection.shutdown().await?;
Ok(())
}
```
## What mqtt-typed-client adds over rumqttc
**Publishing:**
```rust,ignore
// rumqttc - manual topic construction and serialization
let sensor_id = "sensor001";
let data = SensorData { temperature: 23.5 };
let topic = format!("sensors/{}/temperature", sensor_id);
let payload = serde_json::to_vec(&data)?;
client.publish(topic, QoS::AtLeastOnce, false, payload).await?;
// mqtt-typed-client - type-safe, automatic
topic_client.publish(&sensor_id, &data).await?;
```
**Subscribing with routing:** see [The problem](#the-problem) above. The
`eventloop.poll()` dispatch loop is replaced by per-topic typed subscribers.
**For a detailed comparison see:** [docs/COMPARISON_WITH_RUMQTTC.md](https://github.com/holovskyi/mqtt-typed-client/blob/main/docs/COMPARISON_WITH_RUMQTTC.md)
## Alternatives
- **[rumqttc](https://github.com/bytebeamio/rumqtt)** — the async MQTT client this
crate builds on. Use it directly when you want full manual control over topics,
serialization, and the event loop.
- **[paho-mqtt](https://crates.io/crates/paho-mqtt)** — Rust bindings to the Eclipse
Paho C client; a fit when you need that mature C library or its feature set.
- **[ntex-mqtt](https://crates.io/crates/ntex-mqtt)** — MQTT client and server built on
the ntex framework; worth a look if you're already in that ecosystem or need a broker.
Reach for `mqtt-typed-client` when you want typed topic routing and automatic
(de)serialization on top of rumqttc, without hand-writing the dispatch layer.
It is a layer over rumqttc, not a replacement: if your app handles one or two
topics and you want to drive the event loop yourself, raw rumqttc is less machinery.
## License
This project is licensed under either of
* Apache License, Version 2.0, ([LICENSE-APACHE](https://github.com/holovskyi/mqtt-typed-client/blob/main/LICENSE-APACHE) or )
* MIT license ([LICENSE-MIT](https://github.com/holovskyi/mqtt-typed-client/blob/main/LICENSE-MIT) or )
at your option.
## Contributing
Contributions are welcome! Please feel free to submit a Pull Request.
1. Fork the repository
2. Create your feature branch (`git checkout -b feature/amazing-feature`)
3. Commit your changes (`git commit -m 'Add some amazing feature'`)
4. Push to the branch (`git push origin feature/amazing-feature`)
5. Open a Pull Request
See [CONTRIBUTING.md](https://github.com/holovskyi/mqtt-typed-client/blob/main/CONTRIBUTING.md) for detailed guidelines.
## API Reference
For detailed API documentation, visit [docs.rs/mqtt-typed-client](https://docs.rs/mqtt-typed-client).
## See Also
- [rumqttc](https://github.com/bytebeamio/rumqtt) - The underlying MQTT client library
- [MQTT Protocol Specification](https://mqtt.org/) - Official MQTT documentation
- [Rust Async Book](https://rust-lang.github.io/async-book/) - Guide to async Rust programming