MQTT em Rust com paho-mqtt: cliente async e MQTT 5

O crate paho-mqtt é o cliente oficial do Eclipse Paho para Rust. Ele é um wrapper seguro sobre a biblioteca Paho C, ou seja, você ganha o mesmo motor de protocolo testado em campo usado por muitos projetos em C e embarcados, com uma API async idiomática por cima. Suporta MQTT 3.1.1 e MQTT 5, TCP, TLS e WebSockets. Este guia cobre a compilação, um cliente async MQTT 5 completo, reconexão, TLS e os erros que você mais deve encontrar.

Se preferir evitar uma dependência em C, o rumqttc é a alternativa em Rust puro.

Requisitos de compilação

Por padrão o crate compila uma cópia embutida do Paho C com CMake e faz link estático, com OpenSSL para TLS. Instale o toolchain primeiro:

# 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"] }

Confira no crates.io a versão mais nova; a API mostrada aqui se manteve estável nas versões recentes. A primeira compilação demora um pouco mais porque a biblioteca C é compilada; depois disso ela fica em cache como qualquer outro crate.

Um cliente async MQTT 5

O padrão é: criar o cliente, obter o stream de mensagens antes de conectar, conectar, fazer o subscribe e então ler o stream. Um None no stream significa que a conexão caiu.

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

    // Obter o stream antes de conectar para não perder nenhuma mensagem.
    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 {} com MQTT v{}", conn.server_uri, conn.mqtt_version);
    }
    cli.subscribe(TOPIC, mqtt::QOS_1).await?;

    // Publicar cinco mensagens a partir de uma task separada.
    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!("falha no publish: {e}");
            }
            tokio::time::sleep(Duration::from_secs(1)).await;
        }
    });

    while let Some(msg_opt) = stream.next().await {
        match msg_opt {
            Some(msg) => println!("recebido {}: {}", msg.topic(), msg.payload_str()),
            None => {
                eprintln!("conexão perdida, reconectando");
                while let Err(e) = cli.reconnect().await {
                    eprintln!("falha ao reconectar: {e}");
                    tokio::time::sleep(Duration::from_secs(2)).await;
                }
                cli.subscribe(TOPIC, mqtt::QOS_1).await?;
            }
        }
    }
    Ok(())
}

Rode com cargo run. Você deve ver a linha de conexão com MQTT v5 e depois as cinco mensagens voltando. Abra o cliente MQTT online, inscreva-se em testmqtt/rust-paho/# e você verá as mesmas mensagens; publique uma pelo navegador e o programa Rust também vai imprimi-la.

Pontos-chave:

• mqtt_version(MQTT_VERSION_5) nas opções de criação e new_v5() nas opções de conexão — os dois são necessários para uma conexão MQTT 5.
• As futures não dependem de runtime. O Paho C roda suas próprias threads de rede; você pode fazer await dos tokens no Tokio, no async-std ou num simples block_on.
• publish com QoS 1 resolve quando o PUBACK chega, então um erro ali significa que o broker não aceitou a mensagem.
• Refaça o subscribe depois do reconnect() quando usar clean start, porque o broker descartou a sessão antiga.

Reconexão automática e sessões persistentes

Em vez do loop manual, você pode deixar o Paho reconectar e usar um callback para restaurar o 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!("conexão perdida"));

Com clean_start(false) e um intervalo de expiração de sessão, o broker mantém suas inscrições e enfileira as mensagens QoS 1 enquanto você está offline, então não é preciso refazer o subscribe. Observe que o endpoint de testes compartilhado não guarda sessões offline; teste isto na sua própria instância privada de broker. Sessões persistentes explica o que o broker armazena e por quanto tempo.

O Paho também pode persistir no lado do cliente suas próprias mensagens QoS 1/2 em trânsito. Escolha o comportamento explicitamente com .persistence(...) nas opções de criação: mqtt::PersistenceType::None mantém tudo em memória e mqtt::PersistenceType::File grava em disco para sobreviver a um restart do processo. Se aparecerem no diretório de trabalho pastas com o nome do seu client ID, é a persistência em arquivo em ação.

TLS, credenciais e WebSockets

Troque a URI para ssl:// (ou mqtts:// em builds recentes do Paho C) na porta 8883 e adicione as opções 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 uma 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();

Sem trust store, são usados os locais de certificados padrão do OpenSSL. Se a verificação falhar num broker com certificado público, aponte trust_store para o bundle do sistema (por exemplo /etc/ssl/certs/ca-certificates.crt no Debian). Para certificados de cliente use key_store e private_key. O teste de conexão TLS mostra como inspecionar a cadeia com openssl s_client antes de culpar o código.

Em redes que só liberam HTTPS, use wss://broker.connectmq.com:8084/mqtt como URI do servidor com as mesmas opções SSL.

Erros que você vai ver

Consulte qualquer código do MQTT 5 na referência de reason codes. Para uma forma estruturada de exercitar expiração de sessão, user properties e topic aliases no seu broker, veja testando recursos do MQTT 5.

Checklist de testes

1. Ida e volta: publique pelo cliente online, receba em Rust, e o contrário.
2. Desconecte o cabo de rede ou o Wi-Fi por mais tempo que o keep-alive; confirme que o caminho de reconexão roda e que as mensagens voltam.
3. Reinicie o programa com clean_start(false) num broker privado e confirme que as mensagens publicadas enquanto ele estava parado chegam depois.
4. Rode uma vez com credenciais erradas e confirme que o erro é registrado com clareza.

Acompanhe o topic da demo no navegador

O paho-mqtt dá aos programas Rust o motor do Eclipse Paho com MQTT 5, TLS e WebSockets, ao custo de uma etapa de compilação em C. Obtenha o stream antes de conectar, decida como restaurar as inscrições e teste uma desconexão uma vez — a partir daí o cliente se comporta igual do notebook à produção. Se você também trabalha com C, o tutorial de Paho C mostra a biblioteca por baixo.

Perguntas frequentes