Rust MQTT avec rumqttc : publish/subscribe asynchrone en TLS avec Tokio

Ce tutoriel Rust MQTT rumqttc montre comment rumqttc vous offre un client asynchrone 100 % Rust qui s'intègre naturellement aux services Tokio, des passerelles edge aux workers d'ingestion backend. Vous allez construire un client rumqttc complet qui se connecte en TLS avec rustls sur le port 8883, publie de la télémétrie JSON, reçoit des commandes et continue de fonctionner malgré les coupures réseau.

Pourquoi rumqttc pour Rust MQTT

rumqttc est la partie client du projet rumqtt. Il est entièrement écrit en Rust, prend en charge MQTT 3.1.1 à la racine du crate et MQTT 5 dans son module v5, et propose à la fois un Client synchrone et un AsyncClient construit sur Tokio. TLS est fourni par rustls par défaut ; vous n'avez donc pas besoin d'OpenSSL sur la machine cible, ce qui est pratique pour la compilation croisée vers des passerelles ARM.

Sa conception diffère de celle des bibliothèques à callbacks d'autres langages, et mieux vaut la comprendre avant d'écrire du code :

Rien ne se passe sur le réseau tant que quelque chose n'interroge pas l'event loop. Les requêtes d'AsyncClient vont dans un channel borné ; l'event loop le vide, envoie les paquets, lit les paquets entrants et gère les pings. Si poll() renvoie une erreur, l'appel suivant se reconnecte. Cette boucle unique est l'endroit où vit votre logique de connexion.

Préparer le projet

Créez un nouveau crate binaire :

cargo new rust-mqtt-demo
cd rust-mqtt-demo

Modifiez Cargo.toml :

[package]
name = "rust-mqtt-demo"
version = "0.1.0"
edition = "2021"

[dependencies]
rumqttc = "0.25"
tokio = { version = "1", features = ["full"] }
serde = { version = "1", features = ["derive"] }
serde_json = "1"

La feature par défaut use-rustls active TLS via rustls avec le fournisseur cryptographique aws-lc-rs. Compiler ce fournisseur nécessite un compilateur C et CMake sur la machine de build ; sous Windows, NASM peut aussi être requis. Les machines d'exécution n'ont besoin de rien de plus.

Pour le premier essai, utilisez le serveur MQTT public gratuit broker.connectmq.com, qui accepte les clients anonymes avec n'importe quel client ID unique et non vide. Choisissez un préfixe aléatoire comme connectmq/demo/m3c8wd ; l'espace public est lisible par tous, ne publiez donc que des données de test.

L'exemple rumqttc complet

Remplacez src/main.rs :

use rumqttc::{AsyncClient, Event, LastWill, MqttOptions, Packet, QoS, Transport};
use serde::Serialize;
use std::env;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::time;

#[derive(Serialize)]
struct Reading {
    device: String,
    temperature: f64,
    ts: u64,
}

fn var(key: &str, default: &str) -> String {
    env::var(key).unwrap_or_else(|_| default.to_string())
}

