MQTT en Rust avec rumqttc : AsyncClient, event loop et TLS

rumqttc est le client MQTT en pur Rust le plus utilisé. Il découpe le travail en deux : un AsyncClient que vous clonez pour publier et vous abonner, et un EventLoop que vous interrogez (poll) pour faire avancer la connexion réseau. Une fois ce modèle compris, le reste est simple. Ce tutoriel construit un client Tokio qui publie et s'abonne, survit aux coupures réseau et se connecte en TLS avec rustls.

Client et event loop

AsyncClient::new(options, cap) renvoie une paire :

Conséquence : si personne ne poll l'event loop, rien n'est envoyé, et une fois le canal plein, publish().await attend que la boucle rattrape son retard. C'est voulu — cela fournit du backpressure — mais cela surprend ceux qui s'attendent à ce que publish parte seul sur le réseau.

Mise en place

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 active rustls par défaut (feature use-rustls), donc TLS fonctionne sans OpenSSL. Les exemples utilisent l'API de la branche 0.24 ; les versions suivantes gardent la même structure.

Publish et subscribe

Ce programme publie un message par seconde depuis une tâche en arrière-plan et affiche tout ce qu'il reçoit sur le même 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 : un clone du client dans sa propre tâche.
    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!("échec du publish : {e}");
            }
            time::sleep(Duration::from_secs(1)).await;
        }
    });

    // Event loop : fait avancer la connexion et livre les paquets entrants.
    loop {
        match eventloop.poll().await {
            Ok(Event::Incoming(Packet::ConnAck(ack))) => {
                println!("connecté, session present : {}", ack.session_present);
                // try_subscribe ne fait pas d'await et ne peut donc pas bloquer cette boucle.
                if let Err(e) = client.try_subscribe(TOPIC, QoS::AtLeastOnce) {
                    eprintln!("échec du subscribe : {e}");
                }
            }
            Ok(Event::Incoming(Packet::Publish(p))) => {
                println!("reçu {} : {}", p.topic, String::from_utf8_lossy(&p.payload));
            }
            Ok(_) => {}
            Err(e) => {
                eprintln!("erreur de connexion : {e} ; nouvel essai dans 2 s");
                time::sleep(Duration::from_secs(2)).await;
            }
        }
    }
}

Lancez-le avec cargo run. Pour vérifier que les messages atteignent bien le broker, ouvrez le client MQTT en ligne, connectez-vous à broker.connectmq.com:8084 en WebSocket sécurisé et abonnez-vous à testmqtt/rust-rumqttc/# ; publiez depuis le navigateur et le programme Rust affichera votre message.

Trois choix de ce code sont délibérés :

• S'abonner à chaque ConnAck. Par défaut la session est clean : le broker oublie les abonnements quand la connexion tombe. S'abonner à chaque ConnAck les restaure après chaque reconnexion.
• try_subscribe dans la boucle. Appeler client.subscribe(...).await depuis la tâche qui poll l'event loop peut provoquer un interblocage quand le canal est plein, car la boucle est la seule à le vider. Les méthodes try_*, non async, échouent immédiatement à la place.
• Pause après les erreurs. Le poll() suivant une erreur se reconnecte tout de suite. Sans délai, une panne du broker devient une boucle serrée de tentatives de connexion.

Identifiants et TLS avec rustls

Avec un vrai broker, vous voudrez TLS sur le port 8883 ainsi qu'un nom d'utilisateur et un mot de passe :

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() construit une configuration rustls qui fait confiance aux certificats racine natifs de la plateforme, ce qu'il faut pour les brokers dotés de certificats publiquement reconnus. Pour une CA privée, fournissez vous-même le fichier PEM :

use rumqttc::{TlsConfiguration, Transport};

let ca = std::fs::read("ca.pem").expect("lire le fichier de la CA");
opts.set_transport(Transport::Tls(TlsConfiguration::Simple {
    ca,
    alpn: None,
    client_auth: None, // Some((cert_pem_bytes, key_pem_bytes)) pour le TLS mutuel
}));

Si vous avez besoin d'un contrôle total (vérificateurs personnalisés, suites de chiffrement précises), TlsConfiguration::Rustls(Arc<rustls::ClientConfig>) accepte une configuration rustls préparée. Votre dépendance rustls doit avoir la même version majeure que celle de rumqttc — vérifiez avec cargo tree -i rustls — sinon les types ne correspondront pas.

Le nom d'hôte passé à MqttOptions::new est celui que rustls vérifie dans le certificat. Se connecter par adresse IP à un certificat émis pour un nom DNS fait échouer la vérification ; voir hostname mismatch et le test de connexion TLS pour inspecter le certificat présenté par le broker.

WebSockets et MQTT 5

Avec la feature websocket activée (cargo add rumqttc --features websocket), rumqttc peut faire passer MQTT par WebSockets — pratique derrière des proxys qui n'autorisent que HTTPS. Passez l'URL complète comme hôte et choisissez le transport :

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

Pour MQTT 5, importez depuis rumqttc::v5 au lieu de la racine du crate. Le client et l'event loop fonctionnent de la même façon ; les paquets portent des properties (user properties, message expiry, response topic) et les acquittements contiennent des reason codes MQTT 5 que vous pouvez consulter dans la référence des reason codes.

Les erreurs que vous rencontrerez

eventloop.poll() renvoie un ConnectionError. La variante désigne généralement la cause directement :

La dernière piège beaucoup de monde : rumqttc a une taille maximale de paquet prudente (10 Kio dans les versions récentes), donc les gros payloads JSON ou les images exigent une limite explicite. Utilisez le calculateur de taille de paquet pour voir la taille réelle d'un message sur le réseau.

Tester le client

Une checklist rapide avant de faire confiance à un client rumqttc :

1. Aller-retour. Publiez depuis le client en ligne et vérifiez que le programme Rust affiche le message ; puis dans l'autre sens.
2. Coupure réseau. Désactivez le Wi-Fi plus longtemps que le keep-alive, réactivez-le et vérifiez qu'une nouvelle ligne "connecté" apparaît et que les messages reprennent — la preuve que le réabonnement sur ConnAck fonctionne.
3. Mauvais identifiants. Lancez une fois avec un mauvais mot de passe et vérifiez que votre code journalise le refus au lieu de tourner en silence.
4. Gros payload. Publiez un message de plus de 10 Kio et vérifiez que votre réglage de taille de paquet est correct.

Pour une approche plus large, consultez les bonnes pratiques de reconnexion MQTT et le comparatif des bibliothèques dans les meilleures bibliothèques MQTT ; si vous préférez des bindings vers la bibliothèque Eclipse Paho C, MQTT en Rust avec paho-mqtt présente l'alternative.

Ouvrir le topic dans le client en ligne

rumqttc est léger, async dès la conception et en pur Rust : gardez une tâche qui poll l'event loop, réabonnez-vous à chaque ConnAck, temporisez après les erreurs et configurez TLS via Transport. Avec ces quatre habitudes, le même client passe du test sur portable à une flotte en production.

Questions fréquentes