Rust MQTT with rumqttc: Async Publish/Subscribe over TLS with Tokio

This Rust MQTT rumqttc tutorial shows how rumqttc gives you a pure-Rust, async client that fits naturally into Tokio services, from edge gateways to backend ingest workers. In this tutorial you will build a complete rumqttc client that connects over TLS with rustls on port 8883, publishes JSON telemetry, receives commands and keeps running through network drops.

Why rumqttc for Rust MQTT

rumqttc is the client half of the rumqtt project. It is written entirely in Rust, supports MQTT 3.1.1 at the crate root and MQTT 5 in its v5 module, and offers both a synchronous Client and an async AsyncClient built on Tokio. TLS comes from rustls by default, so you do not need OpenSSL on the target machine, which is convenient when cross-compiling for ARM gateways.

The design is different from callback-based libraries in other languages, and it is worth understanding before writing code:

Nothing happens on the network unless something is polling the event loop. Requests from AsyncClient go into a bounded channel; the event loop drains it, sends packets, reads incoming packets and handles pings. If poll() returns an error, the next call reconnects. That single loop is where your connection logic lives.

Project Setup

Create a new binary crate:

cargo new rust-mqtt-demo
cd rust-mqtt-demo

Edit Cargo.toml:

[package]
name = "rust-mqtt-demo"
version = "0.1.0"
edition = "2021"

[dependencies]
rumqttc = "0.25"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"

The default use-rustls feature enables TLS through rustls with the aws-lc-rs crypto provider. Building that provider needs a C compiler and CMake on the build machine; on Windows, NASM may also be required. Runtime machines need nothing extra.

For the first run, use the free public MQTT broker at broker.connectmq.com, which accepts anonymous clients with any non-empty, unique client ID. Choose a random prefix such as connectmq/demo/m3c8wd; the public area is readable by anyone, so publish only test data.

The Complete rumqttc Example

Replace src/main.rs:

use rumqttc::{AsyncClient, Event, LastWill, MqttOptions, Packet, QoS, Transport};
use serde::Serialize;
use std::env;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::time;

#[derive(Serialize)]
struct Reading {
    device: String,
    temperature: f64,
    ts: u64,
}

fn var(key: &str, default: &str) -> String {
    env::var(key).unwrap_or_else(|_| default.to_string())
}

#[tokio::main]
async fn main() {
    let host = var("MQTT_HOST", "broker.connectmq.com");
    let client_id = var("MQTT_CLIENT_ID", "rust-demo-m3c8wd");
    let prefix = var("MQTT_PREFIX", "connectmq/demo/m3c8wd");
    let status_topic = format!("{prefix}/status");
    let command_filter = format!("{prefix}/cmd/#");
    let telemetry_topic = format!("{prefix}/telemetry");

    let mut options = MqttOptions::new(client_id.clone(), host, 8883);
    options.set_keep_alive(Duration::from_secs(30));
    options.set_clean_session(false);
    options.set_last_will(LastWill::new(&status_topic, "offline", QoS::AtLeastOnce, true));
    options.set_transport(Transport::tls_with_default_config());
    if let Ok(user) = env::var("MQTT_USERNAME") {
        options.set_credentials(user, env::var("MQTT_PASSWORD").unwrap_or_default());
    }

    let (client, mut eventloop) = AsyncClient::new(options, 32);

    // Publisher task: sends a reading every 5 seconds.
    let publisher = client.clone();
    let device = client_id.clone();
    tokio::spawn(async move {
        let mut ticker = time::interval(Duration::from_secs(5));
        let mut temperature = 21.0_f64;
        loop {
            ticker.tick().await;
            temperature += 0.1;
            let ts = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
            let reading = Reading { device: device.clone(), temperature, ts };
            let payload = serde_json::to_vec(&reading).expect("serialize");
            if let Err(e) = publisher
                .publish(&telemetry_topic, QoS::AtLeastOnce, false, payload)
                .await
            {
                eprintln!("publish request failed: {e}");
            }
        }
    });

    // Event loop: drives the connection, handles reconnects and incoming messages.
    loop {
        match eventloop.poll().await {
            Ok(Event::Incoming(Packet::ConnAck(ack))) => {
                println!("connected: {:?}", ack.code);
                // Non-blocking requests: never await the client inside the poll loop.
                if let Err(e) = client.try_subscribe(&command_filter, QoS::AtLeastOnce) {
                    eprintln!("subscribe request failed: {e}");
                }
                if let Err(e) = client.try_publish(&status_topic, QoS::AtLeastOnce, true, "online") {
                    eprintln!("status request failed: {e}");
                }
            }
            Ok(Event::Incoming(Packet::Publish(msg))) => {
                println!("command on {}: {}", msg.topic, String::from_utf8_lossy(&msg.payload));
            }
            Ok(_) => {}
            Err(e) => {
                eprintln!("connection error: {e}; reconnecting in 5 s");
                time::sleep(Duration::from_secs(5)).await;
            }
        }
    }
}

Run it:

