coordinate-connector 架构设计

1. 系统概述

coordinate-connector 是 Coordinate 消息系统中的 MQTT 客户端 SDK，基于rumqttc进行二次开发，采用 Rust 异步运行时（tokio）实现。它为应用程序提供简洁的 API 用于连接 MQTT Broker、发布消息和订阅主题。作为 coordinate-broadcast 的配套客户端库，coordinate-connector 主要用于 coordinate-server 与广播组件之间的通信，同时也支持外部应用接入。

从功能定位角度来看，coordinate-connector 主要承担三类职责：第一是作为 MQTT 客户端，封装底层协议细节，提供简洁的异步 API；第二是作为状态管理器，维护连接状态、QoS 流程控制和 packet id 分配；第三是作为传输层抽象，支持 TCP、TLS、WebSocket 等多种传输方式。系统设计遵循高性能、低延迟的原则，使用预分配数据结构和事件驱动模式，整体能够支撑高并发的消息处理。

2. 系统架构

2.1 整体架构图

2.2 模块职责划分

coordinate-connector 采用模块化的架构设计，每个模块承担独立的职责，通过清晰定义的接口进行模块间交互。这种设计使得各模块可以独立演进，同时便于测试和维护。系统主要包含以下五个核心模块：

client 模块负责提供高层的异步客户端抽象，是应用程序使用的主要入口点。它封装了请求通道（Channel），提供了 publish、subscribe、ack、disconnect 等简洁的异步方法。client 模块的设计遵循简单易用原则，隐藏了底层的事件循环细节，应用程序只需关注业务逻辑。

eventloop 模块是系统的核心事件循环引擎，负责驱动整个客户端的运行。它管理网络连接、处理入站和出站数据包、维护 Keep-alive 机制。EventLoop 使用 tokio 的 select 模式实现异步事件处理，能够高效地处理并发请求和网络 I/O。

state 模块负责维护 MQTT 连接状态，包括飞行中消息（inflight）管理、packet id 分配、QoS 流程控制、会话恢复等。state 模块是 MQTT 协议实现的核心支持，确保消息传递的可靠性。

protocol 模块实现了 MQTT 协议的解析与序列化功能。该模块根据 Cargo feature 决定使用不同协议版本：

• v5 Extended（包含 v0 feature）：用于 coordinate-server 连接 broadcast，在内网使用，是 MQTT v5 完整协议并扩展 AddSubscribe/RemoveSubscribe 消息类型，支持动态订阅管理
• v5 Lite（默认）：用于外部客户端连接 broadcast 接收消息，是 MQTT v5 的裁剪集合，仅包含连接管理消息（Connect/ConnAck/Ping/Disconnect）

v5 Extended 扩展功能：

• AddSubscribe：服务端代客户端订阅主题
• RemoveSubscribe：服务端代客户端取消订阅

transport 模块负责底层网络传输，支持 TCP、TCP+TLS、WebSocket、WebSocket+TLS 四种传输方式。传输层抽象使得客户端可以灵活适应不同的网络环境。

3. 核心组件设计

3.1 MqttOptions 配置

MqttOptions 是客户端连接的核心配置类，提供了丰富的配置选项：

