MQTT em Rust com rumqttc: AsyncClient, event loop e TLS

rumqttc é o cliente MQTT em Rust puro mais usado. Ele divide o trabalho em duas partes: um AsyncClient que você clona e usa para publicar e se inscrever, e um EventLoop que você consulta (poll) para movimentar a conexão de rede. Depois que esse design fica claro, o resto é simples. Este tutorial monta um cliente Tokio que publica e se inscreve, se recupera de quedas de rede e conecta via TLS com rustls.

Cliente e event loop

AsyncClient::new(options, cap) retorna um par:

A consequência: se ninguém fizer poll do event loop, nada é enviado, e quando o canal enche, publish().await espera até o loop alcançar. Isso é um recurso — dá backpressure —, mas surpreende quem espera que o publish vá para a rede sozinho.

Configuração

cargo new rumqttc-demo && cd rumqttc-demo
cargo add rumqttc
cargo add tokio --features full

[dependencies]
rumqttc = "0.24"
tokio = { version = "1", features = ["full"] }

O rumqttc habilita rustls por padrão (feature use-rustls), então TLS funciona sem OpenSSL. Os exemplos usam a API da linha 0.24; versões posteriores mantêm a mesma estrutura.

Publish e subscribe

Este programa publica uma mensagem por segundo a partir de uma task em segundo plano e imprime tudo o que recebe no mesmo 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: um clone do cliente em sua própria task.
    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!("falha no publish: {e}");
            }
            time::sleep(Duration::from_secs(1)).await;
        }
    });

    // Event loop: movimenta a conexão e entrega os pacotes recebidos.
    loop {
        match eventloop.poll().await {
            Ok(Event::Incoming(Packet::ConnAck(ack))) => {
                println!("conectado, session present: {}", ack.session_present);
                // try_subscribe não faz await, então não bloqueia este loop.
                if let Err(e) = client.try_subscribe(TOPIC, QoS::AtLeastOnce) {
                    eprintln!("falha no subscribe: {e}");
                }
            }
            Ok(Event::Incoming(Packet::Publish(p))) => {
                println!("recebido {}: {}", p.topic, String::from_utf8_lossy(&p.payload));
            }
            Ok(_) => {}
            Err(e) => {
                eprintln!("erro de conexão: {e}; nova tentativa em 2 s");
                time::sleep(Duration::from_secs(2)).await;
            }
        }
    }
}

Rode com cargo run. Para confirmar que as mensagens chegam mesmo ao broker, abra o cliente MQTT online, conecte em broker.connectmq.com:8084 via WebSocket seguro e se inscreva em testmqtt/rust-rumqttc/#; publique pelo navegador e o programa Rust vai imprimir sua mensagem.

Três escolhas neste código são intencionais:

• Subscribe a cada ConnAck. O padrão é clean session, então o broker esquece as inscrições quando a conexão cai. Fazer o subscribe a cada ConnAck as restaura depois de cada reconexão.
• try_subscribe dentro do loop. Chamar client.subscribe(...).await na mesma task que faz poll do event loop pode travar quando o canal está cheio, porque o loop é a única coisa que o esvazia. Os métodos try_*, que não são async, falham rápido.
• Pausa após erros. O próximo poll() depois de um erro reconecta imediatamente. Sem uma espera, uma queda do broker vira um loop frenético de tentativas de conexão.

Credenciais e TLS com rustls

Em um broker real você vai querer TLS na porta 8883 e usuário e senha:

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() cria uma configuração rustls que confia nos certificados raiz nativos da plataforma, que é o que você quer para brokers com certificados publicamente confiáveis. Para uma CA privada, passe o arquivo PEM você mesmo:

use rumqttc::{TlsConfiguration, Transport};

let ca = std::fs::read("ca.pem").expect("ler o arquivo da CA");
opts.set_transport(Transport::Tls(TlsConfiguration::Simple {
    ca,
    alpn: None,
    client_auth: None, // Some((cert_pem_bytes, key_pem_bytes)) para TLS mútuo
}));

Se precisar de controle total (verificadores próprios, cipher suites específicas), TlsConfiguration::Rustls(Arc<rustls::ClientConfig>) aceita uma configuração rustls pronta. Sua dependência rustls precisa ter a mesma versão major que o rumqttc usa — confira com cargo tree -i rustls —, senão os tipos não batem.

O host que você passa em MqttOptions::new é o nome que o rustls verifica contra o certificado. Conectar por IP a um certificado emitido para um nome DNS falha na verificação; veja hostname mismatch e o teste de conexão TLS para inspecionar o certificado apresentado pelo broker.

WebSockets e MQTT 5

Com a feature websocket habilitada (cargo add rumqttc --features websocket), o rumqttc consegue tunelar MQTT por WebSockets — útil atrás de proxies que só liberam HTTPS. Passe a URL completa como host e escolha o 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, importe de rumqttc::v5 em vez da raiz do crate. O cliente e o event loop funcionam do mesmo jeito; os pacotes carregam properties (user properties, message expiry, response topic) e os acks trazem reason codes do MQTT 5 que você pode consultar na referência de reason codes.

Erros que você vai encontrar

eventloop.poll() retorna um ConnectionError. A variante costuma apontar direto para a causa:

O último pega muita gente: o rumqttc vem com um tamanho máximo de pacote conservador (10 KiB nas versões recentes), então payloads JSON grandes ou imagens precisam de um limite explícito. Use a calculadora de tamanho de pacote para ver o tamanho real de uma mensagem na rede.

Testando o cliente

Um checklist rápido antes de confiar em um cliente rumqttc:

1. Ida e volta. Publique pelo cliente online e confirme que o programa Rust imprime a mensagem; depois faça o caminho inverso.
2. Queda de rede. Desligue o Wi-Fi por mais tempo que o keep-alive, religue e verifique se aparece uma nova linha "conectado" e as mensagens voltam — prova de que o subscribe no ConnAck funciona.
3. Credenciais erradas. Rode uma vez com a senha errada e confirme que seu código registra a recusa em vez de ficar girando em silêncio.
4. Payload grande. Publique uma mensagem acima de 10 KiB e confirme que o ajuste de tamanho de pacote está certo.

Para uma visão mais ampla, veja boas práticas de reconexão MQTT e a comparação de bibliotecas em melhores bibliotecas MQTT; se você prefere bindings para a biblioteca Eclipse Paho C, MQTT em Rust com paho-mqtt cobre a alternativa.

Abra o topic no cliente online

O rumqttc é pequeno, async desde a base e Rust puro: mantenha uma task fazendo poll do event loop, faça o subscribe a cada ConnAck, espere após erros e configure TLS via Transport. Com esses quatro hábitos, o mesmo cliente vai do teste no notebook até uma frota em produção.

Perguntas frequentes