MQTT en Rust con rumqttc: publicación/suscripción asíncrona sobre TLS con Tokio

Este tutorial de MQTT en Rust con rumqttc muestra cómo rumqttc te ofrece un cliente asíncrono escrito íntegramente en Rust que encaja de forma natural en servicios Tokio, desde gateways de borde hasta workers de ingesta en el backend. Construirás un cliente rumqttc completo que se conecta por TLS con rustls en el puerto 8883, publica telemetría JSON, recibe comandos y sigue funcionando aunque se caiga la red.

Por qué rumqttc para MQTT en Rust

rumqttc es la parte cliente del proyecto rumqtt. Está escrito por completo en Rust, admite MQTT 3.1.1 en la raíz del crate y MQTT 5 en su módulo v5, y ofrece tanto un Client síncrono como un AsyncClient asíncrono basado en Tokio. El TLS lo proporciona rustls por defecto, así que no necesitas OpenSSL en la máquina de destino, algo muy práctico al compilar de forma cruzada para gateways ARM.

Su diseño difiere de las bibliotecas basadas en callbacks de otros lenguajes, y conviene entenderlo antes de escribir código:

No ocurre nada en la red a menos que algo esté sondeando el event loop. Las solicitudes de AsyncClient entran en un canal acotado; el event loop lo vacía, envía paquetes, lee los paquetes entrantes y gestiona los pings. Si poll() devuelve un error, la siguiente llamada reconecta. Ese único bucle es donde vive tu lógica de conexión.

Configuración del proyecto

Crea un nuevo crate binario:

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

Edita 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 por defecto use-rustls habilita TLS mediante rustls con el proveedor criptográfico aws-lc-rs. Compilar ese proveedor requiere un compilador de C y CMake en la máquina de compilación; en Windows puede hacer falta también NASM. Las máquinas donde se ejecuta no necesitan nada adicional.

Para la primera prueba, usa el servidor MQTT público gratuito en broker.connectmq.com, pensado para pruebas y no para producción, que acepta clientes anónimos con cualquier client ID único y no vacío. Elige un prefijo aleatorio como connectmq/demo/m3c8wd; el área pública la puede leer cualquiera, así que publica solo datos de prueba.

El ejemplo completo de rumqttc

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

    // Tarea publicadora: envía una lectura cada 5 segundos.
    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: mantiene la conexión, gestiona reconexiones y mensajes entrantes.
    loop {
        match eventloop.poll().await {
            Ok(Event::Incoming(Packet::ConnAck(ack))) => {
                println!("connected: {:?}", ack.code);
                // Solicitudes no bloqueantes: nunca hagas await del cliente dentro del bucle 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;
            }
        }
    }
}

Ejecútalo:

cargo run

En otra terminal, observa los mensajes y envía un comando:

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'

Deberías ver telemetría cada cinco segundos, un online retenido en el topic de estado y el comando impreso por el programa en Rust.

Cómo funciona el cliente rumqttc

TLS con rustls

Transport::tls_with_default_config() crea una configuración de rustls que confía en los certificados raíz del sistema operativo. El broker usa un certificado de Let's Encrypt, en el que confían todos los sistemas operativos habituales, así que no hacen falta archivos de certificado. El puerto 8883 más el transporte TLS es toda la configuración. Si necesitas una CA personalizada o certificados de cliente, crea tú mismo una TlsConfiguration y pásala a Transport::tls_with_config.

Por qué try_subscribe dentro del bucle

El error más habitual con rumqttc es un interbloqueo: llamar a client.subscribe(...).await dentro de la misma tarea que sondea el event loop. Si el canal de solicitudes está lleno, el await espera a que haya espacio, pero el espacio solo se libera cuando se ejecuta poll(), cosa que nunca ocurrirá. Dentro del bucle, usa try_subscribe y try_publish, que retornan de inmediato. En otras tareas, las versiones async están bien y te dan contrapresión de forma natural.

Reconexiones y sesiones

Cuando poll() devuelve un error, la conexión se ha perdido. Volver a llamar a poll() reconecta, y la pausa evita un bucle cerrado mientras la red está caída. Con set_clean_session(false), el broker conserva las suscripciones y los mensajes QoS 1 encolados entre conexiones, y el cliente reenvía las publicaciones QoS 1 no confirmadas. Aun así, volver a suscribirse en cada ConnAck es un buen hábito, porque el broker puede haber expirado la sesión tras una caída larga. En el broker público, las sesiones duran como máximo 30 minutos.