pubstructMqttOptions{broker_addr:String,// Broker 地址port:u16,// 端口transport:Transport,// 传输层类型keep_alive:Duration,// 保活间隔clean_start:bool,// 清理会话标志client_id:String,// 客户端标识credentials:Option<Login>,// 认证凭据last_will:Option<LastWill>,// 遗嘱消息// ... 更多选项}

配置示例：

letmutoptions=MqttOptions::new("client-id","broker-host",21884);options.set_keep_alive(Duration::from_secs(30)).set_clean_start(true).set_credentials("user","password").set_last_will(LastWill::new("will-topic","msg",QoS::AtLeastOnce,false,None));

3.2 Client 客户端

Client 是面向应用程序的高层 API，提供了简洁的异步接口：

pubstructClient{request_tx:Sender<Request>,}// 创建客户端和事件循环let(client,muteventloop)=Client::new(options,cap);// 消息循环loop{matcheventloop.poll().await{Ok(Event::Incoming(Incoming::Publish(publish)))=>{// 处理接收到的消息}Ok(Event::Outgoing(_))=>{/* 发送确认 */}Err(e)=>break,}}

3.3 EventLoop 事件循环

EventLoop 是系统的核心引擎，管理连接和事件处理：

pubstructEventLoop{puboptions:MqttOptions,pubstate:MqttState,requests_rx:Receiver<Request>,pending:VecDeque<Request>,network:Option<Network>,keepalive_timeout:Option<Pin<Box<Sleep>>>,}implEventLoop{pubasyncfnpoll(&mutself)->Result<Event,ConnectionError>{// 连接建立、数据收发、Keep-alive 处理}}

3.4 MqttState 状态管理

MqttState 维护 MQTT 连接状态：

pubstructMqttState{pubawait_pingresp:bool,publast_pkid:u16,pubinflight:u16,puboutgoing_pub:Vec<Option<Publish>>,pubincoming_pub:FixedBitSet,pubevents:VecDeque<Event>,pubmanual_acks:bool,pubbroker_topic_alias_max:u16,pubmax_outgoing_inflight:u16,}

4. 关键技术实现

4.1 客户端创建与连接

// 创建客户端和事件循环let(client,muteventloop)=Client::new(options,10000);// 使用方式loop{matcheventloop.poll().await{Ok(Event::Incoming(Incoming::Publish(publish)))=>{println!("Received: {:?}",publish.topic);}Ok(Event::Outgoing(_))=>{}Err(e)=>break,}}

4.2 消息发布

// 异步发布client.publish("topic/test",QoS::AtLeastOnce,false,"payload").await?;// 非阻塞发布client.try_publish("topic",QoS::AtMostOnce,true,"data")?;

4.3 消息订阅

// 订阅单个主题client.subscribe("home/+/temperature",QoS::AtLeastOnce).await?;// 批量订阅client.subscribe_many(vec![Filter::new("topic1",QoS::AtMostOnce),Filter::new("topic2",QoS::AtLeastOnce),]).await?;// 非阻塞订阅client.try_subscribe("topic",QoS::AtMostOnce)?;

4.4 传输层

系统支持四种传输层类型：

pubenumTransport{Tcp,// 普通 TCPTls(TlsConfiguration),// TLS 加密Ws,// WebSocketWss(TlsConfiguration),// WebSocket + TLS}

4.5 TLS 配置

// 使用默认配置Transport::tls_with_default_config()// 自定义配置Transport::tls(ca:Vec<u8>,client_auth:Option<(Vec<u8>,Vec<u8>)>,alpn:Option<Vec<Vec<u8>>>,)

5. 与 coordinate-broadcast 的交互

5.1 连接配置

coordinate-server 通过 coordinate-connector 连接 broadcast 服务：

[broadcast] host = "192.168.31.195" port = 21884 username = "" password = "" endpoint = "ws://192.168.31.195:8000"

5.2 连接流程

asyncfnconnect(options:&mutMqttOptions)->Result<(Network,ConnAck),ConnectionError>{// 1. 建立网络连接let(network,connack)=timeout(Duration::from_secs(self.options.connection_timeout()),connect(&mutself.options),).await??;// 2. MQTT 握手network.write(Packet::Connect(...)).await?;network.flush().await?;matchnetwork.read().await?{Incoming::ConnAck(connack)ifconnack.code==Success=>Ok(connack),_=>Err(ConnectionError::ConnectionRefused),}}

5.3 消息流

coordinate-server → Client.publish() → EventLoop → MqttState → Network.write() → coordinate-broadcast coordinate-broadcast → Network.read() → EventLoop.poll() → Event::Incoming → 应用程序

6. 配置设计

6.1 Cargo.toml Features

[features] default = [] use-rustls = ["use-rustls-no-provider", "tokio-rustls/default"] use-rustls-no-provider = ["dep:tokio-rustls", "dep:rustls-webpki"] use-native-tls = ["dep:tokio-native-tls", "dep:native-tls"] websocket = ["dep:async-tungstenite", "dep:ws_stream_tungstenite"] v0 = []

6.2 MqttOptions 配置项

6.3 NetworkOptions 配置项

pubstructNetworkOptions{tcp_send_buffer_size:Option<u32>,tcp_recv_buffer_size:Option<u32>,tcp_nodelay:bool,conn_timeout:u64,}

7. 设计模式总结

7.1 架构模式

系统采用以下架构模式实现高性能和高可用性：

生产者-消费者模式：Client 作为生产者，将请求放入通道；EventLoop 作为消费者，从通道取出请求进行处理。通道提供了高效的异步通信机制。

事件驱动模式：EventLoop 基于 tokio 的 select 模式实现异步事件处理，能够高效地处理并发请求和网络 I/O。

状态机模式：MqttState 作为状态机，管理 MQTT 连接的各种状态，包括连接中、已连接、断开等。

7.2 扩展性设计

系统提供了良好的扩展性支持：通过 Cargo features 可以选择不同的协议版本和传输层；通过 TlsConfiguration 可以自定义 TLS 行为；通过 MqttOptions 可以灵活配置连接参数。

8. 技术规格

9. 使用示例

9.1 基本使用

usecoordinate_connector::{Client,Event,MqttOptions,QoS};usestd::time::Duration;#[tokio::main]asyncfnmain()->Result<(),Box<dynstd::error::Error>>{letmutoptions=MqttOptions::new("test-client","localhost",21884);options.set_keep_alive(Duration::from_secs(30));options.set_clean_start(true);let(client,muteventloop)=Client::new(options,10000);// 订阅主题client.subscribe("home/temperature",QoS::AtLeastOnce).await?;// 消息循环loop{tokio::select!{event=eventloop.poll()=>{matchevent?{Event::Incoming(coordinate_connector::Incoming::Publish(publish)))=>{println!("Received: {}",publish.topic);}_=>{}}}}}}

9.2 发布消息

// 发布消息client.publish("home/temperature",QoS::AtLeastOnce,false,"25.5").await?;// 批量订阅client.subscribe_many(vec![coordinate_connector::Filter::new("home/temperature",QoS::AtLeastOnce),coordinate_connector::Filter::new("home/humidity",QoS::AtLeastOnce),]).await?;