简单的MQTTv5客户端

gmqtt-client 是 MQTT 5.0 协议的纯 Rust 实现。

主要功能

• 自动重连
• 重连后重新订阅主题
• 消息队列支持消息过期间隔和保留标志

消息队列

当从 MQTT 代理断开连接时，消息将在 RAM 中保留在队列中。如果客户端从 MQTT 代理断开连接，消息将在 RAM 队列中保留。

具有 消息过期间隔 的消息将倒计时并在过期时丢弃。例如，如果消息以 message_expiry_interval = 60 (秒) 发送，并且客户端在 15 秒后重新连接，则消息将以 message_expiry_interval = 45 发送到代理。

具有 保留标志 且主题相同的消息将在队列中替换。仅保留最后一条保留消息并发送到代理。

这两个功能有助于在客户端长时间离线时减少内存使用。

基本示例

use std::time::Duration;
use tokio::runtime;
use gmqtt_client::{MqttClientBuilder, QoS};
use url::Url;
use serde::Serialize;

#[derive(Serialize)]
struct Event {
    id: u32,
}

fn main() -> anyhow::Result<()> {
    let (mqtt_client, mqtt_worker) = MqttClientBuilder::new(Url::parse("tcp://127.0.0.1:1883")?)
        .on_message_owned_callback(|message, instant| {
            println!("Message: {:?} received {} us ago", message, instant.elapsed().as_micros());
        })
        .subscribe("client/example/receive", QoS::AtLeastOnce)
        .build();

    let worker_runtime = runtime::Builder::new_multi_thread()
        .enable_all()
        .worker_threads(1)
        .build()?;

    worker_runtime.spawn(async { mqtt_worker.run().await });

    let payload = Event {
        id: 10
    };

    mqtt_client.publish_json(
        "client/example/publish",
        &payload,
        false,
        QoS::AtLeastOnce,
        Some(Duration::from_secs(10)),
    )?;

    std::thread::park();
    Ok(())
}

运行示例

sudo apt-get install mosquitto mosquitto-clients

# Run example client
cargo run --example client

# Listen to what is being published
mosquitto_sub -t 'client/example/publish' --pretty -F '%J'

# Send a message over to the client
mosquitto_pub -t 'client/example/receive' -m "hello" -q 1 -V 5 -r -D publish message-expiry-interval 5

# Send 1000 messages
for i in {0..1000}; do mosquitto_pub -t 'client/example/receive' -m "$i" -q 1 -V 5 -r -D publish message-expiry-interval 5; echo "sending $i"; done