#[tokio::main]
async fn main() {
    let host = var("MQTT_HOST", "broker.connectmq.com");
    let client_id = var("MQTT_CLIENT_ID", "rust-demo-m3c8wd");
    let prefix = var("MQTT_PREFIX", "connectmq/demo/m3c8wd");
    let status_topic = format!("{prefix}/status");
    let command_filter = format!("{prefix}/cmd/#");
    let telemetry_topic = format!("{prefix}/telemetry");

    let mut options = MqttOptions::new(client_id.clone(), host, 8883);
    options.set_keep_alive(Duration::from_secs(30));
    options.set_clean_session(false);
    options.set_last_will(LastWill::new(&status_topic, "offline", QoS::AtLeastOnce, true));
    options.set_transport(Transport::tls_with_default_config());
    if let Ok(user) = env::var("MQTT_USERNAME") {
        options.set_credentials(user, env::var("MQTT_PASSWORD").unwrap_or_default());
    }

    let (client, mut eventloop) = AsyncClient::new(options, 32);

    // Tâche de publication : envoie une mesure toutes les 5 secondes.
    let publisher = client.clone();
    let device = client_id.clone();
    tokio::spawn(async move {
        let mut ticker = time::interval(Duration::from_secs(5));
        let mut temperature = 21.0_f64;
        loop {
            ticker.tick().await;
            temperature += 0.1;
            let ts = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
            let reading = Reading { device: device.clone(), temperature, ts };
            let payload = serde_json::to_vec(&reading).expect("serialize");
            if let Err(e) = publisher
                .publish(&telemetry_topic, QoS::AtLeastOnce, false, payload)
                .await
            {
                eprintln!("publish request failed: {e}");
            }
        }
    });

    // Event loop : pilote la connexion, gère les reconnexions et les messages entrants.
    loop {
        match eventloop.poll().await {
            Ok(Event::Incoming(Packet::ConnAck(ack))) => {
                println!("connected: {:?}", ack.code);
                // Requêtes non bloquantes : n'attendez jamais le client dans la boucle de poll.
                if let Err(e) = client.try_subscribe(&command_filter, QoS::AtLeastOnce) {
                    eprintln!("subscribe request failed: {e}");
                }
                if let Err(e) = client.try_publish(&status_topic, QoS::AtLeastOnce, true, "online") {
                    eprintln!("status request failed: {e}");
                }
            }
            Ok(Event::Incoming(Packet::Publish(msg))) => {
                println!("command on {}: {}", msg.topic, String::from_utf8_lossy(&msg.payload));
            }
            Ok(_) => {}
            Err(e) => {
                eprintln!("connection error: {e}; reconnecting in 5 s");
                time::sleep(Duration::from_secs(5)).await;
            }
        }
    }
}

Lancez-le :

cargo run

Dans un autre terminal, observez et envoyez une commande :

mosquitto_sub -h broker.connectmq.com -p 1883 -i watch-m3c8wd -t 'connectmq/demo/m3c8wd/#' -v
mosquitto_pub -h broker.connectmq.com -p 1883 -i send-m3c8wd -t 'connectmq/demo/m3c8wd/cmd/fan' -m 'on'

Vous devriez voir la télémétrie toutes les cinq secondes, un online retenu sur le topic de statut et la commande affichée par le programme Rust.

Comment fonctionne le client rumqttc

TLS avec rustls

Transport::tls_with_default_config() construit une configuration rustls qui fait confiance aux certificats racines du système d'exploitation. Le broker utilise un certificat Let's Encrypt, reconnu par tous les systèmes courants ; aucun fichier de certificat n'est donc nécessaire. Le port 8883 plus le transport TLS, c'est toute la configuration. Si vous avez besoin d'une autorité de certification personnalisée ou de certificats clients, construisez vous-même une TlsConfiguration et passez-la à Transport::tls_with_config.

Pourquoi try_subscribe dans la boucle

Le bug rumqttc le plus courant est un interblocage : appeler client.subscribe(...).await dans la même tâche que celle qui interroge l'event loop. Si le channel de requêtes est plein, l'await attend de la place, mais la place ne se libère que lorsque poll() s'exécute, ce qui n'arrivera jamais. Dans la boucle, utilisez try_subscribe et try_publish, qui rendent la main immédiatement. Dans les autres tâches, les versions asynchrones conviennent et offrent une contre-pression naturelle.

Reconnexions et sessions

Lorsque poll() renvoie une erreur, la connexion est perdue. Rappeler poll() se reconnecte, et la pause évite une boucle effrénée tant que le réseau est coupé. Avec set_clean_session(false), le broker conserve les abonnements et les messages QoS 1 en file d'attente entre les connexions, et le client renvoie les publications QoS 1 non acquittées. Se réabonner à chaque ConnAck reste une bonne habitude, car le broker peut avoir fait expirer la session après une longue coupure. Sur le broker public, les sessions durent au maximum 30 minutes.

Last Will

LastWill demande au broker de publier un offline retenu si le client disparaît sans se déconnecter. Avec le online retenu envoyé après chaque connexion, n'importe quel tableau de bord peut indiquer si le service est actif.

Passer à un broker privé

Pour la production, créez un broker dans le portail ConnectMQ et utilisez le nom d'utilisateur, le mot de passe et le client ID générés. Les client IDs sont liés aux identifiants ; l'identifiant doit donc correspondre exactement :

