MQTT em Rust com rumqttc: Publish/Subscribe Assíncrono com TLS e Tokio

Este tutorial de MQTT em Rust com rumqttc mostra como o rumqttc oferece um cliente assíncrono escrito puramente em Rust que se encaixa naturalmente em serviços Tokio, de gateways de borda em uma usina ou mineradora a workers de ingestão no backend. Neste tutorial você vai criar um cliente rumqttc completo que se conecta via TLS com rustls na porta 8883, publica telemetria em JSON, recebe comandos e continua rodando mesmo com quedas de rede.

Por que rumqttc para MQTT em Rust

O rumqttc é a parte cliente do projeto rumqtt. Ele é escrito inteiramente em Rust, suporta MQTT 3.1.1 na raiz da crate e MQTT 5 no módulo v5, e oferece tanto um Client síncrono quanto um AsyncClient assíncrono baseado no Tokio. O TLS vem do rustls por padrão, então você não precisa de OpenSSL na máquina de destino, o que é prático ao fazer cross-compiling para gateways ARM.

O design é diferente das bibliotecas baseadas em callbacks de outras linguagens, e vale entendê-lo antes de escrever código:

Nada acontece na rede a menos que algo esteja fazendo poll do event loop. As requisições do AsyncClient vão para um channel limitado; o event loop o esvazia, envia pacotes, lê os pacotes recebidos e trata os pings. Se poll() retornar um erro, a próxima chamada reconecta. Esse único loop é onde vive a sua lógica de conexão.

Configuração do projeto

Crie uma nova crate binária:

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

Edite o 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"

A feature padrão use-rustls ativa o TLS via rustls com o provedor criptográfico aws-lc-rs. Compilar esse provedor exige um compilador C e o CMake na máquina de build; no Windows, o NASM também pode ser necessário. As máquinas de execução não precisam de nada extra.

Na primeira execução, use o servidor MQTT público (gratuito) em broker.connectmq.com, que aceita clientes anônimos com qualquer client ID não vazio e único. Escolha um prefixo aleatório como connectmq/demo/m3c8wd; o servidor público pode ser lido por qualquer pessoa e serve apenas para testes, então publique apenas dados de teste.

O exemplo completo com rumqttc

Substitua o 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);

    // Tarefa de publicação: envia uma leitura a 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: conduz a conexão, trata reconexões e mensagens recebidas.
    loop {
        match eventloop.poll().await {
            Ok(Event::Incoming(Packet::ConnAck(ack))) => {
                println!("connected: {:?}", ack.code);
                // Requisições não bloqueantes: nunca faça await do client dentro do loop 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;
            }
        }
    }
}

Execute:

cargo run

Em outro terminal, acompanhe e envie um 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'

Você deve ver telemetria a cada cinco segundos, um online retido no topic de status e o comando impresso pelo programa Rust.

Como o cliente rumqttc funciona

TLS com rustls

Transport::tls_with_default_config() monta uma configuração rustls que confia nos certificados raiz do sistema operacional. O broker usa um certificado Let's Encrypt, no qual todo sistema operacional popular confia, então nenhum arquivo de certificado é necessário. Porta 8883 mais transporte TLS é toda a configuração. Se você precisar de uma CA personalizada ou de certificados de cliente, monte um TlsConfiguration e passe-o para Transport::tls_with_config.

Por que try_subscribe dentro do loop

O bug mais comum com rumqttc é um deadlock: chamar client.subscribe(...).await dentro da mesma tarefa que faz poll do event loop. Se o channel de requisições estiver cheio, o await espera por espaço, mas o espaço só é liberado quando poll() roda, o que nunca vai acontecer. Dentro do loop, use try_subscribe e try_publish, que retornam imediatamente. Em outras tarefas, as versões assíncronas funcionam bem e oferecem back-pressure natural.

Reconexões e sessões

Quando poll() retorna um erro, a conexão caiu. Chamar poll() de novo reconecta, e o sleep evita um loop apertado enquanto a rede está fora. Com set_clean_session(false), o broker mantém as assinaturas e as mensagens QoS 1 enfileiradas entre as conexões, e o cliente reenvia as publicações QoS 1 não confirmadas. Assinar de novo a cada ConnAck continua sendo um bom hábito, porque o broker pode ter expirado a sessão após uma interrupção longa. No servidor MQTT público, as sessões duram no máximo 30 minutos.

