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:

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

#[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:

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

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

[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.

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::<BincodeSerializer>::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.

use mqtt_typed_client::{BincodeSerializer, MqttClient, PublishOptions};
use std::time::Duration;

// protocol=5 selects the v5 stack
let (client, connection) =
    MqttClient::<BincodeSerializer>::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:

#[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<Reply, MyError> — 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.

Examples

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

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<u8> / Bytes)

Enable additional (feature-gated) serializers:

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

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<Foo>; 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)

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).

// 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. See 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 — the supported choice
backend-rumqttc-next	the maintained fork (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:

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<rumqttc::QoS> 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):

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

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::<BincodeSerializer>::connect(
        "mqtt://broker.hivemq.com:1883?client_id=demo_client"
    ).await?;

    // Direct topic operations
    let publisher = client.get_publisher::<SensorData>("sensors/temperature")?;
    let mut subscriber = client.subscribe::<SensorData>("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:

// 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 above. The eventloop.poll() dispatch loop is replaced by per-topic typed subscribers.

For a detailed comparison see: docs/COMPARISON_WITH_RUMQTTC.md

Alternatives

• rumqttc — 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 — Rust bindings to the Eclipse Paho C client; a fit when you need that mature C library or its feature set.
• 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 or http://www.apache.org/licenses/LICENSE-2.0)
• MIT license (LICENSE-MIT or http://opensource.org/licenses/MIT)

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 for detailed guidelines.

API Reference

For detailed API documentation, visit docs.rs/mqtt-typed-client.

See Also

• rumqttc - The underlying MQTT client library
• MQTT Protocol Specification - Official MQTT documentation
• Rust Async Book - Guide to async Rust programming