Last Will

LastWill indica al broker que publique un offline retenido si el cliente desaparece sin desconectarse. Junto con el online retenido que se envía tras cada conexión, cualquier panel puede mostrar si el servicio está activo.

Paso a un broker privado

Para producción, crea un broker en el portal de ConnectMQ y usa el usuario, la contraseña y el client ID generados. Los client ID están vinculados a las credenciales, por lo que el ID debe coincidir exactamente:

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 con rumqttc

El módulo rumqttc::v5 replica la misma API con tipos de MQTT 5. Te da propiedades como la expiración de mensajes, las propiedades de usuario y los reason codes, además de un intervalo de expiración de sesión en lugar del flag clean-session. Cambiar consiste sobre todo en modificar los imports y adaptarse a los campos de Publish que usan bytes para los topics. La guía de funciones de MQTT 5 explica qué propiedades merece la pena usar.

Compilación cruzada para gateways de borde

Rust es popular en gateways ARM como las placas tipo Raspberry Pi, y rumqttc se adapta bien a ellas porque no necesita OpenSSL. El principal obstáculo es el proveedor criptográfico, que compila código C. Herramientas como cross gestionan la toolchain por ti: cross build --release --target aarch64-unknown-linux-gnu genera un binario que puedes copiar a una Pi de 64 bits. Recuerda que el destino sigue necesitando el paquete de CA del sistema, porque la configuración TLS por defecto carga los certificados raíz de la plataforma al arrancar.

Lista de comprobación para producción

1. Un client ID único por proceso. Dos procesos con el mismo ID se desconectan mutuamente en bucle.
2. Dimensiona el canal. La capacidad que pasas a AsyncClient::new limita las solicitudes encoladas. Los publicadores con ráfagas necesitan más margen.
3. Espera tras los errores. Nunca llames a poll() en un bucle cerrado después de un error; haz una pausa, idealmente con retardos crecientes.
4. Elige el QoS a conciencia. QoS 1 sirve para la mayoría de la telemetría; el plan Free admite QoS 0 y 1. Consulta los niveles de QoS de MQTT explicados.
5. Mantén los payloads pequeños. El broker público acepta hasta 64 KB por mensaje y el plan Free, 256 KB.
6. Guarda los secretos en el entorno o en un gestor de secretos, no en el binario.

Solución de problemas

La referencia de la API del crate está en docs.rs.

Tutoriales relacionados

Si partes de tu sistema están escritas en otros lenguajes, el tutorial de cliente MQTT en Go y la guía de cliente MQTT en Java implementan el mismo patrón de telemetría y comandos. La visión general de bibliotecas cliente MQTT compara clientes de distintos ecosistemas.

Ejecuta tus servicios Rust en ConnectMQ

Un broker MQTT gestionado significa que tu gateway en Rust solo necesita una conexión TLS saliente, sin broker que parchear ni certificados que renovar. ConnectMQ admite MQTT 3.1.1 y 5.0 con TLS en el puerto 8883 y un explorador de topics en vivo para comprobar lo que envía tu cliente. Crea un broker Serverless gratuito en app.connectmq.com, sin tarjeta de crédito, y configura las cuatro variables de entorno anteriores.

Preguntas frecuentes

¿rumqttc admite TLS?

Sí. Con la feature por defecto use-rustls, Transport::tls_with_default_config() se conecta por TLS usando los certificados raíz de confianza del sistema operativo. Usa el puerto 8883.

¿rumqttc se reconecta automáticamente?

Se reconecta cuando vuelves a llamar a eventloop.poll() después de un error. Sigue sondeando en un bucle con una breve pausa tras los errores y vuelve a suscribirte en cada ConnAck.

¿rumqttc admite MQTT 5?

Sí, mediante el módulo rumqttc::v5, que ofrece propiedades de MQTT 5, reason codes y expiración de sesión.

¿Por qué se congela mi programa con rumqttc?

Normalmente porque hace await de subscribe o publish dentro de la tarea que sondea el event loop. Usa ahí try_subscribe y try_publish en su lugar.