MQTT en Rust con paho-mqtt: cliente async y MQTT 5

El crate paho-mqtt es el cliente oficial de Eclipse Paho para Rust. Es un wrapper seguro sobre la librería Paho C, lo que significa que obtienes el mismo motor de protocolo probado en batalla que usan muchos proyectos en C y embebidos, con una API async idiomática por encima. Soporta MQTT 3.1.1 y MQTT 5, TCP, TLS y WebSockets. Esta guía cubre la compilación, un cliente async MQTT 5 completo, la reconexión, TLS y los errores que más probablemente verás.

Si prefieres evitar una dependencia de C, rumqttc es la alternativa en Rust puro.

Requisitos de compilación

Por defecto, el crate compila una copia incluida de Paho C con CMake y la enlaza estáticamente, con OpenSSL para TLS. Instala primero el 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"] }

Consulta crates.io para ver la versión más reciente; la API mostrada aquí se ha mantenido estable en las últimas versiones. La primera compilación tarda algo más porque se compila la librería C; después queda en caché como cualquier otro crate.

Un cliente async MQTT 5

El patrón es: crear el cliente, obtener el stream de mensajes antes de conectar, conectar, suscribirse y leer el stream. Un None en el stream significa que se perdió la conexión.

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

    // Obtener el stream antes de conectar para no perder ningún mensaje.
    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!("conectado a {} con MQTT v{}", conn.server_uri, conn.mqtt_version);
    }
    cli.subscribe(TOPIC, mqtt::QOS_1).await?;

    // Publicar cinco mensajes desde una tarea aparte.
    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!("fallo al publicar: {e}");
            }
            tokio::time::sleep(Duration::from_secs(1)).await;
        }
    });

    while let Some(msg_opt) = stream.next().await {
        match msg_opt {
            Some(msg) => println!("recibido {}: {}", msg.topic(), msg.payload_str()),
            None => {
                eprintln!("conexión perdida, reconectando");
                while let Err(e) = cli.reconnect().await {
                    eprintln!("fallo al reconectar: {e}");
                    tokio::time::sleep(Duration::from_secs(2)).await;
                }
                cli.subscribe(TOPIC, mqtt::QOS_1).await?;
            }
        }
    }
    Ok(())
}

Ejecútalo con cargo run. Deberías ver la línea de conexión con MQTT v5 y después los cinco mensajes de vuelta. Abre el cliente MQTT online, suscríbete a testmqtt/rust-paho/# y verás los mismos mensajes; publica uno desde el navegador y el programa Rust también lo imprimirá.

Puntos clave:

• mqtt_version(MQTT_VERSION_5) en las opciones de creación y new_v5() en las de conexión: ambos son necesarios para una conexión MQTT 5.
• Los futures no dependen del runtime. Paho C ejecuta sus propios hilos de red; puedes hacer await de los tokens desde Tokio, async-std o un simple block_on.
• publish con QoS 1 se resuelve cuando llega el PUBACK, así que un error ahí significa que el broker no aceptó el mensaje.
• Vuelve a suscribirte tras reconnect() cuando usas clean start, porque el broker descartó la sesión anterior.

Reconexión automática y sesiones persistentes

En lugar del bucle manual, puedes dejar que Paho reconecte y usar un callback para restaurar el estado:

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)conectado"));
cli.set_connection_lost_callback(|_cli| eprintln!("conexión perdida"));

Con clean_start(false) y un intervalo de expiración de sesión, el broker conserva tus suscripciones y encola los mensajes QoS 1 mientras estás offline, así que no necesitas volver a suscribirte. Ten en cuenta que el endpoint de pruebas compartido no conserva sesiones offline; prueba esto contra tu propia instancia privada de broker. Sesiones persistentes explica qué guarda el broker y durante cuánto tiempo.

Paho también puede persistir en el lado del cliente sus propios mensajes QoS 1/2 en vuelo. Elige el comportamiento de forma explícita con .persistence(...) en las opciones de creación: mqtt::PersistenceType::None lo mantiene todo en memoria y mqtt::PersistenceType::File lo escribe en disco para que sobreviva a un reinicio del proceso. Si ves aparecer en el directorio de trabajo carpetas con el nombre de tu client ID, es la persistencia en fichero funcionando.

TLS, credenciales y WebSockets

Cambia la URI a ssl:// (o mqtts:// con builds recientes de Paho C) en el puerto 8883 y añade las opciones 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()
    // Para una CA privada: .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();

Sin trust store, se usan las ubicaciones de certificados por defecto de OpenSSL. Si la verificación falla con un broker que tiene un certificado público, apunta trust_store al bundle del sistema (por ejemplo /etc/ssl/certs/ca-certificates.crt en Debian). Para certificados de cliente usa key_store y private_key. La prueba de conexión TLS muestra cómo inspeccionar la cadena con openssl s_client antes de culpar al código.

En redes que solo permiten HTTPS, usa wss://broker.connectmq.com:8084/mqtt como URI del servidor con las mismas opciones SSL.

Errores que verás

Busca cualquier código de MQTT 5 en la referencia de reason codes. Para una forma estructurada de probar la expiración de sesión, las user properties y los topic aliases contra tu broker, consulta probar las funciones de MQTT 5.

Lista de pruebas

1. Ida y vuelta: publica desde el cliente online, recibe en Rust, y al revés.
2. Desconecta el cable de red o el Wi-Fi durante más tiempo que el keep-alive; confirma que se ejecuta la ruta de reconexión y que los mensajes vuelven.
3. Reinicia tu programa con clean_start(false) contra un broker privado y confirma que los mensajes publicados mientras estaba caído llegan después.
4. Ejecuta una vez con credenciales erróneas y confirma que el error se registra con claridad.

Observa el topic de la demo en tu navegador

paho-mqtt da a los programas Rust el motor de Eclipse Paho con MQTT 5, TLS y WebSockets, a cambio de un paso de compilación en C. Obtén el stream antes de conectar, decide cómo restauras las suscripciones y prueba una desconexión una vez: a partir de ahí, el cliente se comporta igual desde tu portátil hasta producción. Si también trabajas en C, el tutorial de Paho C muestra la librería que hay debajo.

Preguntas frecuentes