MQTT en Rust con rumqttc: AsyncClient, event loop y TLS

rumqttc es el cliente MQTT en Rust puro más utilizado. Divide el trabajo en dos: un AsyncClient que clonas y usas para publicar y suscribirte, y un EventLoop que sondeas (poll) para mover la conexión de red. Cuando entiendes ese diseño, el resto es sencillo. En este tutorial construimos un cliente Tokio que publica y se suscribe, se recupera de cortes de red y se conecta por TLS con rustls.

Cliente y event loop

AsyncClient::new(options, cap) devuelve un par:

La consecuencia: si nadie hace poll del event loop, no se envía nada, y cuando el canal se llena, publish().await espera hasta que el bucle se ponga al día. Es una ventaja —te da backpressure—, pero sorprende a quien espera que publish llegue a la red por sí solo.

Configuración

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 activa rustls por defecto (feature use-rustls), así que TLS funciona sin OpenSSL. Los ejemplos usan la API de la línea 0.24; las versiones posteriores mantienen la misma estructura.

Publicar y suscribirse

Este programa publica un mensaje por segundo desde una tarea en segundo plano e imprime todo lo que recibe en el mismo 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 clon del cliente en su propia tarea.
    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!("fallo al publicar: {e}");
            }
            time::sleep(Duration::from_secs(1)).await;
        }
    });

    // Event loop: mueve la conexión y entrega los paquetes entrantes.
    loop {
        match eventloop.poll().await {
            Ok(Event::Incoming(Packet::ConnAck(ack))) => {
                println!("conectado, session present: {}", ack.session_present);
                // try_subscribe no hace await, así que no puede bloquear este bucle.
                if let Err(e) = client.try_subscribe(TOPIC, QoS::AtLeastOnce) {
                    eprintln!("fallo al suscribir: {e}");
                }
            }
            Ok(Event::Incoming(Packet::Publish(p))) => {
                println!("recibido {}: {}", p.topic, String::from_utf8_lossy(&p.payload));
            }
            Ok(_) => {}
            Err(e) => {
                eprintln!("error de conexión: {e}; reintento en 2 s");
                time::sleep(Duration::from_secs(2)).await;
            }
        }
    }
}

Ejecútalo con cargo run. Para confirmar que los mensajes llegan de verdad al broker, abre el cliente MQTT online, conéctate a broker.connectmq.com:8084 por WebSocket seguro y suscríbete a testmqtt/rust-rumqttc/#; publica desde el navegador y el programa Rust imprimirá tu mensaje.

Tres decisiones de este código son deliberadas:

• Suscribirse en cada ConnAck. Por defecto se usa clean session, así que el broker olvida las suscripciones cuando cae la conexión. Suscribirse cada vez que llega un ConnAck las restaura tras cada reconexión.
• try_subscribe dentro del bucle. Llamar a client.subscribe(...).await desde la misma tarea que hace poll del event loop puede bloquearse cuando el canal está lleno, porque el bucle es lo único que lo vacía. Los métodos try_*, que no son async, fallan rápido.
• Pausa tras los errores. El siguiente poll() tras un error reconecta de inmediato. Sin una espera, una caída del broker se convierte en un bucle frenético de intentos de conexión.

Credenciales y TLS con rustls

Con un broker real querrás TLS en el puerto 8883 y usuario y contraseña:

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() crea una configuración de rustls que confía en los certificados raíz nativos de la plataforma, que es lo que necesitas con brokers que usan certificados de confianza pública. Para una CA privada, pasa tú el fichero PEM:

use rumqttc::{TlsConfiguration, Transport};

let ca = std::fs::read("ca.pem").expect("leer el fichero de la CA");
opts.set_transport(Transport::Tls(TlsConfiguration::Simple {
    ca,
    alpn: None,
    client_auth: None, // Some((cert_pem_bytes, key_pem_bytes)) para TLS mutuo
}));

Si necesitas control total (verificadores propios, cipher suites concretas), TlsConfiguration::Rustls(Arc<rustls::ClientConfig>) acepta una configuración rustls preparada. Tu dependencia rustls debe tener la misma versión mayor que usa rumqttc —compruébalo con cargo tree -i rustls—; si no, los tipos no coincidirán.

El nombre de host que pasas a MqttOptions::new es el que rustls verifica contra el certificado. Conectarse por IP a un certificado emitido para un nombre DNS falla la verificación; consulta hostname mismatch y la prueba de conexión TLS para inspeccionar el certificado que presenta el broker.

WebSockets y MQTT 5

Con la feature websocket activada (cargo add rumqttc --features websocket), rumqttc puede tunelizar MQTT sobre WebSockets, útil detrás de proxies que solo permiten HTTPS. Pasa la URL completa como host y elige el transporte:

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

Para MQTT 5, importa desde rumqttc::v5 en lugar de la raíz del crate. El cliente y el event loop funcionan igual; los paquetes llevan properties (user properties, message expiry, response topic) y los acks llevan reason codes de MQTT 5 que puedes consultar en la referencia de reason codes.

Errores que te encontrarás

eventloop.poll() devuelve un ConnectionError. La variante suele señalar directamente la causa:

El último sorprende a mucha gente: rumqttc viene con un tamaño máximo de paquete conservador (10 KiB en las versiones recientes), así que los payloads JSON grandes o las imágenes necesitan un límite explícito. Usa la calculadora de tamaño de paquete para ver cuánto ocupa realmente un mensaje en la red.

Probar el cliente

Una lista rápida antes de fiarte de un cliente rumqttc:

1. Ida y vuelta. Publica desde el cliente online y confirma que el programa Rust imprime el mensaje; luego al revés.
2. Corte de red. Desactiva el Wi-Fi durante más tiempo que el keep-alive, actívalo de nuevo y comprueba que aparece una nueva línea "conectado" y que los mensajes vuelven: prueba de que la resuscripción en ConnAck funciona.
3. Credenciales erróneas. Ejecuta una vez con una contraseña incorrecta y confirma que tu código registra el rechazo en lugar de girar en silencio.
4. Payload grande. Publica un mensaje de más de 10 KiB y confirma que tu ajuste de tamaño de paquete es correcto.

Para un enfoque más amplio, consulta buenas prácticas de reconexión MQTT y la comparativa de librerías en las mejores librerías MQTT; si prefieres bindings a la librería Eclipse Paho C, MQTT en Rust con paho-mqtt cubre la alternativa.

Abre el topic en el cliente online

rumqttc es pequeño, async desde el principio y Rust puro: mantén una tarea haciendo poll del event loop, suscríbete en cada ConnAck, espera tras los errores y configura TLS mediante Transport. Con esos cuatro hábitos, el mismo cliente funciona desde una prueba en el portátil hasta una flota en producción.

Preguntas frecuentes