From ab13f485e001d0509a7c043053aebd56af8b3f10 Mon Sep 17 00:00:00 2001 From: JoeAllen <2494136784@qq.com> Date: Tue, 29 Sep 2026 10:33:59 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20MQTT=20=E8=BF=9E=E6=8E=A5=E7=94=9F?= =?UTF-8?q?=E5=91=BD=E5=91=A8=E6=9C=9F=E7=AB=9E=E6=80=81=E4=B8=8E=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 连接句柄（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 --- gpui-app/src/model.rs | 98 +++++++++++--- gpui-app/src/mqtt.rs | 205 ++++++++++++++++++++++------- gpui-app/src/ui/app.rs | 179 +++++++++++++++++-------- gpui-app/src/ui/connection_view.rs | 38 ++++-- 4 files changed, 395 insertions(+), 125 deletions(-) diff --git a/gpui-app/src/model.rs b/gpui-app/src/model.rs index e781a0d..36a1d73 100644 --- a/gpui-app/src/model.rs +++ b/gpui-app/src/model.rs @@ -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, /// payload 超过 [`MAX_PAYLOAD_RETAIN`] 被截断 pub payload_truncated: bool, + /// 非文本负载的原始字节（合法 UTF-8 时为 None）。文本化会失真 + /// （非法字节被替换为 U+FFFD），详情面板的 Hex/Base64 需要真实报文。 + pub raw_bytes: Option>, } /// 消息历史中单条 payload 的保留上限。超限截断，避免个别大报文 @@ -442,20 +456,27 @@ pub struct MqttRecord { pub const MAX_PAYLOAD_RETAIN: usize = 256 * 1024; /// 把原始 payload 字节转成用于历史记录的共享文本，超限截断。 -/// 返回 (Arc, 是否截断)。截断时回退到 UTF-8 字符边界。 -pub fn retained_payload(raw: &[u8]) -> (Arc, 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, Option>, 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 = 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 = "中".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 配置改动应触发重连提示"); } } diff --git a/gpui-app/src/mqtt.rs b/gpui-app/src/mqtt.rs index b37c94b..75fbb62 100644 --- a/gpui-app/src/mqtt.rs +++ b/gpui-app/src/mqtt.rs @@ -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, user_properties: Vec<(String, String)>, + response_topic: Option, + correlation_data: Option, + message_expiry_interval: Option, }, /// 订阅请求结果（tracked，等待 SUBACK） SubscribeResult { @@ -69,14 +73,20 @@ enum ClientKind { } struct ConnHandle { - client: ClientKind, + /// `None` 表示仍在建立中（占位）。建连完成后由 connect 的任务升级为 Some。 + client: Option, cancel_tx: watch::Sender, connected: Arc, + /// 连接代数：同一 id 的第几次 connect。事件循环退出清理时只在 + /// map 中仍是自己这一代时才移除，避免误删重连后的新句柄。 + generation: u64, } pub struct MqttEngine { runtime: tokio::runtime::Runtime, conns: Mutex>, + /// 下一次 connect 使用的代数 + next_generation: AtomicU64, /// 主题/连接 ID 驻留缓存：同一连接内主题高度重复，驻留让消息记录共享字符串 topics: Mutex>>, 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)))) - } else { - build_v4(&cfg) - .map(|(client, el)| (ClientKind::V4(client), EventLoopKind::V4(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))) + }) + } + }) + .await; + let build_error = |engine: &Arc, id: &str, generation: u64, e: String| { + engine.remove_handle_if(id, generation); + engine.emit(EngineEvent::Status { + 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); + }; let (client, eventloop) = match build { - Ok(pair) => pair, - Err(e) => { - engine.emit(EngineEvent::Status { - connection_id: id.clone(), - status: ConnectionStatus::Error, - error: Some(e.clone()), - reason_code: None, - session_present: None, - }); - engine.log(&id, LogLevel::Error, "connect_error", format!("连接失败: {e}"), None); + 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(), - }; - engine - .conns - .lock() - .unwrap() - .insert(id.clone(), handle); + // 建连期间该连接可能已被 close 或更新的 connect 顶掉： + // 仅当占位句柄仍是本代时才升级为可用客户端，否则丢弃本次建连。 + { + let mut map = match engine.conns.lock() { + Ok(m) => m, + Err(_) => return, + }; + 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 { 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, connected: Arc, + 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, connected: Arc, + 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); }); } diff --git a/gpui-app/src/ui/app.rs b/gpui-app/src/ui/app.rs index 69b5847..1f1c4ae 100644 --- a/gpui-app/src/ui/app.rs +++ b/gpui-app/src/ui/app.rs @@ -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)>, /// 系统外观变化订阅：设置为「跟随系统」时重应用主题，随实体存活 _appearance_obs: gpui_kit::Subscription, } @@ -137,49 +137,85 @@ 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 { - weak.update_in(cx, |_app, window, cx| { + 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("更新就绪") - .child( - v_flex().gap_2().child( - div().text_sm().child(format!( - "新版本 v{} 已下载完成（已通过 sha256 校验）。", - staged.version - )), - ), - ) - .footer( - h_flex() - .gap_2() - .justify_end() - .w_full() - .child( - Button::new("upd-later") - .label("下次启动安装") - .outline() - .on_click(|_, window, cx| { - window.close_dialog(cx) - }), - ) - .child( - Button::new("upd-now") - .label("立即安装") - .primary() - .on_click(move |_, window, cx| { - window.close_dialog(cx); - if update::install_staged(&staged).is_ok() { - cx.quit(); - } - }), + .w(px(440.)) + .title("更新就绪") + .child( + v_flex().gap_2().child( + div().text_sm().child(format!( + "新版本 v{} 已下载完成（已通过 sha256 校验）。", + staged.version + )), ), - ) - }); - }) - .ok(); + ) + .footer( + h_flex() + .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(move |_, window, cx| { + // 同意标记：启动时据此自动安装； + // 用户跳过对话框则不会强制升级 + update::mark_install_consent(&ver); + window.close_dialog(cx); + }), + ) + .child( + Button::new("upd-now") + .label("立即安装") + .primary() + .on_click(move |_, window, cx| { + window.close_dialog(cx); + if let Err(e) = + update::install_staged(&staged) + { + window.push_notification( + Notification::error(format!( + "安装失败：{e}" + )), + cx, + ); + } else { + cx.quit(); + } + }), + ), + ) + }); + }) + .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,13 +405,29 @@ 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 { - window.push_notification( - Notification::error(format!("订阅失败: {e}")), - cx, - ); + // 乐观更新已在发送前入库；这里只在订阅仍存在时同步 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, diff --git a/gpui-app/src/ui/connection_view.rs b/gpui-app/src/ui/connection_view.rs index 7e9494e..bb92f19 100644 --- a/gpui-app/src/ui/connection_view.rs +++ b/gpui-app/src/ui/connection_view.rs @@ -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, Vec, usize) = self + let (records, subs, matched_total, total): ( + Vec, + Vec, + usize, + usize, + ) = self .with_app(cx, |app| { let subs: Vec = app .subscriptions @@ -1545,6 +1553,7 @@ impl ConnectionView { .collect(); let mut matched = 0usize; let mut records: Vec = 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 { } /// 展开详情里的 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::(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::>() .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)