export MQTT_HOST=broker.connectmq.com
export MQTT_CLIENT_ID=YOUR_CLIENT_ID
export MQTT_USERNAME=YOUR_USERNAME
export MQTT_PASSWORD=YOUR_PASSWORD
export MQTT_PREFIX=plant1/line2/edge-gateway
cargo run --release

MQTT 5 avec rumqttc

Le module rumqttc::v5 reprend la même API avec les types MQTT 5. Il vous donne accès à des propriétés comme l'expiration des messages, les user properties et les reason codes, ainsi qu'à un intervalle d'expiration de session à la place du drapeau clean session. La migration consiste surtout à changer les imports et à s'adapter aux champs de Publish qui utilisent des octets pour les topics. Le guide des fonctionnalités de MQTT 5 explique quelles propriétés valent la peine d'être utilisées.

Compilation croisée pour les passerelles edge

Rust est populaire sur les passerelles ARM comme les cartes de la classe Raspberry Pi, et rumqttc s'y déploie facilement car il n'a pas besoin d'OpenSSL. Le principal obstacle est le fournisseur cryptographique, qui compile du code C. Des outils comme cross gèrent la chaîne de compilation pour vous : cross build --release --target aarch64-unknown-linux-gnu produit un binaire que vous pouvez copier sur un Pi 64 bits. N'oubliez pas que la cible a toujours besoin du bundle d'autorités de certification du système, car la configuration TLS par défaut charge les certificats racines de la plateforme au démarrage.

Liste de contrôle production

1. Un client ID unique par processus. Deux processus avec le même identifiant se déconnectent mutuellement en boucle.
2. Dimensionnez le channel. La capacité passée à AsyncClient::new limite les requêtes en file d'attente. Les émetteurs en rafales ont besoin de plus de marge.
3. Temporisez après les erreurs. N'appelez jamais poll() en boucle serrée après des erreurs ; faites une pause, idéalement avec des délais croissants.
4. Choisissez la QoS en connaissance de cause. La QoS 1 convient à la plupart des télémétries ; le plan Free prend en charge les QoS 0 et 1. Consultez les niveaux de QoS MQTT expliqués.
5. Gardez des payloads légers. Le serveur MQTT public gratuit accepte jusqu'à 64 Ko par message, le plan Free 256 Ko.
6. Gardez les secrets dans l'environnement ou dans un coffre de secrets, pas dans le binaire.

Dépannage

La référence de l'API du crate se trouve sur docs.rs.

Tutoriels associés

Si certaines parties de votre système sont écrites dans d'autres langages, le tutoriel client MQTT en Go et le guide client MQTT en Java implémentent le même schéma télémétrie et commandes. La présentation des bibliothèques clientes MQTT compare les clients des différents écosystèmes.

Faites tourner vos services Rust sur ConnectMQ

Un broker MQTT managé signifie que votre passerelle Rust n'a besoin que d'une connexion TLS sortante, sans broker à patcher ni certificats à renouveler. ConnectMQ prend en charge MQTT 3.1.1 et 5.0 avec TLS sur 8883 et un explorateur de topics en direct pour vérifier ce qu'envoie votre client. Créez un broker Serverless gratuit sur app.connectmq.com sans carte bancaire et définissez les quatre variables d'environnement ci-dessus.

Questions fréquentes

rumqttc prend-il en charge TLS ?

Oui. Avec la feature par défaut use-rustls, Transport::tls_with_default_config() se connecte en TLS en utilisant les certificats racines approuvés par le système d'exploitation. Utilisez le port 8883.

rumqttc se reconnecte-t-il automatiquement ?

Il se reconnecte lorsque vous rappelez eventloop.poll() après une erreur. Continuez à interroger dans une boucle avec une courte pause après les erreurs, et réabonnez-vous à chaque ConnAck.

rumqttc prend-il en charge MQTT 5 ?

Oui, via le module rumqttc::v5, qui fournit les propriétés MQTT 5, les reason codes et l'expiration de session.

Pourquoi mon programme rumqttc se fige-t-il ?

Généralement parce qu'il attend subscribe ou publish dans la tâche qui interroge l'event loop. Utilisez plutôt try_subscribe et try_publish à cet endroit.