Last Will

O LastWill instrui o broker a publicar um offline retido se o cliente sumir sem se desconectar. Junto com o online retido enviado após cada conexão, qualquer dashboard consegue mostrar se o serviço está no ar.

Migrando para um broker privado

Para produção, crie um broker no portal da ConnectMQ e use o usuário, a senha e o client ID gerados. Os client IDs são vinculados às credenciais, então o ID precisa corresponder exatamente:

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

O módulo rumqttc::v5 espelha a mesma API com os tipos do MQTT 5. Ele oferece propriedades como expiração de mensagem, user properties e reason codes, além de um intervalo de expiração de sessão no lugar da flag clean-session. A troca consiste basicamente em mudar os imports e se adaptar aos campos de Publish que usam bytes para os topics. O guia de recursos do MQTT 5 explica quais propriedades valem a pena usar.

Cross-compiling para gateways de borda

O Rust é popular em gateways ARM, como placas da classe do Raspberry Pi, e o rumqttc funciona bem nesses ambientes porque não precisa de OpenSSL. O principal obstáculo é o provedor criptográfico, que compila código C. Ferramentas como o cross cuidam da toolchain para você: cross build --release --target aarch64-unknown-linux-gnu gera um binário que você pode copiar para um Pi de 64 bits. Lembre-se de que o destino ainda precisa do bundle de CA do sistema, porque a configuração TLS padrão carrega os certificados raiz da plataforma na inicialização.

Checklist de produção

1. Client ID único por processo. Dois processos com o mesmo ID ficam desconectando um ao outro em loop.
2. Dimensione o channel. A capacidade passada para AsyncClient::new limita as requisições enfileiradas. Publicadores com rajadas precisam de mais folga.
3. Faça back-off nos erros. Nunca chame poll() em um loop apertado após erros; use sleep, de preferência com atrasos crescentes.
4. Escolha o QoS de forma consciente. O QoS 1 atende à maior parte da telemetria; o plano Free suporta QoS 0 e 1. Veja níveis de QoS do MQTT explicados.
5. Mantenha os payloads pequenos. O servidor MQTT público aceita até 64 KB por mensagem; o plano Free, 256 KB.
6. Mantenha os segredos no ambiente ou em um cofre de segredos, não no binário.

Solução de problemas

A referência da API da crate está no docs.rs.

Tutoriais relacionados

Se partes do seu sistema são escritas em outras linguagens, o tutorial de cliente MQTT em Go e o guia de cliente MQTT em Java implementam o mesmo padrão de telemetria e comandos. A visão geral de bibliotecas de cliente MQTT compara clientes entre ecossistemas.

Rode seus serviços Rust na ConnectMQ

Um broker MQTT gerenciado significa que o seu gateway Rust só precisa de uma conexão TLS de saída, sem broker para aplicar patches nem certificados para renovar. A ConnectMQ suporta MQTT 3.1.1 e 5.0 com TLS na porta 8883 e um explorador de topics em tempo real para conferir o que o seu cliente envia. Crie um broker Serverless gratuito em app.connectmq.com sem cartão de crédito e defina as quatro variáveis de ambiente acima.

Perguntas frequentes

O rumqttc suporta TLS?

Sim. Com a feature padrão use-rustls, Transport::tls_with_default_config() se conecta via TLS usando os certificados raiz confiáveis do sistema operacional. Use a porta 8883.

O rumqttc reconecta automaticamente?

Ele reconecta quando você chama eventloop.poll() de novo após um erro. Continue fazendo poll em um loop com um sleep curto após os erros e assine novamente a cada ConnAck.

O rumqttc suporta MQTT 5?

Sim, por meio do módulo rumqttc::v5, que oferece propriedades do MQTT 5, reason codes e expiração de sessão.

Por que meu programa com rumqttc congela?

Geralmente porque ele faz await de subscribe ou publish dentro da tarefa que faz poll do event loop. Use try_subscribe e try_publish ali.