cargo run

In another terminal, watch and send a command:

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-m3c8wd -t 'connectmq/demo/m3c8wd/#' -v
mosquitto_pub -h broker.connectmq.com -p 1883 -i send-m3c8wd -t 'connectmq/demo/m3c8wd/cmd/fan' -m 'on'

You should see telemetry every five seconds, a retained online on the status topic, and the command printed by the Rust program.

How the rumqttc Client Works

TLS with rustls

Transport::tls_with_default_config() builds a rustls configuration that trusts the operating system's root certificates. The broker uses a Let's Encrypt certificate, which every mainstream OS trusts, so no certificate files are needed. Port 8883 plus TLS transport is the whole setup. If you need a custom CA or client certificates, build a TlsConfiguration yourself and pass it to Transport::tls_with_config.

Why try_subscribe inside the loop

The most common rumqttc bug is a deadlock: calling client.subscribe(...).await inside the same task that polls the event loop. If the request channel is full, the await waits for space, but space only frees up when poll() runs, which it never will. Inside the loop, use try_subscribe and try_publish, which return immediately. In other tasks, the async versions are fine and give you natural back-pressure.

Reconnects and sessions

When poll() returns an error, the connection is gone. Calling poll() again reconnects, and the sleep prevents a tight loop while the network is down. With set_clean_session(false), the broker keeps subscriptions and queued QoS 1 messages between connections, and the client re-sends unacknowledged QoS 1 publishes. Resubscribing on every ConnAck is still a good habit, because the broker may have expired the session after a long outage. On the public broker, sessions last at most 30 minutes.

Last Will

LastWill tells the broker to publish a retained offline if the client vanishes without disconnecting. Together with the retained online sent after each connect, any dashboard can show whether the service is up.

Moving to a Private Broker

For production, create a broker in the ConnectMQ portal and use its generated username, password and client ID. Client IDs are bound to the credentials, so the ID must match exactly:

export MQTT_HOST=broker.connectmq.com
export MQTT_CLIENT_ID=YOUR_CLIENT_ID
export MQTT_USERNAME=YOUR_USERNAME
export MQTT_PASSWORD=YOUR_PASSWORD
export MQTT_PREFIX=plant1/line2/edge-gateway
cargo run --release

MQTT 5 with rumqttc

The rumqttc::v5 module mirrors the same API with MQTT 5 types. It gives you properties such as message expiry, user properties and reason codes, and a session expiry interval in place of the clean-session flag. Switching is mostly a matter of changing imports and adapting to Publish fields that use bytes for topics. The MQTT 5 features guide explains which properties are worth using.

Cross-Compiling for Edge Gateways

Rust is popular on ARM gateways such as Raspberry Pi-class boards, and rumqttc travels well there because it needs no OpenSSL. The main hurdle is the crypto provider, which compiles C code. Tools like cross handle the toolchain for you: cross build --release --target aarch64-unknown-linux-gnu produces a binary you can copy to a 64-bit Pi. Remember that the target still needs the system CA bundle, because the default TLS configuration loads the platform's root certificates at startup.

Production Checklist

1. Unique client ID per process. Two processes with the same ID disconnect each other in a loop.
2. Size the channel. The capacity passed to AsyncClient::new limits queued requests. Bursty publishers need more headroom.
3. Back off on errors. Never call poll() in a tight loop after errors; sleep, ideally with growing delays.
4. Pick QoS deliberately. QoS 1 suits most telemetry; the Free plan supports QoS 0 and 1. See MQTT QoS levels explained.
5. Keep payloads small. The public broker accepts up to 64 KB per message, the Free plan 256 KB.
6. Keep secrets in the environment or a secret store, not in the binary.

Troubleshooting

The crate's API reference is on docs.rs.

Related Tutorials

If parts of your system are written in other languages, the Go MQTT client tutorial and the Java MQTT client guide implement the same telemetry-and-commands pattern. The MQTT client libraries overview compares clients across ecosystems.

Run Your Rust Services on ConnectMQ

A fully managed MQTT service means your Rust gateway only needs one outbound TLS connection, with no broker to patch or certificates to renew. ConnectMQ supports MQTT 3.1.1 and 5.0 with TLS on 8883 and a live topic explorer for checking what your client sends. Create a free Serverless broker at app.connectmq.com with no credit card and set the four environment variables above.

Frequently Asked Questions

Does rumqttc support TLS?

Yes. With the default use-rustls feature, Transport::tls_with_default_config() connects over TLS using the operating system's trusted root certificates. Use port 8883.

Does rumqttc reconnect automatically?

It reconnects when you call eventloop.poll() again after an error. Keep polling in a loop with a short sleep after errors, and resubscribe on each ConnAck.

Does rumqttc support MQTT 5?

Yes, through the rumqttc::v5 module, which provides MQTT 5 properties, reason codes and session expiry.

Why does my rumqttc program freeze?

Usually because it awaits subscribe or publish inside the task that polls the event loop. Use try_subscribe and try_publish there instead.