MQTT Typed Client

A type-safe async MQTT client built on top of rumqttc

Automatic topic routing and subscription management with compile-time guarantees

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?;

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, no manual if topic.starts_with(...)
• Pluggable serialization — Bincode, JSON, MessagePack, CBOR and more behind one trait
• Reconnect that keeps subscriptions — automatic resubscribe on reconnect (happy path), graceful shutdown, LWT

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.3.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
    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(())
}

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

Enable additional serializers:

[dependencies]
mqtt-typed-client = { version = "0.3.0", features = ["messagepack", "cbor"] }

Custom serializers can be implemented by implementing the MessageSerializer trait.

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,
}

Limitations:

• Using a custom serializer disables the generated TypedClient extension for that topic (typed clients require a generic serializer parameter, while a custom serializer is a concrete type).
• Only a simple type path is accepted. 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
}

TLS and Transport

Transport security and extras are opt-in via feature flags:

(The 0.2 rumqttc-* feature names still work as deprecated aliases and will be removed in 0.4 — except rumqttc-url, which is gone: URL parsing is built in.)

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

// rumqttc - manual pattern matching and dispatching
// while let Ok(event) = eventloop.poll().await {
//     if let Event::Incoming(Packet::Publish(publish)) = event {
//         if publish.topic.starts_with("sensors/") {
//             // Manual topic parsing, manual deserialization...
//         } else if publish.topic.starts_with("alerts/") {
//             // More manual parsing...
//         }
//     }
// }

// mqtt-typed-client - automatic routing to typed handlers
let mut sensor_sub = client.sensor_topic().subscribe().await?;
let mut alert_sub = client.alert_topic().subscribe().await?;

tokio::select! {
    msg = sensor_sub.receive() => { /* typed sensor data ready */ }
    msg = alert_sub.receive() => { /* typed alert data ready */ }
}

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.

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