JoeAllen/mqttx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: MQTT 连接生命周期竞态与事件处理

Browse Source 
连接句柄（mqtt.rs）：
- ConnHandle 增加代数 generation，事件循环退出只在仍是自己这一代时
  才移除句柄——旧循环无条件按 id 删除会把重连后的新句柄删掉，
  导致连接显示在线却无法从 UI 断开、发布/订阅报连接不存在
- connect() 同步插入占位句柄（client=None），建连期间也可被取消；
  建连完成后仅当句柄仍是本代才升级为可用客户端
- 重连 2 秒退避改为可取消的 select：原先 close() 的取消信号在退避
  期间不可见，会对用户已关闭的连接继续真实重连
- 重连成功后清零计数（max_reconnect_times 限流的是连续失败次数）；
  v4 CONNACK 拒绝上报真实返回码；test_handshake 断开后再 poll 冲出
  DISCONNECT 报文；建连的证书文件读取移入 spawn_blocking

消息记录与事件（model.rs / app.rs / connection_view.rs）：
- 非文本负载保留原始字节（MqttRecord.raw_bytes，仅非法 UTF-8 时
  占用，纯文本零开销），详情 Hex/Base64 不再展示 lossy 失真字节
- EngineEvent::Published 补齐 response_topic/correlation_data/
  message_expiry；correlation_data 两侧统一「优先文本、回退 hex」
- 事件入口加连接存在性守卫：删除连接后排队的引擎事件不再复活
  状态/消息（内存泄漏）或为已删连接写孤儿订阅
- SUBACK 失败时移除乐观插入的订阅行；成功事件只刷新已有订阅，
  不再用默认字段复活 SUBACK 前被删除的订阅
- Connecting 期间侧栏电源按钮防抖；分组名恰为 all/ungrouped 时
  chip 元素 id 改前缀避免共享交互状态
- 「清空」按钮按消息总数而非过滤命中数判断禁用；搜索匹配的
  大小写快速路径限定 ASCII（U+1E9B 等小写映射非自身的字符）
