MQTT in Rust with paho-mqtt: Async Client and MQTT 5

The paho-mqtt crate is the official Eclipse Paho client for Rust. It is a safe wrapper around the Paho C library, which means you get the same battle-tested protocol engine used by many C and embedded projects, with an idiomatic async API on top. It supports MQTT 3.1.1 and MQTT 5, TCP, TLS and WebSockets. This guide covers building it, a complete async MQTT 5 client, reconnection, TLS and the errors you are most likely to hit.

If you would rather avoid a C dependency, rumqttc is the pure-Rust alternative.

Build requirements

By default the crate builds a bundled copy of Paho C with CMake and links it statically, with OpenSSL for TLS. Install the toolchain first:

# Debian / Ubuntu
sudo apt install build-essential cmake libssl-dev

# macOS
brew install cmake openssl@3

[dependencies]
paho-mqtt = "0.13"
futures = "0.3"
tokio = { version = "1", features = ["full"] }

Check crates.io for the newest release; the API shown here has been stable across recent versions. The first build takes a little longer because the C library is compiled; after that it is cached like any other crate.

An async MQTT 5 client

The pattern is: create the client, get a message stream before connecting, connect, subscribe, then read the stream. A None on the stream means the connection was lost.

use futures::StreamExt;
use paho_mqtt as mqtt;
use std::time::Duration;

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

#[tokio::main]
async fn main() -> Result<(), mqtt::Error> {
    let create_opts = mqtt::CreateOptionsBuilder::new()
        .server_uri("tcp://broker.connectmq.com:1883")
        .client_id("testmqtt-rust-paho-demo")
        .mqtt_version(mqtt::MQTT_VERSION_5)
        .finalize();
    let mut cli = mqtt::AsyncClient::new(create_opts)?;

    // Get the stream before connecting so no message is missed.
    let mut stream = cli.get_stream(25);

    let conn_opts = mqtt::ConnectOptionsBuilder::new_v5()
        .keep_alive_interval(Duration::from_secs(30))
        .clean_start(true)
        .finalize();

    let rsp = cli.connect(conn_opts).await?;
    if let Some(conn) = rsp.connect_response() {
        println!("connected to {} with MQTT v{}", conn.server_uri, conn.mqtt_version);
    }
    cli.subscribe(TOPIC, mqtt::QOS_1).await?;

    // Publish five messages from a separate task.
    let publisher = cli.clone();
    tokio::spawn(async move {
        for i in 0..5 {
            let msg = mqtt::Message::new(TOPIC, format!("message {i}"), mqtt::QOS_1);
            if let Err(e) = publisher.publish(msg).await {
                eprintln!("publish failed: {e}");
            }
            tokio::time::sleep(Duration::from_secs(1)).await;
        }
    });

    while let Some(msg_opt) = stream.next().await {
        match msg_opt {
            Some(msg) => println!("received {}: {}", msg.topic(), msg.payload_str()),
            None => {
                eprintln!("connection lost, reconnecting");
                while let Err(e) = cli.reconnect().await {
                    eprintln!("reconnect failed: {e}");
                    tokio::time::sleep(Duration::from_secs(2)).await;
                }
                cli.subscribe(TOPIC, mqtt::QOS_1).await?;
            }
        }
    }
    Ok(())
}

Run it with cargo run. You should see the connect line with MQTT v5, then the five messages coming back. Open the online MQTT client, subscribe to testmqtt/rust-paho/# and you will see the same messages; publish one from the browser and the Rust program prints it too.

Key points:

• mqtt_version(MQTT_VERSION_5) on the create options and new_v5() on the connect options — both are needed for an MQTT 5 connection.
• The futures are runtime-agnostic. Paho C runs its own network threads; you can await the tokens from Tokio, async-std or a plain block_on.
• publish with QoS 1 resolves when the PUBACK arrives, so an error there means the broker did not accept the message.
• Resubscribe after reconnect() when you use a clean start, because the broker dropped the old session.

Automatic reconnect and persistent sessions

Instead of the manual loop, you can let Paho reconnect and use a callback to restore state:

let conn_opts = mqtt::ConnectOptionsBuilder::new_v5()
    .keep_alive_interval(Duration::from_secs(30))
    .clean_start(false)
    .properties(mqtt::properties![mqtt::PropertyCode::SessionExpiryInterval => 3600])
    .automatic_reconnect(Duration::from_secs(1), Duration::from_secs(30))
    .finalize();

cli.set_connected_callback(|_cli| println!("(re)connected"));
cli.set_connection_lost_callback(|_cli| eprintln!("connection lost"));

With clean_start(false) and a session expiry interval, the broker keeps your subscriptions and queues QoS 1 messages while you are offline, so you do not need to resubscribe. Note that the shared test endpoint does not keep offline sessions; try this against your own private broker instance. Persistent sessions explains what the broker stores and for how long.

Paho can also persist its own in-flight QoS 1/2 messages on the client side. Pick the behaviour explicitly with .persistence(...) on the create options: mqtt::PersistenceType::None keeps everything in memory, mqtt::PersistenceType::File writes it to disk so it survives a process restart. If you notice directories named after your client ID appearing in the working directory, that is file persistence at work.

TLS, credentials and WebSockets

Switch the URI to ssl:// (or mqtts:// with recent Paho C builds) on port 8883 and add SSL options:

use std::env;

let create_opts = mqtt::CreateOptionsBuilder::new()
    .server_uri("ssl://broker.connectmq.com:8883")
    .client_id("testmqtt-rust-paho-tls")
    .mqtt_version(mqtt::MQTT_VERSION_5)
    .finalize();

let ssl_opts = mqtt::SslOptionsBuilder::new()
    // For a private CA: .trust_store("ca.pem")?
    .finalize();

let conn_opts = mqtt::ConnectOptionsBuilder::new_v5()
    .ssl_options(ssl_opts)
    .user_name(env::var("MQTT_USER").unwrap_or_default())
    .password(env::var("MQTT_PASS").unwrap_or_default())
    .finalize();

Without a trust store, OpenSSL's default certificate locations are used. If verification fails on a broker with a public certificate, point trust_store at your system bundle (for example /etc/ssl/certs/ca-certificates.crt on Debian). For client certificates use key_store and private_key. The TLS connection test shows how to inspect the chain with openssl s_client before blaming the code.

For networks that only allow HTTPS, use wss://broker.connectmq.com:8084/mqtt as the server URI with the same SSL options.

Errors you will see

Look up any MQTT 5 code in the reason code reference. For a structured way to exercise session expiry, user properties and topic aliases against your broker, see testing MQTT 5 features.

Testing checklist

1. Round trip: publish from the online client, receive in Rust, and the other way round.
2. Pull the network cable or disable Wi-Fi for longer than the keep-alive; confirm the reconnect path runs and messages resume.
3. Restart your program with clean_start(false) against a private broker and confirm messages published while it was down arrive afterwards.
4. Run once with wrong credentials and confirm the error is logged clearly.

Watch the demo topic in your browser

paho-mqtt gives Rust programs the Eclipse Paho engine with MQTT 5, TLS and WebSockets, at the cost of a C build step. Get the stream before connecting, decide how you restore subscriptions, and test a disconnect once — then the client behaves the same way from your laptop to production. If you also work in C, the Paho C tutorial shows the library underneath.

Frequently asked questions