MQTT in Rust with rumqttc: AsyncClient, Event Loop and TLS

rumqttc is the most widely used pure-Rust MQTT client. It splits the work in two: an AsyncClient you clone and use to publish and subscribe, and an EventLoop you poll to drive the network connection. Once that design clicks, the rest is straightforward. This tutorial builds a Tokio client that publishes and subscribes, recovers from network drops, and connects over TLS with rustls.

Client and event loop

AsyncClient::new(options, cap) returns a pair:

The consequence: if nobody polls the event loop, nothing is sent, and once the channel is full, publish().await waits until the loop catches up. That is a feature — it gives you backpressure — but it surprises people who expect publish to hit the network by itself.

Setup

cargo new rumqttc-demo && cd rumqttc-demo
cargo add rumqttc
cargo add tokio --features full

[dependencies]
rumqttc = "0.24"
tokio = { version = "1", features = ["full"] }

rumqttc enables rustls by default (use-rustls feature), so TLS works without OpenSSL. The examples use the API of the 0.24 line; later releases keep the same structure.

Publish and subscribe

This program publishes a message every second from a background task and prints everything it receives on the same topic.

use rumqttc::{AsyncClient, Event, MqttOptions, Packet, QoS};
use std::time::Duration;
use tokio::time;

const TOPIC: &str = "testmqtt/rust-rumqttc/demo";

#[tokio::main]
async fn main() {
    let mut opts = MqttOptions::new("testmqtt-rumqttc-demo", "broker.connectmq.com", 1883);
    opts.set_keep_alive(Duration::from_secs(30));

    let (client, mut eventloop) = AsyncClient::new(opts, 10);

    // Publisher: a clone of the client in its own task.
    let publisher = client.clone();
    tokio::spawn(async move {
        for i in 0u32.. {
            let payload = format!("message {i}");
            if let Err(e) = publisher.publish(TOPIC, QoS::AtLeastOnce, false, payload).await {
                eprintln!("publish failed: {e}");
            }
            time::sleep(Duration::from_secs(1)).await;
        }
    });

    // Event loop: drives the connection and delivers incoming packets.
    loop {
        match eventloop.poll().await {
            Ok(Event::Incoming(Packet::ConnAck(ack))) => {
                println!("connected, session present: {}", ack.session_present);
                // try_subscribe does not await, so it cannot block this loop.
                if let Err(e) = client.try_subscribe(TOPIC, QoS::AtLeastOnce) {
                    eprintln!("subscribe failed: {e}");
                }
            }
            Ok(Event::Incoming(Packet::Publish(p))) => {
                println!("received {}: {}", p.topic, String::from_utf8_lossy(&p.payload));
            }
            Ok(_) => {}
            Err(e) => {
                eprintln!("connection error: {e}; retrying in 2 s");
                time::sleep(Duration::from_secs(2)).await;
            }
        }
    }
}

Run it with cargo run. To confirm the messages really reach the broker, open the online MQTT client, connect to broker.connectmq.com:8084 over secure WebSocket and subscribe to testmqtt/rust-rumqttc/#; publish from the browser and the Rust program prints your message.

Three choices in this code are deliberate:

• Subscribe on every ConnAck. The default is a clean session, so the broker forgets subscriptions when the connection drops. Subscribing whenever a ConnAck arrives restores them after each reconnect.
• try_subscribe inside the loop. Calling client.subscribe(...).await from the same task that polls the event loop can deadlock when the channel is full, because the loop is the only thing that drains it. The non-async try_* methods fail fast instead.
• Sleep after errors. The next poll() after an error reconnects immediately. Without a delay, a broker outage turns into a tight loop of connection attempts.

Credentials and TLS with rustls

For a real broker you will want TLS on port 8883 and a username and password:

use rumqttc::{MqttOptions, Transport};
use std::time::Duration;

let mut opts = MqttOptions::new("testmqtt-rumqttc-tls", "broker.connectmq.com", 8883);
opts.set_keep_alive(Duration::from_secs(30));
opts.set_credentials(
    std::env::var("MQTT_USER").unwrap_or_default(),
    std::env::var("MQTT_PASS").unwrap_or_default(),
);
opts.set_transport(Transport::tls_with_default_config());

tls_with_default_config() builds a rustls configuration that trusts the platform's native root certificates, which is what you want for brokers with publicly trusted certificates. For a private CA, pass the PEM file yourself:

use rumqttc::{TlsConfiguration, Transport};

let ca = std::fs::read("ca.pem").expect("read CA file");
opts.set_transport(Transport::Tls(TlsConfiguration::Simple {
    ca,
    alpn: None,
    client_auth: None, // Some((cert_pem_bytes, key_pem_bytes)) for mutual TLS
}));

If you need full control (custom verifiers, specific cipher suites), TlsConfiguration::Rustls(Arc<rustls::ClientConfig>) accepts a prepared rustls config. Your rustls dependency must be the same major version rumqttc uses — check with cargo tree -i rustls — or the types will not match.

The host name you pass to MqttOptions::new is the name rustls verifies against the certificate. Connecting by IP address to a certificate issued for a DNS name fails verification; see hostname mismatch and the TLS connection test for how to inspect the certificate the broker presents.

WebSockets and MQTT 5

With the websocket feature enabled (cargo add rumqttc --features websocket), rumqttc can tunnel MQTT through WebSockets — useful behind proxies that only allow HTTPS. Pass the full URL as the host and select the transport:

let mut opts = MqttOptions::new(
    "testmqtt-rumqttc-wss",
    "wss://broker.connectmq.com:8084/mqtt",
    8084,
);
opts.set_transport(Transport::wss_with_default_config());

For MQTT 5, import from rumqttc::v5 instead of the crate root. The client and event loop work the same way; packets carry properties (user properties, message expiry, response topic) and acknowledgements carry MQTT 5 reason codes you can look up in the reason code reference.

Errors you will meet

eventloop.poll() returns a ConnectionError. The variant usually points straight at the cause:

The last one catches many people: rumqttc ships with a conservative maximum packet size (10 KiB in recent versions), so larger JSON payloads or images need an explicit limit. Use the packet size calculator to see how big a message really is on the wire.

Testing the client

A quick checklist before you trust a rumqttc client:

1. Round trip. Publish from the online client and confirm the Rust program prints the message; then the other way round.
2. Network drop. Disable Wi-Fi for longer than the keep-alive, re-enable it, and check that you see a new "connected" line and messages resume — proof that resubscribe-on-ConnAck works.
3. Bad credentials. Run once with a wrong password and confirm your code logs the refusal instead of spinning silently.
4. Large payload. Publish a message above 10 KiB and confirm your packet size setting is right.

For a broader approach, see MQTT reconnect best practices and the comparison of client libraries in best MQTT libraries; if you prefer bindings to the Eclipse Paho C library, MQTT in Rust with paho-mqtt covers the alternative.

Open the topic in the online client

rumqttc is small, async-first and pure Rust: keep one task polling the event loop, subscribe on every ConnAck, back off after errors and set TLS through Transport. With those four habits the same client runs from a laptop test to a production fleet.

Frequently asked questions