Rust ve rumqttc ile MQTT: AsyncClient, event loop ve TLS

rumqttc, en yaygın kullanılan saf Rust MQTT istemcisidir. İşi ikiye böler: publish ve subscribe için klonlayıp kullandığınız bir AsyncClient ve ağ bağlantısını ilerletmek için poll ettiğiniz bir EventLoop. Bu tasarımı kavradığınızda gerisi kolaydır. Bu rehberde publish ve subscribe yapan, ağ kesintilerinden toparlanan ve rustls ile TLS üzerinden bağlanan bir Tokio istemcisi yazıyoruz.

İstemci ve event loop

AsyncClient::new(options, cap) bir çift döndürür:

Sonuç: event loop'u kimse poll etmiyorsa hiçbir şey gönderilmez ve kanal dolduğunda publish().await, döngü yetişene kadar bekler. Bu bilinçli bir özelliktir — size backpressure sağlar — ama publish'in ağa kendiliğinden gitmesini bekleyenleri şaşırtır.

Kurulum

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 varsayılan olarak rustls'i etkinleştirir (use-rustls feature'ı), bu yüzden TLS, OpenSSL olmadan çalışır. Örnekler 0.24 serisinin API'sini kullanır; sonraki sürümler aynı yapıyı korur.

Publish ve subscribe

Bu program arka plandaki bir task'tan her saniye bir mesaj yayınlar ve aynı topic'te aldığı her şeyi yazdırır.

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: istemcinin ayrı bir task'taki klonu.
    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!("publish başarısız: {e}");
            }
            time::sleep(Duration::from_secs(1)).await;
        }
    });

    // Event loop: bağlantıyı ilerletir ve gelen paketleri teslim eder.
    loop {
        match eventloop.poll().await {
            Ok(Event::Incoming(Packet::ConnAck(ack))) => {
                println!("bağlandı, session present: {}", ack.session_present);
                // try_subscribe await etmez, bu yüzden bu döngüyü bloklayamaz.
                if let Err(e) = client.try_subscribe(TOPIC, QoS::AtLeastOnce) {
                    eprintln!("subscribe başarısız: {e}");
                }
            }
            Ok(Event::Incoming(Packet::Publish(p))) => {
                println!("alındı {}: {}", p.topic, String::from_utf8_lossy(&p.payload));
            }
            Ok(_) => {}
            Err(e) => {
                eprintln!("bağlantı hatası: {e}; 2 sn sonra yeniden denenecek");
                time::sleep(Duration::from_secs(2)).await;
            }
        }
    }
}

cargo run ile çalıştırın. Mesajların broker'a gerçekten ulaştığını doğrulamak için online MQTT istemcisini açın, güvenli WebSocket ile broker.connectmq.com:8084'e bağlanın ve testmqtt/rust-rumqttc/# topic'ine abone olun; tarayıcıdan bir mesaj yayınlayın, Rust programı mesajınızı yazdıracaktır.

Bu koddaki üç tercih bilinçlidir:

• Her ConnAck'te abone olun. Varsayılan clean session'dır; bağlantı koptuğunda broker abonelikleri unutur. Her ConnAck geldiğinde abone olmak, her yeniden bağlanmadan sonra abonelikleri geri getirir.
• Döngü içinde try_subscribe. Event loop'u poll eden task içinden client.subscribe(...).await çağırmak, kanal doluyken kilitlenmeye (deadlock) yol açabilir; çünkü kanalı boşaltan tek şey o döngüdür. Async olmayan try_* metotları ise hemen hata döndürür.
• Hatalardan sonra bekleyin. Bir hatadan sonraki poll() hemen yeniden bağlanır. Bekleme olmadan, broker kesintisi sıkışık bir bağlantı denemesi döngüsüne dönüşür.

Kimlik bilgileri ve rustls ile TLS

Gerçek bir broker için 8883 portunda TLS ve kullanıcı adı ile parola isteyeceksiniz:

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(), platformun yerel kök sertifikalarına güvenen bir rustls yapılandırması oluşturur; herkesçe güvenilen sertifikalara sahip broker'lar için istediğiniz de budur. Özel bir CA için PEM dosyasını kendiniz verin:

