MQTT in Rust mit rumqttc: AsyncClient, Event Loop und TLS

rumqttc ist der meistgenutzte MQTT-Client in reinem Rust. Er teilt die Arbeit in zwei Teile: einen AsyncClient, den du klonst und zum Publishen und Abonnieren nutzt, und einen EventLoop, den du pollst, um die Netzwerkverbindung voranzutreiben. Hat man dieses Design verstanden, ist der Rest einfach. Dieses Tutorial baut einen Tokio-Client, der publisht und abonniert, Netzabbrüche übersteht und sich per TLS mit rustls verbindet.

Client und Event Loop

AsyncClient::new(options, cap) liefert ein Paar:

Die Folge: Pollt niemand den Event Loop, wird nichts gesendet, und ist der Channel voll, wartet publish().await, bis die Schleife aufholt. Das ist gewollt — es liefert Backpressure —, überrascht aber alle, die erwarten, dass publish von selbst ins Netz geht.

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 aktiviert rustls standardmäßig (Feature use-rustls), TLS funktioniert also ohne OpenSSL. Die Beispiele nutzen die API der 0.24-Linie; spätere Releases behalten dieselbe Struktur.

Publish und Subscribe

Dieses Programm publisht aus einem Hintergrund-Task jede Sekunde eine Nachricht und gibt alles aus, was auf demselben Topic ankommt.

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: ein Klon des Clients in einem eigenen 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 fehlgeschlagen: {e}");
            }
            time::sleep(Duration::from_secs(1)).await;
        }
    });

    // Event Loop: treibt die Verbindung und liefert eingehende Pakete.
    loop {
        match eventloop.poll().await {
            Ok(Event::Incoming(Packet::ConnAck(ack))) => {
                println!("verbunden, session present: {}", ack.session_present);
                // try_subscribe wartet nicht und kann diese Schleife daher nicht blockieren.
                if let Err(e) = client.try_subscribe(TOPIC, QoS::AtLeastOnce) {
                    eprintln!("Subscribe fehlgeschlagen: {e}");
                }
            }
            Ok(Event::Incoming(Packet::Publish(p))) => {
                println!("empfangen {}: {}", p.topic, String::from_utf8_lossy(&p.payload));
            }
            Ok(_) => {}
            Err(e) => {
                eprintln!("Verbindungsfehler: {e}; neuer Versuch in 2 s");
                time::sleep(Duration::from_secs(2)).await;
            }
        }
    }
}

Starte es mit cargo run. Um zu prüfen, dass die Nachrichten wirklich beim Broker ankommen, öffne den Online-MQTT-Client, verbinde dich per Secure WebSocket mit broker.connectmq.com:8084 und abonniere testmqtt/rust-rumqttc/#; publishe aus dem Browser, und das Rust-Programm gibt deine Nachricht aus.

Drei Entscheidungen in diesem Code sind bewusst getroffen:

• Bei jedem ConnAck abonnieren. Standard ist eine Clean Session, der Broker vergisst die Subscriptions also beim Verbindungsabbruch. Wer bei jedem ConnAck abonniert, stellt sie nach jedem Reconnect wieder her.
• try_subscribe in der Schleife. client.subscribe(...).await aus demselben Task, der den Event Loop pollt, kann bei vollem Channel einen Deadlock auslösen, weil nur die Schleife den Channel leert. Die nicht-asynchronen try_*-Methoden schlagen stattdessen sofort fehl.
• Nach Fehlern warten. Das nächste poll() nach einem Fehler verbindet sofort neu. Ohne Pause wird ein Broker-Ausfall zu einer engen Schleife aus Verbindungsversuchen.

Zugangsdaten und TLS mit rustls

Bei einem echten Broker willst du TLS auf Port 8883 sowie Benutzername und Passwort:

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() erzeugt eine rustls-Konfiguration, die den nativen Root-Zertifikaten der Plattform vertraut — genau richtig für Broker mit öffentlich vertrauenswürdigen Zertifikaten. Für eine private CA übergibst du die PEM-Datei selbst:

