MQTT in Rust mit paho-mqtt: Async-Client und MQTT 5

Das Crate paho-mqtt ist der offizielle Eclipse-Paho-Client für Rust. Es ist ein sicherer Wrapper um die Paho-C-Bibliothek: Du bekommst dieselbe praxiserprobte Protokoll-Engine, die viele C- und Embedded-Projekte nutzen, mit einer idiomatischen Async-API obendrauf. Unterstützt werden MQTT 3.1.1 und MQTT 5, TCP, TLS und WebSockets. Dieser Guide behandelt den Build, einen vollständigen Async-MQTT-5-Client, Reconnect, TLS und die Fehler, die dir am wahrscheinlichsten begegnen.

Wenn du eine C-Abhängigkeit lieber vermeidest, ist rumqttc die Alternative in reinem Rust.

Build-Voraussetzungen

Standardmäßig baut das Crate eine mitgelieferte Kopie von Paho C mit CMake und linkt sie statisch, mit OpenSSL für TLS. Installiere zuerst die Toolchain:

# 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"] }

Die neueste Version findest du auf crates.io; die hier gezeigte API ist über die letzten Versionen stabil geblieben. Der erste Build dauert etwas länger, weil die C-Bibliothek kompiliert wird; danach ist sie wie jedes andere Crate gecacht.

Ein Async-MQTT-5-Client

Das Muster: Client erzeugen, den Nachrichten-Stream vor dem Verbinden holen, verbinden, abonnieren, dann den Stream lesen. Ein None im Stream bedeutet, dass die Verbindung verloren ging.

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

    // Stream vor dem Verbinden holen, damit keine Nachricht verloren geht.
    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!("verbunden mit {} über MQTT v{}", conn.server_uri, conn.mqtt_version);
    }
    cli.subscribe(TOPIC, mqtt::QOS_1).await?;

    // Fünf Nachrichten aus einem separaten Task publishen.
    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 fehlgeschlagen: {e}");
            }
            tokio::time::sleep(Duration::from_secs(1)).await;
        }
    });

    while let Some(msg_opt) = stream.next().await {
        match msg_opt {
            Some(msg) => println!("empfangen {}: {}", msg.topic(), msg.payload_str()),
            None => {
                eprintln!("Verbindung verloren, verbinde neu");
                while let Err(e) = cli.reconnect().await {
                    eprintln!("Reconnect fehlgeschlagen: {e}");
                    tokio::time::sleep(Duration::from_secs(2)).await;
                }
                cli.subscribe(TOPIC, mqtt::QOS_1).await?;
            }
        }
    }
    Ok(())
}

Starte es mit cargo run. Du solltest die Verbindungszeile mit MQTT v5 sehen und danach die fünf zurückkommenden Nachrichten. Öffne den Online-MQTT-Client, abonniere testmqtt/rust-paho/#, und du siehst dieselben Nachrichten; publishe eine aus dem Browser, und das Rust-Programm gibt sie ebenfalls aus.

Die wichtigsten Punkte:

• mqtt_version(MQTT_VERSION_5) in den Create-Optionen und new_v5() in den Connect-Optionen — beides ist für eine MQTT-5-Verbindung nötig.
• Die Futures sind laufzeitunabhängig. Paho C betreibt eigene Netzwerk-Threads; du kannst die Tokens aus Tokio, async-std oder einem einfachen block_on awaiten.
• publish mit QoS 1 wird aufgelöst, sobald das PUBACK eintrifft — ein Fehler dort heißt, der Broker hat die Nachricht nicht angenommen.
• Nach reconnect() neu abonnieren, wenn du Clean Start nutzt, denn der Broker hat die alte Session verworfen.

Automatisches Reconnect und persistente Sessions

Statt der manuellen Schleife kannst du Paho das Reconnect überlassen und den Zustand per Callback wiederherstellen:

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!("(neu) verbunden"));
cli.set_connection_lost_callback(|_cli| eprintln!("Verbindung verloren"));

Mit clean_start(false) und einem Session-Expiry-Intervall behält der Broker deine Subscriptions und puffert QoS-1-Nachrichten, solange du offline bist — neu abonnieren musst du dann nicht. Beachte, dass der geteilte Test-Endpoint keine Offline-Sessions speichert; teste das gegen deine eigene private Broker-Instanz. Persistente Sessions erklärt, was der Broker speichert und wie lange.

Paho kann außerdem clientseitig seine eigenen QoS-1/2-Nachrichten im Flug persistieren. Lege das Verhalten mit .persistence(...) in den Create-Optionen explizit fest: mqtt::PersistenceType::None hält alles im Speicher, mqtt::PersistenceType::File schreibt es auf die Platte, sodass es einen Prozessneustart übersteht. Tauchen im Arbeitsverzeichnis Ordner mit dem Namen deiner Client-ID auf, ist das die Datei-Persistenz.

TLS, Zugangsdaten und WebSockets

Stelle die URI auf ssl:// (oder mqtts:// bei aktuellen Paho-C-Builds) mit Port 8883 um und ergänze SSL-Optionen:

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()
    // Für eine 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();

Ohne Trust Store werden die Standard-Zertifikatspfade von OpenSSL verwendet. Scheitert die Verifikation bei einem Broker mit öffentlichem Zertifikat, zeige trust_store auf das System-Bundle (zum Beispiel /etc/ssl/certs/ca-certificates.crt unter Debian). Für Client-Zertifikate nutzt du key_store und private_key. Der TLS-Verbindungstest zeigt, wie du die Kette mit openssl s_client prüfst, bevor du den Code verdächtigst.

In Netzen, die nur HTTPS erlauben, nimm wss://broker.connectmq.com:8084/mqtt als Server-URI mit denselben SSL-Optionen.

Fehler, die du sehen wirst

Jeden MQTT-5-Code findest du in der Reason-Code-Referenz. Wie du Session Expiry, User Properties und Topic Aliases strukturiert gegen deinen Broker testest, zeigt MQTT-5-Features testen.

Test-Checkliste

1. Roundtrip: aus dem Online-Client publishen, in Rust empfangen, und umgekehrt.
2. Netzwerkkabel ziehen oder WLAN länger als das Keep-Alive abschalten; prüfen, dass der Reconnect-Pfad läuft und Nachrichten wieder fließen.
3. Programm mit clean_start(false) gegen einen privaten Broker neu starten und prüfen, dass Nachrichten, die während der Ausfallzeit publiziert wurden, danach ankommen.
4. Einmal mit falschen Zugangsdaten starten und prüfen, dass der Fehler klar geloggt wird.

Demo-Topic im Browser beobachten

paho-mqtt bringt Rust-Programmen die Eclipse-Paho-Engine mit MQTT 5, TLS und WebSockets — der Preis ist ein C-Build-Schritt. Hol den Stream vor dem Verbinden, entscheide, wie du Subscriptions wiederherstellst, und teste einmal einen Verbindungsabbruch — dann verhält sich der Client vom Laptop bis zur Produktion gleich. Wenn du auch in C arbeitest, zeigt das Paho-C-Tutorial die Bibliothek darunter.

Häufig gestellte Fragen