Rust ve paho-mqtt ile MQTT: async istemci ve MQTT 5

paho-mqtt crate'i, Rust için resmi Eclipse Paho istemcisidir. Paho C kütüphanesinin etrafında güvenli bir sarmalayıcıdır; yani pek çok C ve gömülü projede kullanılan, sahada kendini kanıtlamış protokol motorunu, üstünde Rust'a uygun bir async API ile elde edersiniz. MQTT 3.1.1 ve MQTT 5, TCP, TLS ve WebSocket destekler. Bu rehber derlemeyi, eksiksiz bir async MQTT 5 istemcisini, yeniden bağlanmayı, TLS'i ve en sık karşılaşacağınız hataları ele alıyor.

C bağımlılığından kaçınmak isterseniz rumqttc saf Rust alternatifidir.

Derleme gereksinimleri

Crate varsayılan olarak Paho C'nin paketlenmiş bir kopyasını CMake ile derler ve TLS için OpenSSL ile birlikte statik olarak bağlar. Önce araç zincirini kurun:

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

En yeni sürüm için crates.io'ya bakın; burada gösterilen API son sürümlerde kararlı kaldı. İlk derleme C kütüphanesi derlendiği için biraz daha uzun sürer; sonrasında diğer crate'ler gibi önbelleğe alınır.

Async bir MQTT 5 istemcisi

Desen şöyledir: istemciyi oluşturun, bağlanmadan önce mesaj stream'ini alın, bağlanın, abone olun ve stream'i okuyun. Stream'de bir None, bağlantının koptuğu anlamına gelir.

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

    // Hiçbir mesajı kaçırmamak için stream'i bağlanmadan önce alın.
    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!("{} adresine MQTT v{} ile bağlandı", conn.server_uri, conn.mqtt_version);
    }
    cli.subscribe(TOPIC, mqtt::QOS_1).await?;

    // Ayrı bir task'tan beş mesaj yayınlayın.
    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!("publish başarısız: {e}");
            }
            tokio::time::sleep(Duration::from_secs(1)).await;
        }
    });

    while let Some(msg_opt) = stream.next().await {
        match msg_opt {
            Some(msg) => println!("alındı {}: {}", msg.topic(), msg.payload_str()),
            None => {
                eprintln!("bağlantı koptu, yeniden bağlanılıyor");
                while let Err(e) = cli.reconnect().await {
                    eprintln!("yeniden bağlanma başarısız: {e}");
                    tokio::time::sleep(Duration::from_secs(2)).await;
                }
                cli.subscribe(TOPIC, mqtt::QOS_1).await?;
            }
        }
    }
    Ok(())
}

cargo run ile çalıştırın. Önce MQTT v5 içeren bağlantı satırını, ardından geri gelen beş mesajı görmelisiniz. Online MQTT istemcisini açıp testmqtt/rust-paho/# topic'ine abone olduğunuzda aynı mesajları görürsünüz; tarayıcıdan bir mesaj yayınlarsanız Rust programı onu da yazdırır.

Önemli noktalar:

• Oluşturma seçeneklerinde mqtt_version(MQTT_VERSION_5), bağlantı seçeneklerinde new_v5(): MQTT 5 bağlantısı için ikisi de gereklidir.
• Future'lar runtime'dan bağımsızdır. Paho C kendi ağ thread'lerini çalıştırır; token'ları Tokio, async-std ya da basit bir block_on içinden await edebilirsiniz.
• QoS 1 ile publish, PUBACK geldiğinde tamamlanır; buradaki bir hata, broker'ın mesajı kabul etmediği anlamına gelir.
• Clean start kullanıyorsanız reconnect() sonrasında yeniden abone olun, çünkü broker eski oturumu silmiştir.

Otomatik yeniden bağlanma ve kalıcı oturumlar

Elle yazılan döngü yerine yeniden bağlanmayı Paho'ya bırakıp durumu bir callback ile geri yükleyebilirsiniz:

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!("(yeniden) bağlandı"));
cli.set_connection_lost_callback(|_cli| eprintln!("bağlantı koptu"));

clean_start(false) ve bir session expiry aralığıyla broker, siz çevrimdışıyken aboneliklerinizi korur ve QoS 1 mesajlarını kuyruğa alır; yeniden abone olmanız gerekmez. Paylaşılan test uç noktasının çevrimdışı oturumları saklamadığını unutmayın; bunu kendi özel broker örneğinizde deneyin. Broker'ın neyi ne kadar süre sakladığını kalıcı oturumlar yazısı anlatıyor.

Paho, istemci tarafında yoldaki kendi QoS 1/2 mesajlarını da kalıcı hâle getirebilir. Davranışı oluşturma seçeneklerinde .persistence(...) ile açıkça seçin: mqtt::PersistenceType::None her şeyi bellekte tutar, mqtt::PersistenceType::File ise süreç yeniden başlasa bile kaybolmaması için diske yazar. Çalışma dizininde client ID'nizin adını taşıyan klasörler belirirse, bu dosya kalıcılığının işidir.

TLS, kimlik bilgileri ve WebSocket

URI'yi 8883 portunda ssl:// (güncel Paho C derlemelerinde mqtts://) olarak değiştirin ve SSL seçeneklerini ekleyin:

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()
    // Özel bir CA için: .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();

Trust store verilmezse OpenSSL'in varsayılan sertifika konumları kullanılır. Herkese açık sertifikalı bir broker'da doğrulama başarısız olursa trust_store'u sistem paketine yönlendirin (örneğin Debian'da /etc/ssl/certs/ca-certificates.crt). İstemci sertifikaları için key_store ve private_key kullanın. TLS bağlantı testi, kodu suçlamadan önce zinciri openssl s_client ile nasıl inceleyeceğinizi gösterir.

Yalnızca HTTPS'e izin veren ağlarda, aynı SSL seçenekleriyle sunucu URI'si olarak wss://broker.connectmq.com:8084/mqtt kullanın.

Göreceğiniz hatalar

Herhangi bir MQTT 5 kodunu reason code başvuru aracında arayabilirsiniz. Broker'ınızda session expiry, user property'ler ve topic alias'ları düzenli biçimde denemek için MQTT 5 özelliklerini test etmek yazısına bakın.

Test kontrol listesi

1. Gidiş-dönüş: online istemciden yayınlayın, Rust'ta alın; sonra tersini yapın.
2. Ağ kablosunu çekin ya da Wi-Fi'ı keep-alive süresinden uzun kapatın; yeniden bağlanma yolunun çalıştığını ve mesajların yeniden geldiğini doğrulayın.
3. Programınızı özel bir broker'a karşı clean_start(false) ile yeniden başlatın ve kapalıyken yayınlanan mesajların ardından geldiğini doğrulayın.
4. Bir kez hatalı kimlik bilgileriyle çalıştırın ve hatanın açıkça loglandığını kontrol edin.

Demo topic'ini tarayıcıda izleyin

paho-mqtt, bir C derleme adımı karşılığında Rust programlarına MQTT 5, TLS ve WebSocket destekli Eclipse Paho motorunu getirir. Stream'i bağlanmadan önce alın, abonelikleri nasıl geri yükleyeceğinize karar verin ve bir kez bağlantı kopmasını test edin; ardından istemci dizüstü bilgisayarınızdan üretime kadar aynı şekilde davranır. C ile de çalışıyorsanız Paho C rehberi alttaki kütüphaneyi gösteriyor.

Sıkça sorulan sorular