十七、Rust集成MQTT Client

1、信息整理

目前了解到的Rust MQTT项目有：

bytebeamio/rumqtt
- 1.3k star、717 commits、Contributors 78、
- tokio、futures、tls、
- rumqttc（client）：cargo add rumqttc
  - https://github.com/bytebeamio/rumqtt/tree/main/rumqttc
- rumqttd（server）：docker run -p 1883:1883 -p 1884:1884 -it bytebeamio/rumqttd
ntex-rs/ntex-mqtt、crates/ntex-mqtt
- 258 star、504 commits、Contributors 15
- ntex、MQTT Client/Server、v5 and v3.1.1 protocols
eclipse/paho.mqtt.rust
- 463 star、368 commits、
- 异步、SSL/TLS、cargo add paho-mqtt = "0.12"、
- 但看自述文档所述，用于 musl linux 环境，有代码侵入。
```
fn is_musl() -> bool {std::env::var("CARGO_CFG_TARGET_ENV").unwrap() == "musl"
}
```
rmqtt/rmqtt：355 star，871 commits
- 基于 ( tokio、ntex、ntex-mqtt )
- 一个纯服务端，遍观项目，貌似不能做 client 使用。自身的测试，使用 paho.mqtt.testing 项目完成。
- 分布式集群（raft），TLS，WebSocket，WebSocket-TLS，x86/Arm
- 单个服务节点上处理百万级别的并发客户端。集群提供相同的连接量、吞吐量，应该只是解决高可用问题。

2、开始集成

本次集成作为 Client 使用，选择了最前面的 bytebeamio/rumqtt 。

boot/mqtt.rs

use std::sync::Arc;
use std::time::Duration;use once_cell::sync::OnceCell;
use rumqttc::{AsyncClient, MqttOptions};
use serde::{Deserialize, Serialize};pub static MQTT_CLIENT: OnceCell<Arc<AsyncClient>> = OnceCell::new();#[derive(Debug, Default, Validate, Serialize, Deserialize)]
pub struct MqttMessage {pub addr: Option<String>,#[validate(range(min = 0, max = 1))]pub action: u8,
}pub async fn start() {let mut mqtt_options = MqttOptions::new("rumqtt-async", "192.168.1.110", 1883);mqtt_options.set_keep_alive(Duration::from_secs(5));let (client, mut event_loop) = AsyncClient::new(mqtt_options, 100);MQTT_CLIENT.get_or_init(|| { Arc::new(client) });tokio::spawn(async move {loop { while let Ok(notification) = event_loop.poll().await {}; }});log::info!("this point run ...");
}

boot/mod.rs

pub mod mqtt;pub async fn start() {// xxx others initializationmqtt::start().await;
}

modules/mod.rs

pub mod switch;pub mod handler {use actix_web::dev::HttpServiceFactory;use actix_web::web;use crate::module::{switch};pub fn api_routes() -> impl HttpServiceFactory {web::scope("").service(switch::api::index)}
}

modules/switch/api.rs

use rumqttc::QoS::AtLeastOnce;
use validator::Validate;use crate::boot::mqtt::{MQTT_CLIENT, MqttMessage};#[post("/switch/{id}")]
pub async fn index(Path(id): Path<String>, mut msg: Json<MqttMessage>) -> impl Responder {let ok = vec!["Hello World!", "Hello World!"];if let Err(e) = msg.validate() { return HttpResponse::BadRequest().json(e); }msg.addr = Some(id);let json = serde_json::to_string(&msg).unwrap();MQTT_CLIENT.get().unwrap().publish("/hello", AtLeastOnce, false, json).await.unwrap();HttpResponse::Ok().json(ok)
}

main.rs

use actix_web::{App, HttpServer, middleware};
use booking::{boot, module};#[actix_web::main]
async fn main() -> std::io::Result<()> {boot::start().await;HttpServer::new(move || App::new()// TODO other initialization.service(module::handler::api_routes())// other routes).bind("0.0.0.0:8080")?.run().await
}