MQTT en Rust avec paho-mqtt : client async et MQTT 5

Le crate paho-mqtt est le client Eclipse Paho officiel pour Rust. C'est un wrapper sûr autour de la bibliothèque Paho C : vous bénéficiez du même moteur de protocole éprouvé que de nombreux projets C et embarqués, avec une API async idiomatique par-dessus. Il prend en charge MQTT 3.1.1 et MQTT 5, TCP, TLS et WebSockets. Ce guide couvre la compilation, un client async MQTT 5 complet, la reconnexion, TLS et les erreurs que vous rencontrerez le plus probablement.

Si vous préférez éviter une dépendance C, rumqttc est l'alternative en pur Rust.

Prérequis de compilation

Par défaut, le crate compile une copie embarquée de Paho C avec CMake et la lie statiquement, avec OpenSSL pour TLS. Installez d'abord la 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"] }

Vérifiez la dernière version sur crates.io ; l'API présentée ici est restée stable au fil des versions récentes. La première compilation est un peu plus longue car la bibliothèque C est compilée ; ensuite elle est mise en cache comme n'importe quel crate.

Un client async MQTT 5

Le schéma : créer le client, récupérer le stream de messages avant de se connecter, se connecter, s'abonner, puis lire le stream. Un None dans le stream signifie que la connexion a été perdue.

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

    // Récupérer le stream avant la connexion pour ne manquer aucun message.
    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!("connecté à {} en MQTT v{}", conn.server_uri, conn.mqtt_version);
    }
    cli.subscribe(TOPIC, mqtt::QOS_1).await?;

    // Publier cinq messages depuis une tâche séparée.
    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!("échec du publish : {e}");
            }
            tokio::time::sleep(Duration::from_secs(1)).await;
        }
    });

    while let Some(msg_opt) = stream.next().await {
        match msg_opt {
            Some(msg) => println!("reçu {} : {}", msg.topic(), msg.payload_str()),
            None => {
                eprintln!("connexion perdue, reconnexion");
                while let Err(e) = cli.reconnect().await {
                    eprintln!("échec de la reconnexion : {e}");
                    tokio::time::sleep(Duration::from_secs(2)).await;
                }
                cli.subscribe(TOPIC, mqtt::QOS_1).await?;
            }
        }
    }
    Ok(())
}

Lancez-le avec cargo run. Vous devriez voir la ligne de connexion avec MQTT v5, puis les cinq messages revenir. Ouvrez le client MQTT en ligne, abonnez-vous à testmqtt/rust-paho/# et vous verrez les mêmes messages ; publiez-en un depuis le navigateur et le programme Rust l'affichera aussi.

Points clés :

• mqtt_version(MQTT_VERSION_5) dans les options de création et new_v5() dans les options de connexion — les deux sont nécessaires pour une connexion MQTT 5.
• Les futures sont indépendantes du runtime. Paho C fait tourner ses propres threads réseau ; vous pouvez attendre les tokens depuis Tokio, async-std ou un simple block_on.
• publish en QoS 1 se résout à l'arrivée du PUBACK : une erreur à ce niveau signifie que le broker n'a pas accepté le message.
• Réabonnez-vous après reconnect() quand vous utilisez clean start, car le broker a supprimé l'ancienne session.

Reconnexion automatique et sessions persistantes

Au lieu de la boucle manuelle, vous pouvez laisser Paho se reconnecter et restaurer l'état via un callback :

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)connecté"));
cli.set_connection_lost_callback(|_cli| eprintln!("connexion perdue"));

Avec clean_start(false) et un intervalle d'expiration de session, le broker conserve vos abonnements et met en file les messages QoS 1 pendant que vous êtes hors ligne : inutile de vous réabonner. Notez que l'endpoint de test partagé ne conserve pas les sessions hors ligne ; testez cela avec votre propre instance de broker privée. Sessions persistantes explique ce que le broker stocke et pendant combien de temps.

Paho peut aussi persister côté client ses propres messages QoS 1/2 en transit. Choisissez explicitement le comportement avec .persistence(...) dans les options de création : mqtt::PersistenceType::None garde tout en mémoire, mqtt::PersistenceType::File écrit sur disque pour survivre à un redémarrage du processus. Si des dossiers portant le nom de votre client ID apparaissent dans le répertoire de travail, c'est la persistance sur fichier.

TLS, identifiants et WebSockets

Passez l'URI en ssl:// (ou mqtts:// avec les builds récents de Paho C) sur le port 8883 et ajoutez les options SSL :

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()
    // Pour une CA privée : .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();

Sans trust store, les emplacements de certificats par défaut d'OpenSSL sont utilisés. Si la vérification échoue sur un broker doté d'un certificat public, faites pointer trust_store vers le bundle du système (par exemple /etc/ssl/certs/ca-certificates.crt sous Debian). Pour les certificats client, utilisez key_store et private_key. Le test de connexion TLS montre comment inspecter la chaîne avec openssl s_client avant d'accuser le code.

Sur les réseaux qui n'autorisent que HTTPS, utilisez wss://broker.connectmq.com:8084/mqtt comme URI serveur avec les mêmes options SSL.

Les erreurs que vous verrez

Retrouvez n'importe quel code MQTT 5 dans la référence des reason codes. Pour tester de façon structurée l'expiration de session, les user properties et les topic aliases sur votre broker, voir tester les fonctionnalités de MQTT 5.

Checklist de test

1. Aller-retour : publiez depuis le client en ligne, recevez en Rust, puis l'inverse.
2. Débranchez le câble réseau ou coupez le Wi-Fi plus longtemps que le keep-alive ; vérifiez que le chemin de reconnexion s'exécute et que les messages reprennent.
3. Redémarrez votre programme avec clean_start(false) sur un broker privé et vérifiez que les messages publiés pendant l'arrêt arrivent ensuite.
4. Lancez une fois avec de mauvais identifiants et vérifiez que l'erreur est clairement journalisée.

Suivez le topic de la démo dans le navigateur

paho-mqtt apporte aux programmes Rust le moteur Eclipse Paho avec MQTT 5, TLS et WebSockets, au prix d'une étape de compilation C. Récupérez le stream avant de vous connecter, décidez comment restaurer les abonnements et testez une déconnexion une fois : le client se comportera alors de la même façon du portable à la production. Si vous travaillez aussi en C, le tutoriel Paho C présente la bibliothèque sous-jacente.

Questions fréquentes