- 应用发布预设后同步模板变量面板，残留行不再写回全局变量
- 日志文件套 BufWriter，高频消息流不再逐条落盘 syscall
This commit is contained in:
JoeAllen
2026-09-29 10:33:59 +08:00
parent 6ff05f7216
commit ab13f485e0
4 changed files with 395 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gpui-app/src/model.rs
+83 -15
View File
@@ -84,7 +84,7 @@ impl TransportKind {
/// MQTT 5 遗嘱消息（3.1.1 下忽略 v5 专属属性）。
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct LastWill {
pub topic: String,
pub payload: String,
@@ -97,7 +97,7 @@ pub struct LastWill {
}
/// SSL/TLS 自定义配置（CA、双向认证、跳过校验）。
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct SslConfig {
/// 自定义 CA 文件路径（PEM），作为系统根证书的补充
#[serde(default)]
@@ -250,15 +250,21 @@ impl ConnectionConfig {
/// 与另一份配置相比，影响 broker 会话建立的关键参数是否变化。
/// 这些字段改动只有在重新建立连接后才会生效，用于提示用户重连。
/// WS 路径、遗嘱、TLS 配置、keep_alive 同样随 CONNECT 报文一次性生效，
/// 必须纳入比较。
pub fn session_params_changed(&self, other: &ConnectionConfig) -> bool {
self.host != other.host
|| self.port != other.port
|| self.path != other.path
|| self.transport != other.transport
|| self.protocol != other.protocol
|| self.client_id != other.client_id
|| self.username != other.username
|| self.password != other.password
|| self.clean_start != other.clean_start
|| self.keep_alive != other.keep_alive
|| self.last_will != other.last_will
|| self.ssl != other.ssl
}
}
@@ -398,6 +404,11 @@ impl PayloadFormat {
if !cleaned.len().is_multiple_of(2) {
return Err("Hex 长度必须为偶数（每两个字符一个字节）".into());
}
// 必须先做全字节校验：长度检查按字节计数，多字节字符（如中文）
// 可凑成偶数字节，直接按字节索引切片会落在字符边界内而 panic。
if !cleaned.bytes().all(|b| b.is_ascii_hexdigit()) {
return Err("Hex 解码失败: 含非十六进制字符".into());
}
(0..cleaned.len())
.step_by(2)
.map(|i| {
@@ -435,6 +446,9 @@ pub struct MqttRecord {
pub subscription_identifier: Option<u32>,
/// payload 超过 [`MAX_PAYLOAD_RETAIN`] 被截断
pub payload_truncated: bool,
/// 非文本负载的原始字节（合法 UTF-8 时为 None）。文本化会失真
/// （非法字节被替换为 U+FFFD），详情面板的 Hex/Base64 需要真实报文。
pub raw_bytes: Option<Arc<[u8]>>,
}
/// 消息历史中单条 payload 的保留上限。超限截断，避免个别大报文
@@ -442,20 +456,27 @@ pub struct MqttRecord {
pub const MAX_PAYLOAD_RETAIN: usize = 256 * 1024;
/// 把原始 payload 字节转成用于历史记录的共享文本，超限截断。
/// 返回 (Arc<str>, 是否截断)。截断时回退到 UTF-8 字符边界。
pub fn retained_payload(raw: &[u8]) -> (Arc<str>, bool) {
if raw.len() <= MAX_PAYLOAD_RETAIN {
(Arc::from(String::from_utf8_lossy(raw).as_ref()), false)
} else {
/// 返回 (文本, 非文本时的原始字节, 是否截断)。截断时回退到 UTF-8 字符边界。
///
/// 仅当文本化会失真（非合法 UTF-8，字节被替换为 U+FFFD）时才保留原始
/// 字节（同样受 [`MAX_PAYLOAD_RETAIN`] 限制）；纯文本负载零额外开销。
pub fn retained_payload(raw: &[u8]) -> (Arc<str>, Option<Arc<[u8]>>, bool) {
let truncated = raw.len() > MAX_PAYLOAD_RETAIN;
let text_part = if truncated {
let mut end = MAX_PAYLOAD_RETAIN;
while end > 0 && (raw[end] & 0xC0) == 0x80 {
end -= 1;
}
(
Arc::from(String::from_utf8_lossy(&raw[..end]).as_ref()),
true,
)
}
&raw[..end]
} else {
raw
};
let raw_bytes = match std::str::from_utf8(text_part) {
Ok(_) => None,
Err(_) => Some(Arc::from(text_part.to_vec())),
};
let text: Arc<str> = Arc::from(String::from_utf8_lossy(text_part).as_ref());
(text, raw_bytes, truncated)
}
// ─── 发布参数 / 发布预设 ──────────────────────────────────────────────────────
@@ -897,13 +918,14 @@ mod tests {
#[test]
fn retained_payload_caps_and_preserves_utf8() {
let small = b"hello";
let (p, trunc) = retained_payload(small);
let (p, raw, trunc) = retained_payload(small);
assert_eq!(&*p, "hello");
assert!(raw.is_none(), "合法 UTF-8 不保留原始字节");
assert!(!trunc);
// 中文每字符 3 字节，让上限落在字符中间（256K 不是 3 的倍数）
let big: Vec<u8> = "中".repeat(MAX_PAYLOAD_RETAIN).into_bytes();
let (p, trunc) = retained_payload(&big);
let (p, _, trunc) = retained_payload(&big);
assert!(trunc, "超过上限必须截断");
assert!(p.len() <= MAX_PAYLOAD_RETAIN);
assert_eq!(p.chars().last(), Some('中'), "不能切坏 UTF-8 字符");
@@ -914,7 +936,53 @@ mod tests {
// 恰好上限不截断
let exact = vec![b'a'; MAX_PAYLOAD_RETAIN];
let (_, trunc) = retained_payload(&exact);
let (_, raw, trunc) = retained_payload(&exact);
assert!(!trunc);
assert!(raw.is_none());
}
// 非文本负载：保留原始字节供 Hex/Base64 详情，文本侧 lossy 展示
#[test]
fn retained_payload_keeps_raw_bytes_for_binary() {
let binary = vec![0x00, 0xff, 0xfe, 0x01];
let (p, raw, trunc) = retained_payload(&binary);
assert!(!trunc);
assert_eq!(&*p, String::from_utf8_lossy(&binary));
let raw = raw.expect("非法 UTF-8 必须保留原始字节");
assert_eq!(&*raw, &binary);
}
// Hex 编码：多字节字符不得 panic（曾按字节切片落入字符边界内而崩溃）
#[test]
fn hex_encode_rejects_non_hex_without_panic() {
assert!(PayloadFormat::Hex.encode("deadBEEF").is_ok());
assert!(PayloadFormat::Hex.encode("de ad be ef").is_ok());
assert!(PayloadFormat::Hex.encode("abc").is_err(), "奇数长度报错");
assert!(PayloadFormat::Hex.encode("zz").is_err(), "非法字符报错");
// 偶数字节长度的多字节字符：修复前此处在字符边界内切片 panic
assert!(PayloadFormat::Hex.encode("中文").is_err());
assert!(PayloadFormat::Hex.encode("😀").is_err());
}
// 会话参数比较必须覆盖所有随 CONNECT 报文一次性生效的字段
#[test]
fn session_params_changed_covers_connect_time_fields() {
let a = ConnectionConfig::default();
let mut b = a.clone();
assert!(!a.session_params_changed(&b));
b.keep_alive = a.keep_alive + 1;
assert!(a.session_params_changed(&b), "keep_alive 改动应触发重连提示");
b = a.clone();
b.last_will = Some(LastWill {
topic: "w/t".into(),
..Default::default()
});
assert!(a.session_params_changed(&b), "遗嘱改动应触发重连提示");
b = a.clone();
b.ssl.ca_file = "/tmp/ca.pem".into();
assert!(a.session_params_changed(&b), "TLS 配置改动应触发重连提示");
}
}
gpui-app/src/mqtt.rs
+150 -41
View File
@@ -38,6 +38,7 @@ pub enum EngineEvent {
/// 收到消息
Message(MqttRecord),
/// 一条本地发出的消息（已成功交给客户端）。payload 为实际发送的字节。
/// 属性与接收记录对齐，本地发送记录可完整回显。
Published {
connection_id: String,
topic: String,
@@ -46,6 +47,9 @@ pub enum EngineEvent {
retain: bool,
content_type: Option<String>,
user_properties: Vec<(String, String)>,
response_topic: Option<String>,
correlation_data: Option<String>,
message_expiry_interval: Option<u32>,
},
/// 订阅请求结果（tracked，等待 SUBACK）
SubscribeResult {
@@ -69,14 +73,20 @@ enum ClientKind {
}
struct ConnHandle {
client: ClientKind,
/// `None` 表示仍在建立中（占位）。建连完成后由 connect 的任务升级为 Some。
client: Option<ClientKind>,
cancel_tx: watch::Sender<bool>,
connected: Arc<AtomicBool>,
/// 连接代数：同一 id 的第几次 connect。事件循环退出清理时只在
/// map 中仍是自己这一代时才移除，避免误删重连后的新句柄。
generation: u64,
}
pub struct MqttEngine {
runtime: tokio::runtime::Runtime,
conns: Mutex<HashMap<String, ConnHandle>>,
/// 下一次 connect 使用的代数
next_generation: AtomicU64,
/// 主题/连接 ID 驻留缓存：同一连接内主题高度重复，驻留让消息记录共享字符串
topics: Mutex<HashMap<String, Arc<str>>>,
tx: EventSender,
@@ -98,6 +108,7 @@ impl MqttEngine {
Arc::new(Self {
runtime,
conns: Mutex::new(HashMap::new()),
next_generation: AtomicU64::new(1),
topics: Mutex::new(HashMap::new()),
tx,
seq: Arc::new(AtomicU64::new(1)),
@@ -139,6 +150,21 @@ impl MqttEngine {
self.close(&cfg.id, false);
let id = cfg.id.clone();
let generation = self.next_generation.fetch_add(1, Ordering::Relaxed) + 1;
let (cancel_tx, cancel_rx) = watch::channel(false);
let connected = Arc::new(AtomicBool::new(false));
// 同步占位（client=None）：建连期间 close() 也能取消本次连接，
// 且 is_connected() 在建连完成前保持 false。
self.conns.lock().unwrap().insert(
id.clone(),
ConnHandle {
client: None,
cancel_tx,
connected: connected.clone(),
generation,
},
);
self.emit(EngineEvent::Status {
connection_id: id.clone(),
status: ConnectionStatus::Connecting,
@@ -162,41 +188,58 @@ impl MqttEngine {
let engine = Arc::clone(self);
self.runtime.spawn(async move {
let build = if cfg.protocol.is_v5() {
build_v5(&cfg)
.map(|(client, el)| (ClientKind::V5(client), EventLoopKind::V5(Box::new(el))))
let auto_reconnect = cfg.auto_reconnect;
let max_reconnect_times = cfg.max_reconnect_times;
// build 含证书/CA 文件读取等阻塞 IO，放 spawn_blocking，
// 避免慢盘上卡住 2 线程 runtime 里其他连接的调度
let build = tokio::task::spawn_blocking(move || {
if cfg.protocol.is_v5() {
build_v5(&cfg).map(|(client, el)| {
(ClientKind::V5(client), EventLoopKind::V5(Box::new(el)))
})
} else {
build_v4(&cfg)
.map(|(client, el)| (ClientKind::V4(client), EventLoopKind::V4(Box::new(el))))
};
build_v4(&cfg).map(|(client, el)| {
(ClientKind::V4(client), EventLoopKind::V4(Box::new(el)))
})
}
})
.await;
let (client, eventloop) = match build {
Ok(pair) => pair,
Err(e) => {
let build_error = |engine: &Arc<MqttEngine>, id: &str, generation: u64, e: String| {
engine.remove_handle_if(id, generation);
engine.emit(EngineEvent::Status {
connection_id: id.clone(),
connection_id: id.to_string(),
status: ConnectionStatus::Error,
error: Some(e.clone()),
reason_code: None,
session_present: None,
});
engine.log(&id, LogLevel::Error, "connect_error", format!("连接失败: {e}"), None);
engine.log(id, LogLevel::Error, "connect_error", format!("连接失败: {e}"), None);
};
let (client, eventloop) = match build {
Ok(Ok(pair)) => pair,
Ok(Err(e)) => {
build_error(&engine, &id, generation, e);
return;
}
Err(j) => {
build_error(&engine, &id, generation, format!("内部任务失败: {j}"));
return;
}
};
let connected = Arc::new(AtomicBool::new(false));
let (cancel_tx, cancel_rx) = watch::channel(false);
let handle = ConnHandle {
client,
cancel_tx,
connected: connected.clone(),
// 建连期间该连接可能已被 close 或更新的 connect 顶掉：
// 仅当占位句柄仍是本代时才升级为可用客户端，否则丢弃本次建连。
{
let mut map = match engine.conns.lock() {
Ok(m) => m,
Err(_) => return,
};
engine
.conns
.lock()
.unwrap()
.insert(id.clone(), handle);
match map.get_mut(&id) {
Some(h) if h.generation == generation => h.client = Some(client),
_ => return,
}
}
match eventloop {
EventLoopKind::V5(el) => {
@@ -206,8 +249,9 @@ impl MqttEngine {
*el,
cancel_rx,
connected,
cfg.auto_reconnect,
cfg.max_reconnect_times,
generation,
auto_reconnect,
max_reconnect_times,
);
}
EventLoopKind::V4(el) => {
@@ -217,8 +261,9 @@ impl MqttEngine {
*el,
cancel_rx,
connected,
cfg.auto_reconnect,
cfg.max_reconnect_times,
generation,
auto_reconnect,
max_reconnect_times,
);
}
}
@@ -231,15 +276,16 @@ impl MqttEngine {
if let Some(h) = handle {
let _ = h.cancel_tx.send(true);
let connected = h.connected.clone();
// 发送 MQTT DISCONNECT 报文后丢弃客户端
// 发送 MQTT DISCONNECT 报文后丢弃客户端（建连未完成时无需发送）
self.runtime.spawn(async move {
match h.client {
ClientKind::V5(c) => {
Some(ClientKind::V5(c)) => {
let _ = c.disconnect().await;
}
ClientKind::V4(c) => {
Some(ClientKind::V4(c)) => {
let _ = c.disconnect().await;
}
None => {}
}
connected.store(false, Ordering::SeqCst);
});
@@ -328,6 +374,9 @@ impl MqttEngine {
retain: params.retain,
content_type: params.content_type,
user_properties: params.user_properties,
response_topic: params.response_topic,
correlation_data: params.correlation_data,
message_expiry_interval: params.message_expiry_interval,
});
}
Err(e) => {
@@ -537,14 +586,26 @@ impl MqttEngine {
/// 克隆一个连接的异步客户端用于发命令。
/// 不持有 std Mutex 守卫跨 await，避免与 tokio runtime 的 Send 要求冲突。
/// 占位句柄（建连未完成）视为不存在。
fn client_of(&self, id: &str) -> Option<ClientKind> {
let map = self.conns.lock().ok()?;
let h = map.get(id)?;
Some(match &h.client {
let client = map.get(id)?.client.as_ref()?;
Some(match client {
ClientKind::V5(c) => ClientKind::V5(c.clone()),
ClientKind::V4(c) => ClientKind::V4(c.clone()),
})
}
/// 仅当 map 中该 id 的句柄仍是 `generation` 这一代时才移除。
/// 事件循环退出清理必须走这里：无条件按 id 删除会把重连后的
/// 新句柄误删，导致连接无法再从 UI 断开。
fn remove_handle_if(&self, id: &str, generation: u64) {
if let Ok(mut map) = self.conns.lock()
&& map.get(id).map(|h| h.generation) == Some(generation)
{
map.remove(id);
}
}
}
enum EventLoopKind {
@@ -927,8 +988,15 @@ async fn test_handshake(cfg_in: &ConnectionConfig) -> Result<(), String> {
match eventloop.poll().await {
Ok(V5Event::Incoming(V5Packet::ConnAck(ack))) => {
if ack.code == V5ReturnCode::Success {
// 测试完成，主动断开避免在 broker 侧残留会话
// 测试完成，主动断开避免在 broker 侧残留会话。
// disconnect() 只是把请求入队，需再 poll 才会把
// DISCONNECT 真正写上网络。
let _ = client.disconnect().await;
for _ in 0..3 {
if eventloop.poll().await.is_err() {
break;
}
}
return Ok(());
}
return Err(format!("连接被拒绝: {:?}", ack.code));
@@ -946,7 +1014,13 @@ async fn test_handshake(cfg_in: &ConnectionConfig) -> Result<(), String> {
match eventloop.poll().await {
Ok(rv4::Event::Incoming(rv4::Packet::ConnAck(ack))) => {
if ack.code == rv4::ConnectReturnCode::Success {
// 同 v5：disconnect() 只入队，需 poll 把报文冲出
let _ = client.disconnect().await;
for _ in 0..3 {
if eventloop.poll().await.is_err() {
break;
}
}
return Ok(());
}
return Err(format!("连接被拒绝: {:?}", ack.code));
@@ -965,7 +1039,7 @@ async fn test_handshake(cfg_in: &ConnectionConfig) -> Result<(), String> {
fn v5_record(engine: &MqttEngine, seq: u64, conn_id: &str, p: &V5Publish) -> MqttRecord {
let props = p.properties.as_ref();
let (payload, payload_truncated) = crate::model::retained_payload(&p.payload);
let (payload, raw_bytes, payload_truncated) = crate::model::retained_payload(&p.payload);
MqttRecord {
seq,
connection_id: engine.intern(conn_id),
@@ -983,12 +1057,23 @@ fn v5_record(engine: &MqttEngine, seq: u64, conn_id: &str, p: &V5Publish) -> Mqt
correlation_data: props.and_then(|p| {
p.correlation_data
.as_ref()
.map(|d| d.iter().map(|b| format!("{b:02x}")).collect())
.map(|d| correlation_display(d))
}),
message_expiry_interval: props.and_then(|p| p.message_expiry_interval),
subscription_identifier: props
.and_then(|p| p.subscription_identifiers.first().map(|v| *v as u32)),
payload_truncated,
raw_bytes,
}
}
/// correlation data 的可读表示：优先按 UTF-8 文本展示，非文本回退 hex。
/// 发送端把字符串按字节原样发出，两侧展示保持对称（此前收端一律 hex、
/// 发端保留原文，同一条数据两侧对不上）。
fn correlation_display(data: &[u8]) -> String {
match std::str::from_utf8(data) {
Ok(s) => s.to_string(),
Err(_) => data.iter().map(|b| format!("{b:02x}")).collect(),
}
}
@@ -1006,9 +1091,11 @@ fn spawn_v5_loop(
mut eventloop: V5EventLoop,
mut cancel_rx: watch::Receiver<bool>,
connected: Arc<AtomicBool>,
generation: u64,
auto_reconnect: bool,
max_reconnect_times: u32,
) {
#![allow(clippy::too_many_arguments)]
let runtime = engine.runtime.handle().clone();
runtime.spawn(async move {
let mut first_connect = true;
@@ -1035,6 +1122,9 @@ fn spawn_v5_loop(
connected.store(ack.code == V5ReturnCode::Success, Ordering::SeqCst);
if ack.code == V5ReturnCode::Success {
first_connect = false;
// 重连成功后清零计数：max_reconnect_times 限流的是
// 「连续失败次数」，不能把历史错误累计进来
reconnect_attempts = 0;
engine.log(
&id,
LogLevel::Info,
@@ -1148,7 +1238,12 @@ fn spawn_v5_loop(
reconnect_attempts += 1;
engine.log(&id, LogLevel::Warn, "connection_error",
format!("连接错误，2 秒后重连（第 {reconnect_attempts} 次）: {e}"), None);
tokio::time::sleep(Duration::from_secs(2)).await;
// 退避等待也必须可取消：否则 close() 的取消信号最长
// 2 秒不可见，期间会对用户已关闭的连接继续重连。
tokio::select! {
_ = cancel_rx.changed() => break,
_ = tokio::time::sleep(Duration::from_secs(2)) => {}
}
}
}
}
@@ -1158,7 +1253,8 @@ fn spawn_v5_loop(
}
}
}
engine.conns.lock().ok().and_then(|mut m| m.remove(&id));
// 仅当 map 中仍是本代句柄时才移除，避免误删重连后的新句柄
engine.remove_handle_if(&id, generation);
});
}
@@ -1168,9 +1264,11 @@ fn spawn_v4_loop(
mut eventloop: rv4::EventLoop,
mut cancel_rx: watch::Receiver<bool>,
connected: Arc<AtomicBool>,
generation: u64,
auto_reconnect: bool,
max_reconnect_times: u32,
) {
#![allow(clippy::too_many_arguments)]
let runtime = engine.runtime.handle().clone();
runtime.spawn(async move {
let mut first_connect = true;
@@ -1181,7 +1279,7 @@ fn spawn_v4_loop(
event = eventloop.poll() => {
match event {
Ok(rv4::Event::Incoming(rv4::Packet::Publish(p))) => {
let (payload, payload_truncated) =
let (payload, raw_bytes, payload_truncated) =
crate::model::retained_payload(&p.payload);
let record = MqttRecord {
seq: engine.next_seq(),
@@ -1203,6 +1301,7 @@ fn spawn_v4_loop(
message_expiry_interval: None,
subscription_identifier: None,
payload_truncated,
raw_bytes,
};
engine.log(
&id, LogLevel::Info, "publish_received",
@@ -1213,6 +1312,9 @@ fn spawn_v4_loop(
connected.store(ack.code == rv4::ConnectReturnCode::Success, Ordering::SeqCst);
if ack.code == rv4::ConnectReturnCode::Success {
first_connect = false;
// 重连成功后清零计数：max_reconnect_times 限流的是
// 「连续失败次数」，不能把历史错误累计进来
reconnect_attempts = 0;
engine.log(&id, LogLevel::Info, "connack",
format!("连接成功 (session_present: {})", ack.session_present), None);
engine.emit(EngineEvent::Status {
@@ -1229,7 +1331,8 @@ fn spawn_v4_loop(
connection_id: id.clone(),
status: ConnectionStatus::Error,
error: Some(format!("{:?}", ack.code)),
reason_code: Some(1),
// 上报真实的 CONNACK 拒绝码，不再硬编码为 1
reason_code: Some(ack.code as u8),
session_present: Some(ack.session_present),
});
break;
@@ -1294,7 +1397,12 @@ fn spawn_v4_loop(
reconnect_attempts += 1;
engine.log(&id, LogLevel::Warn, "connection_error",
format!("连接错误，2 秒后重连（第 {reconnect_attempts} 次）: {e}"), None);
tokio::time::sleep(Duration::from_secs(2)).await;
// 退避等待也必须可取消：否则 close() 的取消信号最长
// 2 秒不可见，期间会对用户已关闭的连接继续重连。
tokio::select! {
_ = cancel_rx.changed() => break,
_ = tokio::time::sleep(Duration::from_secs(2)) => {}
}
}
}
}
@@ -1304,6 +1412,7 @@ fn spawn_v4_loop(
}
}
}
engine.conns.lock().ok().and_then(|mut m| m.remove(&id));
// 仅当 map 中仍是本代句柄时才移除，避免误删重连后的新句柄
engine.remove_handle_if(&id, generation);
});
}
gpui-app/src/ui/app.rs
+88 -13
View File
@@ -100,7 +100,7 @@ pub struct MqttXApp {
/// 原生文件对话框进行中标记（对话框阻塞后台线程，防止重复弹出）
file_dialog_open: bool,
/// 按日期缓存的日志文件句柄（日期, 文件），避免每条日志都重新打开文件
log_file: Option<(String, std::fs::File)>,
log_file: Option<(String, std::io::BufWriter<std::fs::File>)>,
/// 系统外观变化订阅：设置为「跟随系统」时重应用主题，随实体存活
_appearance_obs: gpui_kit::Subscription,
}
@@ -137,10 +137,13 @@ impl MqttXApp {
let rx = engine.run_blocking(update::check_and_download);
let weak = cx.entity().downgrade();
cx.spawn_in(window, async move |_this, cx: &mut gpui_kit::AsyncWindowContext| {
if let Ok(Ok(Some(staged))) = rx.recv().await {
match rx.recv().await {
Ok(Ok(Some(staged))) => {
weak.update_in(cx, |_app, window, cx| {
window.open_dialog(cx, move |dialog, _, _cx| {
let staged = staged.clone();
let staged_skip = staged.clone();
let ver = staged.version.clone();
dialog
.w(px(440.))
.title("更新就绪")
@@ -157,12 +160,26 @@ impl MqttXApp {
.gap_2()
.justify_end()
.w_full()
.child(
Button::new("upd-skip")
.label("暂不更新")
.outline()
.on_click(move |_, window, cx| {
// 拒绝本次更新：删除暂存包，
// 下次启动不会被自动安装
update::discard_staged(&staged_skip);
window.close_dialog(cx);
}),
)
.child(
Button::new("upd-later")
.label("下次启动安装")
.outline()
.on_click(|_, window, cx| {
window.close_dialog(cx)
.on_click(move |_, window, cx| {
// 同意标记：启动时据此自动安装；
// 用户跳过对话框则不会强制升级
update::mark_install_consent(&ver);
window.close_dialog(cx);
}),
)
.child(
@@ -171,7 +188,16 @@ impl MqttXApp {
.primary()
.on_click(move |_, window, cx| {
window.close_dialog(cx);
if update::install_staged(&staged).is_ok() {
if let Err(e) =
update::install_staged(&staged)
{
window.push_notification(
Notification::error(format!(
"安装失败：{e}"
)),
cx,
);
} else {
cx.quit();
}
}),
@@ -181,6 +207,16 @@ impl MqttXApp {
})
.ok();
}
// 检查/下载失败不打扰用户，但也不能完全无声
Ok(Err(e)) => {
eprintln!("[app] 启动更新检查失败: {e}");
sentry::capture_message(
&format!("startup update check failed: {e}"),
sentry::Level::Warning,
);
}
_ => {}
}
})
.detach();
}
@@ -289,6 +325,12 @@ impl MqttXApp {
error,
..
} => {
// 引擎 close 与事件循环退出之间有窗口，删除连接后其排队
// 事件仍会送达：为已删除的连接重建状态/消息属于泄漏，
// 一律丢弃（删除时已主动 close）。
if !self.connections.iter().any(|c| c.id == connection_id) {
return;
}
self.statuses.insert(connection_id.clone(), status);
match error {
Some(e) => {
@@ -308,6 +350,9 @@ impl MqttXApp {
}
EngineEvent::Message(mut record) => {
let id = record.connection_id.clone();
if !self.connections.iter().any(|c| c.id == id.as_ref()) {
return;
}
record.seq = self.next_seq();
self.push_message(id.clone(), record);
if let Some(view) = self.views.get(id.as_ref()) {
@@ -322,8 +367,14 @@ impl MqttXApp {
retain,
content_type,
user_properties,
response_topic,
correlation_data,
message_expiry_interval,
} => {
let (payload, payload_truncated) = retained_payload(&payload);
if !self.connections.iter().any(|c| c.id == connection_id) {
return;
}
let (payload, raw_bytes, payload_truncated) = retained_payload(&payload);
let record = MqttRecord {
seq: self.next_seq(),
connection_id: Arc::from(connection_id.as_str()),
@@ -335,11 +386,12 @@ impl MqttXApp {
timestamp: chrono::Local::now().timestamp_millis(),
user_properties,
content_type,
response_topic: None,
correlation_data: None,
message_expiry_interval: None,
response_topic,
correlation_data,
message_expiry_interval,
subscription_identifier: None,
payload_truncated,
raw_bytes,
};
self.push_message(Arc::from(connection_id.as_str()), record);
if let Some(view) = self.views.get(connection_id.as_str()) {
@@ -353,14 +405,30 @@ impl MqttXApp {
ok,
error,
} => {
if !self.connections.iter().any(|c| c.id == connection_id) {
return;
}
if ok {
self.add_subscription(Subscription::new(connection_id.clone(), topic, qos));
} else if let Some(e) = error {
// 乐观更新已在发送前入库；这里只在订阅仍存在时同步 QoS，
// 避免 SUBACK 前被删除的订阅被此事件用默认字段复活。
if let Some(existing) = self.subscriptions.iter_mut().find(|s| {
s.connection_id == connection_id && s.topic == topic
}) {
existing.qos = qos;
}
} else {
if let Some(e) = error {
window.push_notification(
Notification::error(format!("订阅失败: {e}")),
cx,
);
}
// broker 拒绝订阅：移除乐观插入的订阅行，与引擎状态保持一致
// （不发 UNSUBSCRIBE——本就未订阅成功）
self.subscriptions
.retain(|s| !(s.connection_id == connection_id && s.topic == topic));
self.storage.save_subscriptions(&self.subscriptions);
}
if let Some(view) = self.views.get(&connection_id) {
view.update(cx, |_, cx| cx.notify());
}
@@ -448,6 +516,8 @@ impl MqttXApp {
}
/// 打开（或复用按日缓存的）日志文件并追加一行。
/// 文件套 BufWriter：高频消息流下每条日志不再触发多次落盘 syscall；
/// 跨日重开/出错关句柄时随 drop 自动 flush。
fn append_log_line(&mut self, entry: &LogEntry) -> std::io::Result<()> {
use std::io::Write as _;
use chrono::TimeZone as _;
@@ -469,7 +539,7 @@ impl MqttXApp {
.create(true)
.append(true)
.open(path)?;
self.log_file = Some((day, file));
self.log_file = Some((day, std::io::BufWriter::new(file)));
}
// 连接名查不到时回退为 id，保证行格式恒定
let conn = self
@@ -547,6 +617,11 @@ impl MqttXApp {
}
pub fn toggle_connection(&self, cfg: &ConnectionConfig) {
// Connecting 期间 is_connected 恒为 false：防抖，双击/误触不会
// 并发两次 connect（会互相顶掉句柄并在 broker 侧形成重连风暴）
if self.statuses.get(&cfg.id) == Some(&ConnectionStatus::Connecting) {
return;
}
if self.engine.is_connected(&cfg.id) {
self.engine.close(&cfg.id, true);
} else {
@@ -1273,7 +1348,7 @@ impl MqttXApp {
.filter(|c| c.group.as_deref().map(str::trim) == Some(name.as_str()))
.count();
row = row.child(self.render_group_chip(
SharedString::from(format!("chip-{}", name)),
SharedString::from(format!("chip-g-{name}")),
format!("{name} {count}"),
GroupFilter::Named(name),
cx,
gpui-app/src/ui/connection_view.rs
+28 -10
View File
@@ -699,6 +699,9 @@ impl ConnectionView {
s.set_selected_index(Some(IndexPath::new(fmt_idx)), window, cx)
});
self.retain = params.retain;
// 同步模板变量面板：预设替换了模板，旧占位符行必须随之清理，
// 否则残留行仍会写回全局变量
self.sync_var_rows(window, cx);
cx.notify();
}
}
@@ -1352,7 +1355,7 @@ impl ConnectionView {
if expanded {
let fmt = self.detail_format;
let detail_text = format_payload_detail(&record.payload, fmt);
let detail_text = format_payload_detail(&record.payload, record.raw_bytes.as_deref(), fmt);
let mut details = v_flex().gap_1p5().pt_1();
// 格式切换 + 复制按钮；stop_propagation 避免点击时又收起整行
@@ -1535,7 +1538,12 @@ impl ConnectionView {
let dir = self.msg_dir;
let sub_filter = self.sub_filter.clone();
// 引用过滤 + 计数命中总数，仅克隆前 MAX_RENDERED_MESSAGES 条，避免全量拷贝
let (records, subs, matched_total): (Vec<MqttRecord>, Vec<Subscription>, usize) = self
let (records, subs, matched_total, total): (
Vec<MqttRecord>,
Vec<Subscription>,
usize,
usize,
) = self
.with_app(cx, |app| {
let subs: Vec<Subscription> = app
.subscriptions
@@ -1545,6 +1553,7 @@ impl ConnectionView {
.collect();
let mut matched = 0usize;
let mut records: Vec<MqttRecord> = Vec::new();
let total = app.messages.get(conn.as_str()).map_or(0, |m| m.len());
if let Some(m) = app.messages.get(conn.as_str()) {
for r in m.iter().rev() {
// 方向 / 搜索 / 订阅过滤叠加，过滤后再截断保证最新消息优先
@@ -1567,7 +1576,7 @@ impl ConnectionView {
}
}
}
(records, subs, matched)
(records, subs, matched, total)
})
.unwrap_or_default();
@@ -1722,7 +1731,9 @@ impl ConnectionView {
.child({
let weak = cx.weak_entity();
let clear_id = self.conn_id.clone();
let empty = records.is_empty();
// 清空针对连接的全部消息（非过滤结果）：
// 禁用与否依据总数，过滤命中 0 条不应禁用
let empty = total == 0;
Button::new(SharedString::from(format!("msg-clear-{}", self.conn_id)))
.icon(IconName::Eraser)
.label("清空")
@@ -2577,22 +2588,24 @@ fn pick_subscription_color(subs: &[Subscription], topic: &str) -> Option<f32> {
}
/// 展开详情里的 payload 渲染：自动 = 合法 JSON 美化，否则原文。
fn format_payload_detail(payload: &str, format: DetailFormat) -> String {
/// Hex/Base64 优先使用入库保留的原始字节——非文本负载的 lossy 文本
/// 已把非法字节替换为 U+FFFD，按文本字节展示会与线上报文不符。
fn format_payload_detail(payload: &str, raw: Option<&[u8]>, format: DetailFormat) -> String {
match format {
DetailFormat::Auto => serde_json::from_str::<serde_json::Value>(payload)
.ok()
.and_then(|v| serde_json::to_string_pretty(&v).ok())
.unwrap_or_else(|| payload.to_string()),
DetailFormat::Text => payload.to_string(),
DetailFormat::Hex => payload
.as_bytes()
DetailFormat::Hex => raw
.unwrap_or(payload.as_bytes())
.iter()
.map(|b| format!("{b:02x}"))
.collect::<Vec<_>>()
.join(" "),
DetailFormat::Base64 => {
use base64::Engine as _;
base64::engine::general_purpose::STANDARD.encode(payload.as_bytes())
base64::engine::general_purpose::STANDARD.encode(raw.unwrap_or(payload.as_bytes()))
}
}
}
@@ -2663,8 +2676,13 @@ fn contains_ignore_case(hay: &str, needle: &str) -> bool {
if hay.contains(needle) {
return true;
}
// 原文没有任何大写字符时，小写化不会改变匹配结果，直接判否
if !hay.chars().any(|c| c.is_uppercase()) {
// 纯 ASCII 快速路径：双方均为 ASCII 且原文无 ASCII 大写时，
// 小写化恒等，可直接判否。不能用 is_uppercase 判断——个别
// 小写字符（如 U+1E9B）的小写映射并不等于自身。
if needle.is_ascii()
&& hay.is_ascii()
&& !hay.bytes().any(|b| b.is_ascii_uppercase())
{
return false;
}
hay.to_lowercase().contains(needle)