JoeAllen
96865c0588
- 消息缓冲 Vec → VecDeque：原来溢出时 drain(0..n) 每条消息 O(n) 搬移，高频消息下产生大量拷贝；环形缓冲均摊 O(1)，并按配置上限 预分配避免反复扩容 - connection_id/topic/payload 改 Arc<str>：引擎侧主题驻留 （10k 上限，超出整体重建），同主题数千条记录只存一份； 消息列表每帧克隆最多 300 条用于渲染，payload 共享后零拷贝 - 单条 payload 历史保留上限 256KB（retained_payload，UTF-8 边界安全截断），超大报文不再长期占据环形缓冲；消息行显示 「已截断」徽章；发布记录改用实际发送字节（修 Base64/Hex 历史内容与真实报文不一致的隐患） - live 测试适配 Arc<str>；补截断边界单元测试（22 个 lib 测试 + 2 个 live 测试通过）
115 lines
3.5 KiB
Rust
115 lines
3.5 KiB
Rust
//! 真实 MQTT 引擎连通性测试：连接公共 broker，订阅并验证消息回环。
|
|
//! 默认忽略；设置 MQTTX_LIVE=1 并加 --ignored 运行（依赖外网）。
|
|
|
|
use std::time::{Duration, Instant};
|
|
|
|
use mqttx_desktop::model::{
|
|
ConnectionConfig, ConnectionStatus, ProtocolVersion, PublishParams, Subscription, TransportKind,
|
|
};
|
|
use mqttx_desktop::mqtt::{EngineEvent, MqttEngine};
|
|
|
|
#[test]
|
|
#[ignore = "需外网：MQTTX_LIVE=1 cargo test --test engine_live -- --ignored"]
|
|
fn v5_tcp_pubsub_roundtrip() {
|
|
let (tx, rx) = smol::channel::unbounded();
|
|
let engine = MqttEngine::new(tx);
|
|
|
|
let mut cfg = ConnectionConfig::new();
|
|
cfg.name = "live-v5".into();
|
|
cfg.host = "broker.emqx.io".into();
|
|
cfg.port = 1883;
|
|
cfg.transport = TransportKind::Tcp;
|
|
cfg.protocol = ProtocolVersion::V5;
|
|
cfg.auto_reconnect = false;
|
|
cfg.auto_resubscribe = false;
|
|
let cid = cfg.id.clone();
|
|
let topic = format!("mqttx/gpui-test/{}/{}", cid, uuid::Uuid::new_v4());
|
|
|
|
engine.connect(cfg);
|
|
|
|
assert!(wait_connected(&rx), "未能在 15 秒内连接 broker.emqx.io");
|
|
|
|
engine.subscribe(cid.clone(), Subscription::new(cid.clone(), topic.clone(), 0));
|
|
assert!(wait_subscribe(&rx, &topic), "订阅未确认");
|
|
|
|
// 等 broker 传播订阅状态
|
|
std::thread::sleep(Duration::from_millis(500));
|
|
|
|
engine.publish(
|
|
cid.clone(),
|
|
PublishParams {
|
|
topic: topic.clone(),
|
|
payload: "hello-gpui".into(),
|
|
qos: 0,
|
|
..Default::default()
|
|
},
|
|
);
|
|
|
|
assert!(wait_message(&rx, &topic), "未收到自己发布的消息（回环失败）");
|
|
}
|
|
|
|
#[test]
|
|
#[ignore = "需外网：MQTTX_LIVE=1 cargo test --test engine_live -- --ignored"]
|
|
fn v311_tcp_connect() {
|
|
let (tx, rx) = smol::channel::unbounded();
|
|
let engine = MqttEngine::new(tx);
|
|
|
|
let mut cfg = ConnectionConfig::new();
|
|
cfg.host = "broker.emqx.io".into();
|
|
cfg.port = 1883;
|
|
cfg.protocol = ProtocolVersion::V311;
|
|
cfg.auto_reconnect = false;
|
|
cfg.auto_resubscribe = false;
|
|
|
|
engine.connect(cfg);
|
|
assert!(wait_connected(&rx), "MQTT 3.1.1 连接失败");
|
|
}
|
|
|
|
fn pump<F: FnMut(&EngineEvent) -> bool>(
|
|
rx: &smol::channel::Receiver<EngineEvent>,
|
|
timeout: Duration,
|
|
mut pred: F,
|
|
) -> bool {
|
|
let deadline = Instant::now() + timeout;
|
|
while Instant::now() < deadline {
|
|
match rx.try_recv() {
|
|
Ok(ev) => {
|
|
if pred(&ev) {
|
|
return true;
|
|
}
|
|
}
|
|
Err(_) => std::thread::sleep(Duration::from_millis(30)),
|
|
}
|
|
}
|
|
false
|
|
}
|
|
|
|
fn wait_connected(rx: &smol::channel::Receiver<EngineEvent>) -> bool {
|
|
pump(rx, Duration::from_secs(15), |ev| match ev {
|
|
EngineEvent::Status {
|
|
status: ConnectionStatus::Connected,
|
|
..
|
|
} => true,
|
|
EngineEvent::Status {
|
|
status: ConnectionStatus::Error,
|
|
error,
|
|
..
|
|
} => panic!("连接错误: {error:?}"),
|
|
_ => false,
|
|
})
|
|
}
|
|
|
|
fn wait_subscribe(rx: &smol::channel::Receiver<EngineEvent>, topic: &str) -> bool {
|
|
pump(rx, Duration::from_secs(8), |ev| match ev {
|
|
EngineEvent::SubscribeResult { topic: t, ok, .. } => t == topic && *ok,
|
|
_ => false,
|
|
})
|
|
}
|
|
|
|
fn wait_message(rx: &smol::channel::Receiver<EngineEvent>, topic: &str) -> bool {
|
|
pump(rx, Duration::from_secs(10), |ev| match ev {
|
|
EngineEvent::Message(m) => &*m.topic == topic && &*m.payload == "hello-gpui",
|
|
_ => false,
|
|
})
|
|
}
|