use rumqttc::{TlsConfiguration, Transport};

let ca = std::fs::read("ca.pem").expect("CA-Datei lesen");
opts.set_transport(Transport::Tls(TlsConfiguration::Simple {
    ca,
    alpn: None,
    client_auth: None, // Some((cert_pem_bytes, key_pem_bytes)) für Mutual TLS
}));

Brauchst du volle Kontrolle (eigene Verifier, bestimmte Cipher Suites), nimmt TlsConfiguration::Rustls(Arc<rustls::ClientConfig>) eine fertige rustls-Konfiguration an. Deine rustls-Abhängigkeit muss dieselbe Major-Version haben wie die von rumqttc — prüfe das mit cargo tree -i rustls —, sonst passen die Typen nicht.

Der Hostname, den du an MqttOptions::new übergibst, ist der Name, den rustls gegen das Zertifikat prüft. Eine Verbindung per IP-Adresse zu einem Zertifikat, das für einen DNS-Namen ausgestellt ist, scheitert an der Verifikation; siehe Hostname Mismatch und den TLS-Verbindungstest, um das vom Broker präsentierte Zertifikat zu prüfen.

WebSockets und MQTT 5

Mit aktiviertem Feature websocket (cargo add rumqttc --features websocket) tunnelt rumqttc MQTT über WebSockets — praktisch hinter Proxys, die nur HTTPS durchlassen. Übergib die vollständige URL als Host und wähle den Transport:

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

Für MQTT 5 importierst du aus rumqttc::v5 statt aus der Crate-Wurzel. Client und Event Loop funktionieren genauso; Pakete tragen Properties (User Properties, Message Expiry, Response Topic), und Acks liefern MQTT-5-Reason-Codes, die du in der Reason-Code-Referenz nachschlagen kannst.

Fehler, die dir begegnen

eventloop.poll() liefert einen ConnectionError. Die Variante zeigt meist direkt auf die Ursache:

Der letzte Punkt erwischt viele: rumqttc hat eine konservative maximale Paketgröße (10 KiB in aktuellen Versionen), größere JSON-Payloads oder Bilder brauchen also ein explizites Limit. Mit dem Paketgrößen-Rechner siehst du, wie groß eine Nachricht auf der Leitung wirklich ist.

Den Client testen

Eine kurze Checkliste, bevor du einem rumqttc-Client vertraust:

1. Roundtrip. Aus dem Online-Client publishen und prüfen, dass das Rust-Programm die Nachricht ausgibt; dann umgekehrt.
2. Netzabbruch. WLAN länger als das Keep-Alive abschalten, wieder einschalten und prüfen, dass eine neue "verbunden"-Zeile erscheint und Nachrichten wieder fließen — der Beweis, dass Neu-Abonnieren beim ConnAck funktioniert.
3. Falsche Zugangsdaten. Einmal mit falschem Passwort starten und prüfen, dass dein Code die Ablehnung loggt, statt still im Kreis zu laufen.
4. Große Payload. Eine Nachricht über 10 KiB publishen und prüfen, ob die Paketgrößen-Einstellung passt.

Für einen breiteren Blick siehe MQTT-Reconnect Best Practices und den Bibliotheksvergleich unter die besten MQTT-Bibliotheken; wenn du lieber Bindings zur Eclipse-Paho-C-Bibliothek nutzt, behandelt MQTT in Rust mit paho-mqtt die Alternative.

Topic im Online-Client öffnen

rumqttc ist klein, von Grund auf async und reines Rust: Ein Task pollt den Event Loop, bei jedem ConnAck wird abonniert, nach Fehlern wird gewartet, und TLS läuft über Transport. Mit diesen vier Gewohnheiten läuft derselbe Client vom Laptop-Test bis zur Produktionsflotte.

Häufig gestellte Fragen