use rumqttc::{TlsConfiguration, Transport};

let ca = std::fs::read("ca.pem").expect("CA dosyasını oku");
opts.set_transport(Transport::Tls(TlsConfiguration::Simple {
    ca,
    alpn: None,
    client_auth: None, // karşılıklı TLS için Some((cert_pem_bytes, key_pem_bytes))
}));

Tam denetim gerekiyorsa (özel doğrulayıcılar, belirli cipher suite'ler), TlsConfiguration::Rustls(Arc<rustls::ClientConfig>) hazır bir rustls yapılandırması kabul eder. Kendi rustls bağımlılığınız, rumqttc'nin kullandığıyla aynı major sürümde olmalıdır — cargo tree -i rustls ile kontrol edin — aksi hâlde tipler uyuşmaz.

MqttOptions::new'a verdiğiniz host adı, rustls'in sertifikayla karşılaştırdığı addır. Bir DNS adı için verilmiş sertifikaya IP adresiyle bağlanmak doğrulamada başarısız olur; broker'ın sunduğu sertifikayı incelemek için hostname mismatch ve TLS bağlantı testi yazılarına bakın.

WebSocket ve MQTT 5

websocket feature'ı etkinken (cargo add rumqttc --features websocket), rumqttc MQTT'yi WebSocket üzerinden tünelleyebilir; yalnızca HTTPS'e izin veren proxy'lerin arkasında işe yarar. Tam URL'yi host olarak verin ve taşımayı seçin:

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

MQTT 5 için crate kökü yerine rumqttc::v5'ten import edin. İstemci ve event loop aynı şekilde çalışır; paketler property'ler (user property, message expiry, response topic) taşır ve onaylar, reason code başvuru aracında arayabileceğiniz MQTT 5 reason code'larını içerir.

Karşılaşacağınız hatalar

eventloop.poll() bir ConnectionError döndürür. Varyant çoğu zaman doğrudan nedeni gösterir:

Sonuncusu pek çok kişiyi şaşırtır: rumqttc temkinli bir azami paket boyutuyla gelir (güncel sürümlerde 10 KiB); büyük JSON payload'lar veya görseller için açık bir sınır gerekir. Bir mesajın kablo üzerinde gerçekte ne kadar yer kapladığını görmek için paket boyutu hesaplayıcısını kullanın.

İstemciyi test etmek

Bir rumqttc istemcisine güvenmeden önce hızlı bir kontrol listesi:

1. Gidiş-dönüş. Online istemciden yayın yapın ve Rust programının mesajı yazdırdığını doğrulayın; sonra tersini deneyin.
2. Ağ kesintisi. Wi-Fi'ı keep-alive süresinden daha uzun kapatın, tekrar açın ve yeni bir "bağlandı" satırı görüp mesajların yeniden aktığını kontrol edin; bu, ConnAck'te yeniden aboneliğin çalıştığının kanıtıdır.
3. Hatalı kimlik bilgileri. Bir kez yanlış parolayla çalıştırın ve kodunuzun sessizce dönmek yerine reddi logladığını doğrulayın.
4. Büyük payload. 10 KiB'tan büyük bir mesaj yayınlayın ve paket boyutu ayarınızın doğru olduğunu teyit edin.

Daha geniş bir bakış için MQTT yeniden bağlanma en iyi uygulamaları ve en iyi MQTT kütüphaneleri karşılaştırmasına bakın; Eclipse Paho C kütüphanesine bağlamaları tercih ediyorsanız Rust ve paho-mqtt ile MQTT alternatifi anlatıyor.

Topic'i online istemcide açın

rumqttc küçük, baştan async ve saf Rust'tır: bir task event loop'u poll etsin, her ConnAck'te abone olun, hatalardan sonra bekleyin ve TLS'i Transport üzerinden ayarlayın. Bu dört alışkanlıkla aynı istemci dizüstü testinden üretimdeki cihaz filosuna kadar çalışır.

Sıkça sorulan sorular