diff --git a/Cargo.lock b/Cargo.lock index 5daf713..229d22e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4073,8 +4073,11 @@ dependencies = [ "dirs 6.0.0", "gpui-kit", "hmac", + "rfd", "rumqttc-v4-next", "rumqttc-v5-next", + "rustls", + "rustls-native-certs", "serde", "serde_json", "sha1", @@ -5673,6 +5676,33 @@ dependencies = [ "zune-jpeg 0.5.15", ] +[[package]] +name = "rfd" +version = "0.17.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "20dafead71c16a34e1ff357ddefc8afc11e7d51d6d2b9fbd07eaa48e3e540220" +dependencies = [ + "block2 0.6.2", + "dispatch2", + "js-sys", + "libc", + "log", + "objc2 0.6.4", + "objc2-app-kit 0.3.2", + "objc2-core-foundation", + "objc2-foundation 0.3.2", + "percent-encoding", + "pollster 0.4.0", + "raw-window-handle", + "wasm-bindgen", + "wasm-bindgen-futures", + "wayland-backend", + "wayland-client", + "wayland-protocols", + "web-sys", + "windows-sys 0.61.2", +] + [[package]] name = "rgb" version = "0.8.53" diff --git a/README.md b/README.md index 6095a5f..cf6a52a 100644 --- a/README.md +++ b/README.md @@ -9,16 +9,21 @@ - **双协议支持** — MQTT 5.0 与 MQTT 3.1.1（[`rumqttc-next`](https://crates.io/crates/rumqttc-v5-next)） - **多传输方式** — 明文 TCP、TLS、WebSocket(ws)、WebSocket over TLS(wss)，默认 rustls（aws-lc） -- **多连接管理** — 同时管理多个连接，侧边栏实时显示连接状态（未连接/连接中/已连接/错误），多标签页切换 -- **完整 MQTT 5 属性** — 用户属性（可视化键值对编辑器）、Content-Type、消息过期间隔、Response Topic、Correlation Data -- **发布 / 订阅** — QoS 0/1/2、Retain、遗嘱消息（Last Will）、订阅自动恢复（auto resubscribe）、断线自动重连 -- **消息流** — 收发方向区分、时间戳、QoS/Retain 标记、点击展开查看格式化 JSON 与 v5 属性、主题/内容过滤 -- **全局变量** — 发布主题与负载中使用 `{{变量名}}`，内置 `{{$ts}}`、`{{$ts_ms}}`、`{{$uuid}}` +- **多连接管理** — 同时管理多个连接，侧边栏实时显示连接状态（未连接/连接中/已连接/错误），多标签页切换；连接分组（分组 chips 过滤 + 表单分组归属） +- **完整 MQTT 5 属性** — 用户属性（可视化键值对编辑器）、Content-Type、消息过期间隔、Response Topic、Correlation Data（连接/遗嘱/发布三处均支持） +- **发布 / 订阅** — QoS 0/1/2、Retain、遗嘱消息（Last Will）、订阅自动恢复（auto resubscribe）、断线自动重连（可配置最大重连次数与连接超时） +- **订阅增强** — 逗号/换行多主题批量订阅、订阅别名、颜色标签（消息按订阅着色）、启用/禁用、点击订阅过滤消息流、MQTT 5 订阅选项（订阅标识符 / No Local / Retain As Published / Retain Handling） +- **消息流** — 收发方向区分、时间戳、QoS/Retain 标记、点击展开查看格式化 JSON 与 v5 属性、全部/接收/发布过滤、主题/内容搜索、逐条 payload 格式切换（自动/文本/Hex/Base64）、复制主题/负载/详情、清空历史 +- **连接导入 / 导出** — 原生文件对话框（rfd）导出/导入连接 JSON，按 id 去重 +- **SSL/TLS 证书** — 自定义 CA、客户端证书/私钥（双向认证）、忽略 CA 校验 +- **全局变量** — 发布主题与负载中使用 `{{变量名}}`，内置 `{{$ts}}`、`{{$ts_ms}}`、`{{$uuid}}`，用户属性同样参与渲染 - **负载格式** — Plaintext / JSON（发送前校验）/ Base64 / Hex - **阿里云 IoT** — Token（GroupId@@@DeviceId + HMAC-SHA1 签名）与一机一密两种鉴权方式一键生成连接 -- **连接测试** — 表单内 5 秒握手测试，保存前验证连通性 -- **日志面板** — 连接级事件日志（CONNACK/SUBACK/PUBACK/错误/重连等） -- **主题** — 浅色 / 深色切换 +- **连接测试** — 表单内握手测试（超时可配置），保存前验证连通性 +- **日志面板** — 连接级事件日志（CONNACK/SUBACK/PUBACK/错误/重连等），按日落盘 `mqttx-YYYY-MM-DD.log`，设置内一键打开日志目录 +- **快捷键** — `Ctrl+N` 新建连接、`Ctrl+,` 设置、`Ctrl+Shift+E` 导出、`Ctrl+Shift+I` 导入、`Ctrl+Enter` 发送消息 +- **主题** — 浅色 / 深色 / 跟随系统 +- **设置** — 自动检查更新开关、消息缓存条数、时间戳毫秒、数据目录展示与打开、关于页（版本 / 许可） - **本地持久化** — 连接、订阅、预设、变量、设置以 JSON 存于系统数据目录 ## 技术栈 diff --git a/gpui-app/Cargo.toml b/gpui-app/Cargo.toml index 65427e9..7c8f2ce 100644 --- a/gpui-app/Cargo.toml +++ b/gpui-app/Cargo.toml @@ -31,3 +31,11 @@ dirs = "6" hmac = "0.12" sha1 = "0.10" base64 = "0.22" + +# 自定义 TLS（自定义 CA / 双向认证 / 跳过校验），版本与 rumqttc 内部使用的 rustls 一致 +rustls = { version = "0.23", default-features = false, features = ["std", "tls12", "aws_lc_rs"] } +# 加载系统根证书，作为自定义 CA 的补充 +rustls-native-certs = "0.8" + +# 原生文件对话框（连接导入/导出）；Windows 走内建 COM 后端，Linux 用默认 xdg-portal +rfd = "0.17" diff --git a/gpui-app/src/model.rs b/gpui-app/src/model.rs index 5ebdfa3..c3b17ef 100644 --- a/gpui-app/src/model.rs +++ b/gpui-app/src/model.rs @@ -94,6 +94,33 @@ pub struct LastWill { pub response_topic: Option, } +/// SSL/TLS 自定义配置（CA、双向认证、跳过校验）。 +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct SslConfig { + /// 自定义 CA 文件路径（PEM），作为系统根证书的补充 + #[serde(default)] + pub ca_file: String, + /// 客户端证书（PEM），双向认证用 + #[serde(default)] + pub client_cert_file: String, + /// 客户端私钥（PEM），与客户端证书配套 + #[serde(default)] + pub client_key_file: String, + /// 跳过服务器证书校验（insecure），仅调试环境使用 + #[serde(default)] + pub ignore_ca: bool, +} + +impl SslConfig { + /// 是否完全未配置（走 rumqttc 默认 TLS 即可）。 + pub fn is_default(&self) -> bool { + !self.ignore_ca + && self.ca_file.is_empty() + && self.client_cert_file.is_empty() + && self.client_key_file.is_empty() + } +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ConnectionConfig { pub id: String, @@ -144,6 +171,19 @@ pub struct ConnectionConfig { /// 创建时间（unix 秒），用于排序 #[serde(default)] pub created_at: i64, + + /// 连接分组（对标 MQTTX folder） + #[serde(default)] + pub group: Option, + /// 连接超时（秒），覆盖建连与 CONNACK 等待 + #[serde(default = "default_connection_timeout")] + pub connection_timeout_secs: u16, + /// 最大重连次数，0 = 无限重连 + #[serde(default)] + pub max_reconnect_times: u32, + /// SSL/TLS 自定义配置 + #[serde(default)] + pub ssl: SslConfig, } fn default_true() -> bool { @@ -152,6 +192,14 @@ fn default_true() -> bool { fn default_keep_alive() -> u16 { 60 } +fn default_connection_timeout() -> u16 { + 10 +} + +/// 生成随机 Client ID（新建连接与表单「重新生成」共用同一逻辑）。 +pub fn generate_client_id() -> String { + format!("mqttx_{}", &uuid::Uuid::new_v4().simple().to_string()[..8]) +} impl ConnectionConfig { pub fn new() -> Self { @@ -164,7 +212,7 @@ impl ConnectionConfig { path: "/mqtt".into(), protocol: ProtocolVersion::V5, transport, - client_id: format!("mqttx_{}", &uuid::Uuid::new_v4().simple().to_string()[..8]), + client_id: generate_client_id(), username: String::new(), password: String::new(), clean_start: true, @@ -177,6 +225,10 @@ impl ConnectionConfig { auto_reconnect: true, last_will: None, created_at: chrono::Local::now().timestamp(), + group: None, + connection_timeout_secs: default_connection_timeout(), + max_reconnect_times: 0, + ssl: SslConfig::default(), } } @@ -212,6 +264,24 @@ pub struct Subscription { /// 本地备注颜色（与 MQTTX 一致，按订阅给消息着色），0..=359 色相 #[serde(default)] pub color: Option, + /// 订阅别名（对标 MQTTX topic alias），仅用于界面展示 + #[serde(default)] + pub alias: Option, + /// 是否启用该订阅；禁用后界面置灰、恢复订阅时跳过 + #[serde(default = "default_true")] + pub enabled: bool, + /// MQTT 5 订阅标识符 + #[serde(default)] + pub sub_identifier: Option, + /// MQTT 5 No Local (NL) + #[serde(default)] + pub no_local: bool, + /// MQTT 5 Retain As Published (RAP) + #[serde(default)] + pub retain_as_published: bool, + /// MQTT 5 Retain Handling (0/1/2) + #[serde(default)] + pub retain_handling: u8, } impl Subscription { @@ -222,6 +292,38 @@ impl Subscription { topic: topic.into(), qos, color: None, + alias: None, + enabled: true, + sub_identifier: None, + no_local: false, + retain_as_published: false, + retain_handling: 0, + } + } +} + +/// MQTT 5 订阅选项（NL/RAP/Retain Handling/订阅标识符）；v4 连接会忽略这些选项。 +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct SubscribeOptions { + /// 订阅标识符，0 视为未设置 + #[serde(default)] + pub sub_identifier: Option, + #[serde(default)] + pub no_local: bool, + #[serde(default)] + pub retain_as_published: bool, + /// 0=每次订阅都发送保留消息，1=仅新订阅发送，2=不发送 + #[serde(default)] + pub retain_handling: u8, +} + +impl From<&Subscription> for SubscribeOptions { + fn from(sub: &Subscription) -> Self { + Self { + sub_identifier: sub.sub_identifier, + no_local: sub.no_local, + retain_as_published: sub.retain_as_published, + retain_handling: sub.retain_handling, } } } @@ -655,4 +757,42 @@ mod tests { PayloadFormat::Hex ); } + + // 向后兼容：旧版 connections.json 不含 Phase 1 新字段，必须能反序列化且取默认值 + #[test] + fn legacy_connection_json_fills_new_defaults() { + let legacy = r#"{ + "id": "c1", + "name": "旧连接", + "host": "broker.emqx.io", + "port": 1883, + "client_id": "legacy_cid", + "keep_alive": 30 + }"#; + let c: ConnectionConfig = serde_json::from_str(legacy).expect("旧版 JSON 应能反序列化"); + assert_eq!(c.name, "旧连接"); + assert_eq!(c.keep_alive, 30); + assert!(c.group.is_none(), "group 默认应为 None"); + assert_eq!(c.connection_timeout_secs, 10, "连接超时默认 10 秒"); + assert_eq!(c.max_reconnect_times, 0, "最大重连次数默认 0（无限）"); + assert!(c.ssl.ca_file.is_empty()); + assert!(c.ssl.client_cert_file.is_empty()); + assert!(c.ssl.client_key_file.is_empty()); + assert!(!c.ssl.ignore_ca, "ignore_ca 默认关闭"); + } + + // 向后兼容：旧版订阅 JSON 必须默认 enabled=true，其余 v5 选项为关闭/None + #[test] + fn legacy_subscription_json_defaults_enabled() { + let legacy = r#"{"id":"s1","connection_id":"c1","topic":"a/b","qos":1}"#; + let s: Subscription = serde_json::from_str(legacy).expect("旧版订阅应能反序列化"); + assert!(s.enabled, "旧订阅默认启用"); + assert!(s.alias.is_none()); + assert!(s.sub_identifier.is_none()); + assert!(!s.no_local); + assert!(!s.retain_as_published); + assert_eq!(s.retain_handling, 0); + // Subscription::new 同样产出默认启用的订阅 + assert!(Subscription::new("c1", "t", 0).enabled); + } } diff --git a/gpui-app/src/mqtt.rs b/gpui-app/src/mqtt.rs index c0d070a..766c1bd 100644 --- a/gpui-app/src/mqtt.rs +++ b/gpui-app/src/mqtt.rs @@ -4,6 +4,7 @@ use std::collections::HashMap; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; +use std::time::Duration; use rumqttc::mqttbytes::v5::{ ConnectReturnCode as V5ReturnCode, Packet as V5Packet, Publish as V5Publish, @@ -12,14 +13,16 @@ use rumqttc::mqttbytes::v5::{ use rumqttc::{ AsyncClient as V5AsyncClient, Broker as V5Broker, ConnectionError as V5Error, Event as V5Event, EventLoop as V5EventLoop, MqttOptions as V5Options, Outgoing as V5Outgoing, - PublishOptions as V5PublishOptions, QoS as V5QoS, Transport as V5Transport, + PublishOptions as V5PublishOptions, QoS as V5QoS, RetainForwardRule, SubscribeFilterInput, + SubscribeProperties as V5SubscribeProperties, TlsConfiguration, Transport as V5Transport, }; use rumqttc_v4 as rv4; +use rustls::pki_types::{pem::PemObject, CertificateDer, PrivateKeyDer}; use tokio::sync::watch; use crate::model::{ ConnectionConfig, ConnectionStatus, Direction, LogEntry, LogLevel, MqttRecord, PublishParams, - Subscription, TransportKind, + SslConfig, SubscribeOptions, Subscription, TransportKind, };// ─── 引擎事件 ──────────────────────────────────────────────────────────────── #[derive(Debug, Clone)] @@ -201,6 +204,7 @@ impl MqttEngine { cancel_rx, connected, cfg.auto_reconnect, + cfg.max_reconnect_times, ); } EventLoopKind::V4(el) => { @@ -211,6 +215,7 @@ impl MqttEngine { cancel_rx, connected, cfg.auto_reconnect, + cfg.max_reconnect_times, ); } } @@ -335,7 +340,19 @@ impl MqttEngine { // ── 订阅 / 取消订阅 ───────────────────────────────────────────────────── + /// 旧签名薄封装：不携带 v5 订阅选项（连接视图等现有调用点继续可用）。 pub fn subscribe(self: &Arc, connection_id: String, sub: Subscription) { + self.subscribe_with_options(connection_id, sub, SubscribeOptions::default()); + } + + /// 带 MQTT 5 订阅选项的订阅（sub_identifier / NL / RAP / Retain Handling）； + /// v4 连接会忽略这些选项。 + pub fn subscribe_with_options( + self: &Arc, + connection_id: String, + sub: Subscription, + opts: SubscribeOptions, + ) { let engine = Arc::clone(self); self.runtime.spawn(async move { let Some(client) = engine.client_of(&connection_id) else { @@ -357,8 +374,23 @@ impl MqttEngine { ); let result = match &client { ClientKind::V5(client) => { + // NL/RAP/Retain Handling 是「按主题过滤器」的订阅选项， + // 订阅标识符则在订阅报文属性里，分别写入两处。 + let filter = SubscribeFilterInput::new(sub.topic.clone(), v5_qos(sub.qos)) + .no_local(opts.no_local) + .preserve_retain(opts.retain_as_published) + .retain_forward_rule(match opts.retain_handling { + 1 => RetainForwardRule::OnNewSubscribe, + 2 => RetainForwardRule::Never, + _ => RetainForwardRule::OnEverySubscribe, + }); + let properties = V5SubscribeProperties { + // 0 不是合法的订阅标识符，按未设置处理 + id: opts.sub_identifier.filter(|v| *v > 0).map(|v| v as usize), + user_properties: Vec::new(), + }; match client - .subscribe_tracked(sub.topic.clone(), v5_qos(sub.qos)) + .subscribe_many_with_properties_tracked(std::iter::once(filter), properties) .await { Ok(notice) => notice.wait_completion_async().await.map_err(|e| { @@ -441,20 +473,24 @@ impl MqttEngine { }); } - /// 测试连接：5 秒内等待 CONNACK，结果通过 channel 返回。 + /// 测试连接：使用配置的连接超时（上限 30 秒，防止误填过大值长时间挂起）， + /// 结果通过 channel 返回。 pub fn test_connection( self: &Arc, cfg: ConnectionConfig, ) -> smol::channel::Receiver> { let (tx, rx) = smol::channel::bounded(1); + let secs = u64::from(cfg.connection_timeout_secs.clamp(1, 30)); self.runtime.spawn(async move { - let result = tokio::time::timeout(std::time::Duration::from_secs(5), async { + let result = tokio::time::timeout(Duration::from_secs(secs), async { test_handshake(&cfg).await }) .await; let reply = match result { Ok(inner) => inner, - Err(_) => Err("连接测试超时（5 秒），请检查主机地址与端口".into()), + Err(_) => Err(format!( + "连接测试超时（{secs} 秒），请检查主机地址与端口" + )), }; let _ = tx.try_send(reply); }); @@ -483,9 +519,180 @@ enum EventLoopKind { // ─── 选项构建 ──────────────────────────────────────────────────────────────── +/// 连接超时秒数：至少 1 秒，避免误填 0 导致立刻超时。 +fn connection_timeout_secs(cfg: &ConnectionConfig) -> u64 { + u64::from(cfg.connection_timeout_secs.max(1)) +} + +/// SSL 配置非默认时构建自定义 rustls 配置；返回 None 表示沿用 rumqttc 默认 TLS。 +/// 文件读取/解析失败在此返回 Err，由 connect 的错误通道上报为连接失败事件。 +fn resolve_tls_config(cfg: &ConnectionConfig) -> Result, String> { + if !cfg.transport.is_tls() || cfg.ssl.is_default() { + return Ok(None); + } + Ok(Some(TlsConfiguration::Rustls(build_rustls_config(&cfg.ssl)?))) +} + +/// 按 SslConfig 构造 rustls ClientConfig： +/// - `ignore_ca`：跳过服务器证书链/主机名校验； +/// - `ca_file`：在系统根证书之外追加自定义 CA； +/// - `client_cert_file` + `client_key_file`：双向认证。 +fn build_rustls_config(ssl: &SslConfig) -> Result, String> { + // 只填证书或只填私钥必然握手失败，提前给出明确错误 + if ssl.client_cert_file.is_empty() != ssl.client_key_file.is_empty() { + return Err("客户端证书与客户端密钥必须同时配置".into()); + } + let provider = ensure_crypto_provider(); + let builder = rustls::ClientConfig::builder(); + let builder = if ssl.ignore_ca { + builder + .dangerous() + .with_custom_certificate_verifier(Arc::new(NoVerifier { + provider: provider.clone(), + })) + } else { + let mut roots = rustls::RootCertStore::empty(); + let native = rustls_native_certs::load_native_certs(); + if native.certs.is_empty() && ssl.ca_file.is_empty() { + let detail = native + .errors + .iter() + .map(|e| e.to_string()) + .collect::>() + .join("; "); + return Err(format!("加载系统根证书失败: {detail}")); + } + // 个别系统证书加载失败不阻断连接：自定义 CA 与其余根证书仍参与校验 + for cert in native.certs { + let _ = roots.add(cert); + } + if !ssl.ca_file.is_empty() { + let pem = std::fs::read(&ssl.ca_file) + .map_err(|e| format!("读取 CA 文件 {} 失败: {e}", ssl.ca_file))?; + let mut added = 0usize; + for cert in CertificateDer::pem_slice_iter(&pem) { + let cert = cert.map_err(|e| format!("解析 CA 文件 {} 失败: {e}", ssl.ca_file))?; + roots + .add(cert) + .map_err(|e| format!("CA 文件 {} 中存在无效证书: {e}", ssl.ca_file))?; + added += 1; + } + if added == 0 { + return Err(format!("CA 文件 {} 中没有找到任何证书", ssl.ca_file)); + } + } + builder.with_root_certificates(roots) + }; + + let config = if ssl.client_cert_file.is_empty() { + builder.with_no_client_auth() + } else { + let chain_pem = std::fs::read(&ssl.client_cert_file).map_err(|e| { + format!("读取客户端证书文件 {} 失败: {e}", ssl.client_cert_file) + })?; + let key_pem = std::fs::read(&ssl.client_key_file) + .map_err(|e| format!("读取客户端私钥文件 {} 失败: {e}", ssl.client_key_file))?; + let certs = CertificateDer::pem_slice_iter(&chain_pem) + .collect::, _>>() + .map_err(|e| format!("解析客户端证书文件 {} 失败: {e}", ssl.client_cert_file))?; + if certs.is_empty() { + return Err(format!( + "客户端证书文件 {} 中没有找到证书", + ssl.client_cert_file + )); + } + let key = PrivateKeyDer::from_pem_slice(&key_pem) + .map_err(|e| format!("解析客户端私钥文件 {} 失败: {e}", ssl.client_key_file))?; + builder + .with_client_auth_cert(certs, key) + .map_err(|e| format!("加载客户端证书/私钥失败: {e}"))? + }; + Ok(Arc::new(config)) +} + +/// 取得（必要时安装）进程级 CryptoProvider。 +/// 工作区同时启用 ring 与 aws-lc-rs 特性时 rustls 无法自动推断默认 provider， +/// 不显式安装会让 `ClientConfig::builder()` panic。 +fn ensure_crypto_provider() -> Arc { + if let Some(provider) = rustls::crypto::CryptoProvider::get_default() { + return Arc::clone(provider); + } + let provider = rustls::crypto::aws_lc_rs::default_provider(); + // 输掉安装竞争时直接采用已安装的那个 + match provider.clone().install_default() { + Ok(()) => Arc::new(provider), + Err(installed) => installed, + } +} + +/// 「忽略 CA 校验」用：跳过服务器证书链与主机名校验， +/// 握手签名仍用 provider 校验（与 rustls 官方示例做法一致）。 +struct NoVerifier { + provider: Arc, +} + +impl std::fmt::Debug for NoVerifier { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("NoVerifier").finish_non_exhaustive() + } +} + +impl rustls::client::danger::ServerCertVerifier for NoVerifier { + fn verify_server_cert( + &self, + _end_entity: &CertificateDer<'_>, + _intermediates: &[CertificateDer<'_>], + _server_name: &rustls::pki_types::ServerName<'_>, + _ocsp_response: &[u8], + _now: rustls::pki_types::UnixTime, + ) -> Result { + Ok(rustls::client::danger::ServerCertVerified::assertion()) + } + + fn verify_tls12_signature( + &self, + message: &[u8], + cert: &CertificateDer<'_>, + dss: &rustls::DigitallySignedStruct, + ) -> Result { + rustls::crypto::verify_tls12_signature( + message, + cert, + dss, + &self.provider.signature_verification_algorithms, + ) + } + + fn verify_tls13_signature( + &self, + message: &[u8], + cert: &CertificateDer<'_>, + dss: &rustls::DigitallySignedStruct, + ) -> Result { + rustls::crypto::verify_tls13_signature( + message, + cert, + dss, + &self.provider.signature_verification_algorithms, + ) + } + + fn supported_verify_schemes(&self) -> Vec { + self.provider + .signature_verification_algorithms + .supported_schemes() + } +} + +/// 去掉空白字符串，避免表单存了空串也写进报文属性。 +fn non_blank(s: Option) -> Option { + s.filter(|v| !v.trim().is_empty()) +} + fn build_v5( cfg: &ConnectionConfig, ) -> Result<(V5AsyncClient, V5EventLoop), String> { + let custom_tls = resolve_tls_config(cfg)?; let mut opts = match cfg.transport { TransportKind::Tcp | TransportKind::Tls => { V5Options::new(cfg.client_id.clone(), V5Broker::tcp(&cfg.host, cfg.port)) @@ -496,17 +703,40 @@ fn build_v5( } TransportKind::Wss => { let url = format!("wss://{}:{}{}", cfg.host, cfg.port, normalize_path(&cfg.path)); - V5Options::try_websocket_with_default_tls(cfg.client_id.clone(), url) - .map_err(|e| format!("{e:?}"))? + match &custom_tls { + Some(tls) => V5Options::websocket_with_tls_config( + cfg.client_id.clone(), + url, + tls.clone(), + ) + .map_err(|e| format!("{e:?}"))?, + None => V5Options::try_websocket_with_default_tls(cfg.client_id.clone(), url) + .map_err(|e| format!("{e:?}"))?, + } } }; if cfg.transport == TransportKind::Tls { - opts.set_transport( - V5Transport::try_tls_with_default_config().map_err(|e| format!("TLS 初始化失败: {e:?}"))?, - ); + match &custom_tls { + Some(tls) => { + opts.set_transport(V5Transport::tls_with_config(tls.clone())); + } + None => { + opts.set_transport( + V5Transport::try_tls_with_default_config() + .map_err(|e| format!("TLS 初始化失败: {e:?}"))?, + ); + } + }; } + // 连接超时同时覆盖底层建连与 CONNACK 等待 + let timeout = Duration::from_secs(connection_timeout_secs(cfg)); + opts.set_connect_timeout(timeout); + let mut network_options = rumqttc::NetworkOptions::new(); + network_options.set_connection_timeout(connection_timeout_secs(cfg)); + opts.set_network_options(network_options); + opts.set_keep_alive(cfg.keep_alive); opts.set_clean_start(cfg.clean_start); if cfg.session_expiry_interval > 0 { @@ -526,12 +756,26 @@ fn build_v5( } if let Some(will) = &cfg.last_will && !will.topic.is_empty() { + // v5 遗嘱属性：content_type / response_topic（3.1.1 下无此扩展） + let properties = if will.content_type.is_some() || will.response_topic.is_some() { + Some(rumqttc::mqttbytes::v5::LastWillProperties { + delay_interval: None, + payload_format_indicator: None, + message_expiry_interval: None, + content_type: non_blank(will.content_type.clone()), + response_topic: non_blank(will.response_topic.clone()), + correlation_data: None, + user_properties: Vec::new(), + }) + } else { + None + }; let lw = rumqttc::mqttbytes::v5::LastWill::new( will.topic.clone(), will.payload.as_bytes().to_vec(), v5_qos(will.qos), will.retain, - None, + properties, ); opts.set_last_will(lw); } @@ -544,6 +788,7 @@ fn build_v4( ) -> Result<(rv4::AsyncClient, rv4::EventLoop), String> { use rv4::{Broker as V4Broker, MqttOptions as V4Options, Transport as V4Transport}; + let custom_tls = resolve_tls_config(cfg)?; let mut opts = match cfg.transport { TransportKind::Tcp | TransportKind::Tls => { V4Options::new(cfg.client_id.clone(), V4Broker::tcp(&cfg.host, cfg.port)) @@ -554,15 +799,31 @@ fn build_v4( } TransportKind::Wss => { let url = format!("wss://{}:{}{}", cfg.host, cfg.port, normalize_path(&cfg.path)); - V4Options::try_websocket_with_default_tls(cfg.client_id.clone(), url) - .map_err(|e| format!("{e:?}"))? + match &custom_tls { + Some(tls) => V4Options::websocket_with_tls_config( + cfg.client_id.clone(), + url, + tls.clone(), + ) + .map_err(|e| format!("{e:?}"))?, + None => V4Options::try_websocket_with_default_tls(cfg.client_id.clone(), url) + .map_err(|e| format!("{e:?}"))?, + } } }; if cfg.transport == TransportKind::Tls { - opts.set_transport(V4Transport::try_tls_with_default_config().map_err( - |e| format!("TLS 初始化失败: {e:?}"), - )?); + match &custom_tls { + Some(tls) => { + opts.set_transport(V4Transport::tls_with_config(tls.clone())); + } + None => { + opts.set_transport( + V4Transport::try_tls_with_default_config() + .map_err(|e| format!("TLS 初始化失败: {e:?}"))?, + ); + } + }; } opts.set_keep_alive(cfg.keep_alive); @@ -572,6 +833,7 @@ fn build_v4( } if let Some(will) = &cfg.last_will && !will.topic.is_empty() { + // 3.1.1 的遗嘱没有 content_type / response_topic 属性，忽略 let lw = rv4::mqttbytes::v4::LastWill::new( will.topic.clone(), will.payload.as_bytes().to_vec(), @@ -581,7 +843,12 @@ fn build_v4( opts.set_last_will(lw); } - Ok(rv4::AsyncClient::builder(opts).build()) + let (client, mut eventloop) = rv4::AsyncClient::builder(opts).build(); + // v4 的连接超时（含 CONNACK 等待）由 EventLoop 的 NetworkOptions 控制 + let mut network_options = rv4::NetworkOptions::new(); + network_options.set_connection_timeout(connection_timeout_secs(cfg)); + eventloop.set_network_options(network_options); + Ok((client, eventloop)) } fn normalize_path(path: &str) -> String { @@ -702,10 +969,13 @@ fn spawn_v5_loop( mut cancel_rx: watch::Receiver, connected: Arc, auto_reconnect: bool, + max_reconnect_times: u32, ) { let runtime = engine.runtime.handle().clone(); runtime.spawn(async move { let mut first_connect = true; + // 已执行的重连次数，用于 max_reconnect_times（0=无限）限流 + let mut reconnect_attempts: u32 = 0; loop { tokio::select! { event = eventloop.poll() => { @@ -831,9 +1101,16 @@ fn spawn_v5_loop( format!("连接错误: {e}"), None); break; } + // max_reconnect_times>0 时按次数停止重试；0 表示无限重连 + if max_reconnect_times > 0 && reconnect_attempts >= max_reconnect_times { + engine.log(&id, LogLevel::Error, "reconnect_give_up", + format!("已重连 {reconnect_attempts} 次仍失败，达到最大重连次数 {max_reconnect_times}，停止重连: {e}"), None); + break; + } + reconnect_attempts += 1; engine.log(&id, LogLevel::Warn, "connection_error", - format!("连接错误，2 秒后重连: {e}"), None); - tokio::time::sleep(std::time::Duration::from_secs(2)).await; + format!("连接错误，2 秒后重连（第 {reconnect_attempts} 次）: {e}"), None); + tokio::time::sleep(Duration::from_secs(2)).await; } } } @@ -854,10 +1131,13 @@ fn spawn_v4_loop( mut cancel_rx: watch::Receiver, connected: Arc, auto_reconnect: bool, + max_reconnect_times: u32, ) { let runtime = engine.runtime.handle().clone(); runtime.spawn(async move { let mut first_connect = true; + // 已执行的重连次数，用于 max_reconnect_times（0=无限）限流 + let mut reconnect_attempts: u32 = 0; loop { tokio::select! { event = eventloop.poll() => { @@ -964,9 +1244,16 @@ fn spawn_v4_loop( format!("连接错误: {e}"), None); break; } + // max_reconnect_times>0 时按次数停止重试；0 表示无限重连 + if max_reconnect_times > 0 && reconnect_attempts >= max_reconnect_times { + engine.log(&id, LogLevel::Error, "reconnect_give_up", + format!("已重连 {reconnect_attempts} 次仍失败，达到最大重连次数 {max_reconnect_times}，停止重连: {e}"), None); + break; + } + reconnect_attempts += 1; engine.log(&id, LogLevel::Warn, "connection_error", - format!("连接错误，2 秒后重连: {e}"), None); - tokio::time::sleep(std::time::Duration::from_secs(2)).await; + format!("连接错误，2 秒后重连（第 {reconnect_attempts} 次）: {e}"), None); + tokio::time::sleep(Duration::from_secs(2)).await; } } } diff --git a/gpui-app/src/store.rs b/gpui-app/src/store.rs index 1edbd1e..987679b 100644 --- a/gpui-app/src/store.rs +++ b/gpui-app/src/store.rs @@ -49,9 +49,9 @@ impl Storage { } let mut text = std::fs::read_to_string(&p) .with_context(|| format!("读取 {} 失败", p.display()))?; - // 容忍 Windows 编辑器/PowerShell 写入的 UTF-8 BOM - if text.starts_with('\u{feff}') { - text.drain(..3); + // 容忍 Windows 编辑器/PowerShell 写入的 UTF-8 BOM（BOM 只有 1 个字符，逐字符剥离） + if let Some(rest) = text.strip_prefix('\u{feff}') { + text = rest.to_string(); } if text.trim().is_empty() { return Ok(None); @@ -164,3 +164,102 @@ impl Default for Storage { Self::new() } } + +// ── 连接导入 / 导出（Phase 2 UI 调用） ────────────────────────────────────── + +/// 导出连接为 JSON 数组（与 connections.json 同构：缩进格式化、UTF-8 无 BOM）。 +pub fn export_connections_to(path: &Path, conns: &[ConnectionConfig]) -> Result<(), String> { + let text = serde_json::to_string_pretty(conns) + .map_err(|e| format!("序列化连接失败: {e}"))?; + std::fs::write(path, text).map_err(|e| format!("写入 {} 失败: {e}", path.display())) +} + +/// 从 JSON 文件导入连接。 +/// +/// 容错规则：容忍 UTF-8 BOM；顶层允许是数组，也可能是 `{ "connections": [...] }` 对象； +/// 任一项不是合法连接配置时返回带序号的错误信息。 +pub fn import_connections_from(path: &Path) -> Result, String> { + let bytes = std::fs::read(path).map_err(|e| format!("读取 {} 失败: {e}", path.display()))?; + let mut text = + String::from_utf8(bytes).map_err(|_| format!("{} 不是有效的 UTF-8 文件", path.display()))?; + if let Some(rest) = text.strip_prefix('\u{feff}') { + text = rest.to_string(); + } + let value: serde_json::Value = + serde_json::from_str(text.trim()).map_err(|e| format!("解析 JSON 失败: {e}"))?; + let items = match value { + serde_json::Value::Array(items) => items, + serde_json::Value::Object(mut map) => match map.remove("connections") { + Some(serde_json::Value::Array(items)) => items, + _ => { + return Err( + "JSON 对象中缺少 \"connections\" 数组字段".to_string(), + ); + } + }, + _ => return Err("JSON 顶层应为连接数组".to_string()), + }; + let mut conns = Vec::with_capacity(items.len()); + for (i, item) in items.into_iter().enumerate() { + match serde_json::from_value::(item) { + Ok(c) => conns.push(c), + Err(e) => return Err(format!("第 {} 项不是合法的连接配置: {e}", i + 1)), + } + } + Ok(conns) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn temp_path(name: &str) -> PathBuf { + std::env::temp_dir().join(format!( + "mqttx-store-test-{}-{name}", + uuid::Uuid::new_v4().simple() + )) + } + + #[test] + fn export_import_roundtrip() { + let path = temp_path("roundtrip.json"); + let mut c = ConnectionConfig::new(); + c.group = Some("测试组".into()); + c.connection_timeout_secs = 25; + c.max_reconnect_times = 5; + export_connections_to(&path, std::slice::from_ref(&c)).expect("导出应成功"); + let back = import_connections_from(&path).expect("导入应成功"); + assert_eq!(back.len(), 1); + assert_eq!(back[0].id, c.id); + assert_eq!(back[0].group.as_deref(), Some("测试组")); + assert_eq!(back[0].connection_timeout_secs, 25); + assert_eq!(back[0].max_reconnect_times, 5); + let _ = std::fs::remove_file(&path); + } + + #[test] + fn import_tolerates_bom_and_object_form() { + let path = temp_path("bom-object.json"); + let c = ConnectionConfig::new(); + let body = serde_json::to_string(&serde_json::json!({ + "connections": [serde_json::to_value(&c).unwrap()] + })) + .unwrap(); + // 写入 BOM 前缀，验证容错 + std::fs::write(&path, format!("\u{feff}{body}")).expect("写入测试文件"); + let back = import_connections_from(&path).expect("BOM + 对象形式应可导入"); + assert_eq!(back.len(), 1); + let _ = std::fs::remove_file(&path); + } + + #[test] + fn import_reports_invalid_item_index() { + let path = temp_path("invalid.json"); + let good = serde_json::to_value(ConnectionConfig::new()).unwrap(); + let body = serde_json::json!([good, {"id": "bad"}]).to_string(); + std::fs::write(&path, body).expect("写入测试文件"); + let err = import_connections_from(&path).expect_err("非法项应报错"); + assert!(err.contains("第 2 项"), "错误应定位到序号: {err}"); + let _ = std::fs::remove_file(&path); + } +} diff --git a/gpui-app/src/ui/app.rs b/gpui-app/src/ui/app.rs index 8cb9f36..a321caa 100644 --- a/gpui-app/src/ui/app.rs +++ b/gpui-app/src/ui/app.rs @@ -1,6 +1,7 @@ //! 应用外壳：标题栏 + 连接侧边栏 + 多标签连接视图 + 引擎事件泵。 use std::collections::{HashMap, VecDeque}; +use std::path::PathBuf; use gpui_kit::component::button::{Button, ButtonVariants as _}; use gpui_kit::component::scroll::ScrollableElement as _; @@ -13,14 +14,16 @@ use gpui_kit::component::{ }; use gpui_kit::prelude::FluentBuilder as _; use gpui_kit::{ - div, px, App, AppContext as _, Context, Entity, InteractiveElement as _, StatefulInteractiveElement as _, - IntoElement, ParentElement as _, Render, SharedString, Styled as _, Window, + div, px, App, AppContext as _, Context, Entity, InteractiveElement as _, IntoElement, + KeyBinding, ParentElement as _, Render, SharedString, StatefulInteractiveElement as _, + Styled as _, Window, }; use crate::aliyun::AliyunPreset; use crate::model::{ AppSettings, ConnectionConfig, ConnectionStatus, Direction, GlobalVariable, LogEntry, - LogLevel, MqttRecord, PublishParams, PublishPreset, Subscription, ThemeModePref, + LogLevel, MqttRecord, PublishParams, PublishPreset, SubscribeOptions, Subscription, + ThemeModePref, }; use crate::mqtt::{EngineEvent, MqttEngine}; use crate::store::Storage; @@ -33,6 +36,35 @@ use crate::ui::{ const MAX_LOGS: usize = 3000; +// 应用级快捷键动作（context=None：任意焦点状态下都匹配）。 +gpui_kit::actions!(mqttx, [NewConnection, OpenSettings, ExportConnections, ImportConnections]); + +/// 侧边栏分组过滤；chips 只做列表过滤，分组重命名/删除通过编辑连接的分组字段完成。 +#[derive(Clone, PartialEq, Eq)] +enum GroupFilter { + All, + Ungrouped, + Named(String), +} + +/// 全局 action 回调只有 `&mut App`：先 defer 出当前分发栈，再取活动窗口回到实体执行， +/// 避免在窗口借用期间重入 `update`。 +fn run_on_active_window( + cx: &mut App, + weak: gpui_kit::WeakEntity, + f: impl FnOnce(Entity, &mut Window, &mut App) + 'static, +) { + let Some(handle) = cx.active_window() else { + return; + }; + cx.defer(move |cx| { + let Some(entity) = weak.upgrade() else { + return; + }; + let _ = handle.update(cx, move |_, window, cx| f(entity, window, cx)); + }); +} + pub struct MqttXApp { storage: Storage, pub engine: std::sync::Arc, @@ -59,6 +91,10 @@ pub struct MqttXApp { views: HashMap>, search: Entity, + /// 分组过滤状态（chips 点击切换，与搜索条件叠加） + group_filter: GroupFilter, + /// 原生文件对话框进行中标记（对话框阻塞后台线程，防止重复弹出） + file_dialog_open: bool, } impl MqttXApp { @@ -88,7 +124,51 @@ impl MqttXApp { let search = cx.new(|cx| InputState::new(window, cx).placeholder("搜索连接…")); - + // ── 全局快捷键：绑定 + action 注册 ── + // 元素级 on_action 依赖焦点路径，无焦点时分发从树根开始收不到； + // 全局 action 在冒泡末端必达，经 defer 取窗口避免在分发栈内重入。 + cx.bind_keys([ + KeyBinding::new("ctrl-n", NewConnection, None), + KeyBinding::new("ctrl-,", OpenSettings, None), + KeyBinding::new("ctrl-shift-e", ExportConnections, None), + KeyBinding::new("ctrl-shift-i", ImportConnections, None), + ]); + let weak = cx.entity().downgrade(); + // Context 上有同名 on_action（绘制期注册），这里必须走 App 的全局注册 + App::on_action::(cx, move |_, cx| { + let w = weak.clone(); + run_on_active_window(cx, w, |entity, window, cx| { + if window.has_active_dialog(cx) { + return; + } + entity.update(cx, |app, cx| app.open_connection_form(None, window, cx)); + }); + }); + let weak = cx.entity().downgrade(); + App::on_action::(cx, move |_, cx| { + let w = weak.clone(); + run_on_active_window(cx, w, |entity, window, cx| { + if window.has_active_dialog(cx) { + return; + } + crate::ui::settings_dialog::open(entity, window, cx); + }); + }); + let weak = cx.entity().downgrade(); + App::on_action::(cx, move |_, cx| { + let w = weak.clone(); + run_on_active_window(cx, w, |entity, window, cx| { + entity.update(cx, |app, cx| app.export_connections(window, cx)); + }); + }); + let weak = cx.entity().downgrade(); + App::on_action::(cx, move |_, cx| { + let w = weak.clone(); + run_on_active_window(cx, w, |entity, window, cx| { + entity.update(cx, |app, cx| app.import_connections(window, cx)); + }); + }); + Self { storage, engine, @@ -108,6 +188,8 @@ impl MqttXApp { active_tab: None, views: HashMap::new(), search, + group_filter: GroupFilter::All, + file_dialog_open: false, } } @@ -206,7 +288,7 @@ impl MqttXApp { } fn on_connected(&mut self, connection_id: &str) { - // 连接成功后自动恢复已保存订阅（auto resubscribe）。 + // 连接成功后自动恢复已保存订阅（auto resubscribe），禁用的订阅跳过。 let Some(config) = self.connections.iter().find(|c| c.id == connection_id).cloned() else { return; }; @@ -217,10 +299,13 @@ impl MqttXApp { .subscriptions .iter() .filter(|s| s.connection_id == connection_id) + .filter(|s| s.enabled) .cloned() .collect(); for sub in subs { - self.engine.subscribe(connection_id.to_string(), sub); + let opts = SubscribeOptions::from(&sub); + self.engine + .subscribe_with_options(connection_id.to_string(), sub, opts); } } @@ -242,12 +327,61 @@ impl MqttXApp { } fn push_log(&mut self, entry: LogEntry) { + // 落盘只在事件路径执行（不在 render 热路径），与内存环形缓冲互不影响 + self.write_log_file(&entry); if self.logs.len() >= MAX_LOGS { self.logs.pop_front(); } self.logs.push_back(entry); } + /// 追加写入按日切分的日志文件 `mqttx-YYYY-MM-DD.log`。 + /// 每条都尝试写入；失败只在首次 eprintln 一次后保持静默，且绝不回调 + /// log/push_log（防止递归触发日志）。 + fn write_log_file(&self, entry: &LogEntry) { + use std::io::Write as _; + use std::sync::atomic::{AtomicBool, Ordering}; + + static WRITE_FAILED: AtomicBool = AtomicBool::new(false); + + let result = (|| -> std::io::Result<()> { + use chrono::TimeZone as _; + let ts = chrono::Local + .timestamp_millis_opt(entry.timestamp) + .single() + .unwrap_or_else(chrono::Local::now); + let dir = self.storage.log_dir(); + std::fs::create_dir_all(&dir)?; + let path = dir.join(format!("mqttx-{}.log", ts.format("%Y-%m-%d"))); + // 连接名查不到时回退为 id，保证行格式恒定 + let conn = self + .connections + .iter() + .find(|c| c.id == entry.connection_id) + .map(|c| c.name.as_str()) + .unwrap_or(entry.connection_id.as_str()); + let mut file = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(path)?; + writeln!( + file, + "{} [{}] {}/{}: {}", + ts.format("%Y-%m-%d %H:%M:%S%.3f"), + entry.level.label(), + conn, + entry.event, + entry.message + ) + })(); + + if let Err(e) = result + && !WRITE_FAILED.swap(true, Ordering::Relaxed) + { + eprintln!("[app] 写入日志文件失败（后续静默）: {e}"); + } + } + pub fn log(&mut self, conn: &str, level: LogLevel, event: &str, message: String) { self.push_log(LogEntry { timestamp: chrono::Local::now().timestamp_millis(), @@ -386,6 +520,134 @@ impl MqttXApp { self.storage.save_settings(&self.settings); } + /// 数据目录（设置对话框展示 / 打开用）。 + pub fn data_dir(&self) -> PathBuf { + self.storage.dir().to_path_buf() + } + + /// 日志目录（设置对话框展示 / 打开用）。 + pub fn log_dir(&self) -> PathBuf { + self.storage.log_dir() + } + + // ── 连接导入 / 导出 ──────────────────────────────────────────────────── + + /// 在后台线程弹出阻塞式原生文件对话框，完成后回主线程执行 `done`。 + /// 对话框进行中置位，防止重复弹出。 + fn spawn_file_dialog( + &mut self, + work: F, + done: D, + window: &mut Window, + cx: &mut Context, + ) where + F: FnOnce() -> R + Send + 'static, + R: Send + 'static, + D: FnOnce(&mut Self, R, &mut Window, &mut Context) + 'static, + { + if self.file_dialog_open { + return; + } + self.file_dialog_open = true; + // rfd 的阻塞对话框不可占用 UI 线程：结果经 smol 通道送回异步任务 + let (tx, rx) = smol::channel::bounded::(1); + std::thread::spawn(move || { + let _ = tx.send_blocking(work()); + }); + let weak = cx.entity().downgrade(); + cx.spawn_in(window, async move |_this, cx: &mut gpui_kit::AsyncWindowContext| { + let result = rx.recv().await.ok(); + weak.update_in(cx, |app, window, cx| { + app.file_dialog_open = false; + if let Some(result) = result { + done(app, result, window, cx); + } + }) + .ok(); + }) + .detach(); + } + + /// 导出全部连接为 JSON（原生保存对话框，默认文件名 connections-export.json）。 + pub fn export_connections(&mut self, window: &mut Window, cx: &mut Context) { + let dir = self.storage.dir().to_path_buf(); + self.spawn_file_dialog( + move || { + rfd::FileDialog::new() + .set_title("导出连接") + .set_directory(dir) + .set_file_name("connections-export.json") + .add_filter("JSON", &["json"]) + .save_file() + }, + |app, path, window, cx| { + let Some(path) = path else { + return; // 用户取消 + }; + let count = app.connections.len(); + match crate::store::export_connections_to(&path, &app.connections) { + Ok(()) => window.push_notification( + Notification::success(format!( + "已导出 {count} 条连接到 {}", + path.display() + )), + cx, + ), + Err(e) => window.push_notification(Notification::error(e), cx), + } + }, + window, + cx, + ); + } + + /// 导入连接：按 id 去重（同 id 跳过），新 id 保留文件原值；读取解析在后台线程完成。 + pub fn import_connections(&mut self, window: &mut Window, cx: &mut Context) { + let dir = self.storage.dir().to_path_buf(); + self.spawn_file_dialog( + move || { + let Some(path) = rfd::FileDialog::new() + .set_title("导入连接") + .set_directory(dir) + .add_filter("JSON", &["json"]) + .pick_file() + else { + return Ok(None); // 用户取消 + }; + crate::store::import_connections_from(&path).map(Some) + }, + |app, result, window, cx| match result { + Err(e) => { + window.push_notification(Notification::error(e), cx); + } + Ok(None) => {} + Ok(Some(items)) => { + let existing: std::collections::HashSet = + app.connections.iter().map(|c| c.id.clone()).collect(); + let mut added = 0usize; + let mut skipped = 0usize; + for item in items { + if existing.contains(&item.id) { + skipped += 1; + } else { + app.connections.push(item); + added += 1; + } + } + if added > 0 { + app.storage.save_connections(&app.connections); + } + window.push_notification( + Notification::success(format!("导入 {added} 条，跳过 {skipped} 条")), + cx, + ); + } + }, + window, + cx, + ); + } + // ── 对话框 ──────────────────────────────────────────────────────────── pub fn open_connection_form( @@ -396,7 +658,16 @@ impl MqttXApp { ) { let app = cx.entity().downgrade(); let engine = self.engine.clone(); - let form = cx.new(|cx| ConnectionForm::new(engine, edit, window, cx)); + // 表单的分组下拉需要现有分组列表（去重、排序） + let mut groups: Vec = self + .connections + .iter() + .filter_map(|c| c.group.as_ref().map(|g| g.trim().to_string())) + .filter(|g| !g.is_empty()) + .collect(); + groups.sort(); + groups.dedup(); + let form = cx.new(|cx| ConnectionForm::new(engine, edit, groups, window, cx)); window.open_dialog(cx, move |dialog, _, cx| { let form_ok = form.clone(); let form_test = form.clone(); @@ -550,6 +821,8 @@ impl MqttXApp { let w1 = app_weak.clone(); let w2 = app_weak.clone(); let w3 = app_weak.clone(); + let w4 = app_weak.clone(); + let w5 = app_weak.clone(); menu.item( PopupMenuItem::new("阿里云设备") .on_click(move |_, window, cx| { @@ -566,6 +839,25 @@ impl MqttXApp { }, )) .separator() + .item(PopupMenuItem::new("导出连接").on_click( + move |_, window, cx| { + if let Some(entity) = w4.upgrade() { + entity.update(cx, |app, cx| { + app.export_connections(window, cx) + }); + } + }, + )) + .item(PopupMenuItem::new("导入连接").on_click( + move |_, window, cx| { + if let Some(entity) = w5.upgrade() { + entity.update(cx, |app, cx| { + app.import_connections(window, cx) + }); + } + }, + )) + .separator() .item(PopupMenuItem::new("设置").on_click( move |_, window, cx| { if let Some(entity) = w3.upgrade() { @@ -602,12 +894,23 @@ impl MqttXApp { fn render_sidebar(&mut self, _window: &mut Window, cx: &mut Context) -> impl IntoElement { let query = self.search.read(cx).value().to_lowercase(); let search = self.search.clone(); + let filter = self.group_filter.clone(); let total = self.connections.len(); let hover_bg = cx.theme().muted; let mut rows = v_flex().gap_0p5().flex_1().overflow_y_scrollbar(); let mut shown = 0usize; for conn in self.connections.clone() { + // 分组过滤与搜索叠加（AND） + let group = conn.group.as_deref().map(str::trim).unwrap_or(""); + let match_group = match &filter { + GroupFilter::All => true, + GroupFilter::Ungrouped => group.is_empty(), + GroupFilter::Named(name) => group == name.as_str(), + }; + if !match_group { + continue; + } if !query.is_empty() && !conn.name.to_lowercase().contains(&query) { continue; } @@ -692,8 +995,6 @@ impl MqttXApp { .text_color(cx.theme().muted_foreground) .child("还没有连接，点击上方 + 新建"), ) - } else if query.is_empty() { - v_flex().flex_1().child(rows) } else if shown == 0 { v_flex() .flex_1() @@ -748,9 +1049,91 @@ impl MqttXApp { .child(v_flex().px_2().pb_2().child(Input::new(&search).small().prefix( gpui_kit::component::Icon::new(IconName::Search).small(), ))) + .child(self.render_group_chips(cx)) .child(body) } + /// 搜索框下方的分组 chips：全部 / 未分组 / 各分组（带计数徽标），点击切换过滤。 + fn render_group_chips(&self, cx: &mut Context) -> impl IntoElement { + let mut names: Vec = self + .connections + .iter() + .filter_map(|c| c.group.as_deref().map(str::trim)) + .filter(|g| !g.is_empty()) + .map(str::to_string) + .collect(); + names.sort(); + names.dedup(); + + let ungrouped = self + .connections + .iter() + .filter(|c| c.group.as_deref().map(str::trim).unwrap_or("").is_empty()) + .count(); + + let mut row = h_flex().gap_1().flex_wrap().px_2().pb_1(); + row = row.child(self.render_group_chip( + "chip-all", + format!("全部 {}", self.connections.len()), + GroupFilter::All, + cx, + )); + row = row.child(self.render_group_chip( + "chip-ungrouped", + format!("未分组 {ungrouped}"), + GroupFilter::Ungrouped, + cx, + )); + for name in names { + let count = self + .connections + .iter() + .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)), + format!("{name} {count}"), + GroupFilter::Named(name), + cx, + )); + } + row + } + + fn render_group_chip( + &self, + id: impl Into, + label: String, + filter: GroupFilter, + cx: &mut Context, + ) -> impl IntoElement { + let selected = match (&self.group_filter, &filter) { + (GroupFilter::All, GroupFilter::All) + | (GroupFilter::Ungrouped, GroupFilter::Ungrouped) => true, + (GroupFilter::Named(a), GroupFilter::Named(b)) => a == b, + _ => false, + }; + div() + .id(id) + .px_2() + .py_0p5() + .rounded_md() + .text_xs() + .when(selected, |d| d.bg(cx.theme().secondary).font_semibold()) + .when(!selected, |d| { + d.bg(cx.theme().muted) + .text_color(cx.theme().muted_foreground) + .hover(|d| d.bg(cx.theme().secondary)) + }) + .cursor_pointer() + .on_click(cx.listener(move |this, _, _, cx| { + // listener 是 Fn（可多次调用），只能克隆而不能移出捕获的 filter + this.group_filter = filter.clone(); + cx.notify(); + })) + .child(label) + } + fn render_conn_power( &self, cfg: &ConnectionConfig, diff --git a/gpui-app/src/ui/connection_form.rs b/gpui-app/src/ui/connection_form.rs index 64dd573..aaa4490 100644 --- a/gpui-app/src/ui/connection_form.rs +++ b/gpui-app/src/ui/connection_form.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use gpui_kit::component::scroll::ScrollableElement as _; +use gpui_kit::component::button::{Button, ButtonVariants as _}; use gpui_kit::component::input::{Input, InputState}; use gpui_kit::component::select::{Select, SelectEvent, SelectState}; use gpui_kit::component::switch::Switch; @@ -18,18 +19,22 @@ use gpui_kit::{ Styled as _, Subscription, Window, }; -use crate::model::{ConnectionConfig, LastWill, ProtocolVersion, TransportKind}; +use crate::model::{ConnectionConfig, LastWill, ProtocolVersion, SslConfig, TransportKind}; use crate::mqtt::MqttEngine; use crate::ui::widgets::{field, make_select, OptionDelegate}; const PROTOCOLS: [&str; 2] = ["MQTT 5.0", "MQTT 3.1.1"]; const TRANSPORTS: [&str; 4] = ["TCP", "TLS", "WebSocket", "WSS"]; const QOS: [&str; 3] = ["QoS 0", "QoS 1", "QoS 2"]; +/// 分组下拉的首项：选择它表示不归属任何分组。 +const NO_GROUP: &str = "无分组"; pub struct ConnectionForm { engine: Arc, editing_id: Option, created_at: i64, + /// 已有连接的分组列表（去重），用于分组下拉 + groups: Vec, name: Entity, host: Entity, @@ -43,24 +48,39 @@ pub struct ConnectionForm { receive_max: Entity, max_packet: Entity, topic_alias: Entity, + conn_timeout: Entity, + max_reconnect: Entity, will_topic: Entity, will_payload: Entity, + will_content_type: Entity, + will_response_topic: Entity, + + // SSL/TLS 区块（transport 为 TLS/WSS 时显示） + ssl_ca: Entity, + ssl_client_cert: Entity, + ssl_client_key: Entity, + + // 新分组名输入；非空时优先于下拉选择 + group_new: Entity, protocol: Entity>, transport: Entity>, will_qos: Entity>, + group_select: Entity>, clean_start: bool, auto_resubscribe: bool, auto_reconnect: bool, will_enabled: bool, will_retain: bool, + ssl_ignore_ca: bool, // 按需显示：高级分组默认折叠 show_auth: bool, show_mqtt5: bool, show_will: bool, + show_ssl: bool, testing: bool, _subs: Vec, @@ -105,6 +125,7 @@ impl ConnectionForm { pub fn new( engine: Arc, edit: Option, + groups: Vec, window: &mut Window, cx: &mut Context, ) -> Self { @@ -127,6 +148,18 @@ impl ConnectionForm { .masked(true) }); let keep_alive = input(&c.keep_alive.to_string(), "60", window, cx); + let conn_timeout = input( + &c.connection_timeout_secs.to_string(), + "秒（默认 10）", + window, + cx, + ); + let max_reconnect = input( + &c.max_reconnect_times.to_string(), + "0=无限重连", + window, + cx, + ); let session_expiry = input( &if c.session_expiry_interval > 0 { c.session_expiry_interval.to_string() @@ -141,7 +174,13 @@ impl ConnectionForm { let max_packet = optional_input(c.maximum_packet_size, "最大报文长度", window, cx); let topic_alias = optional_input(c.topic_alias_maximum.map(u32::from), "主题别名上限", window, cx); - let (will_topic, will_payload, will_enabled, will_retain, will_qos_idx) = + let ssl_ca = input(&c.ssl.ca_file, "CA 证书 PEM 路径（可选）", window, cx); + let ssl_client_cert = input(&c.ssl.client_cert_file, "客户端证书 PEM 路径（可选）", window, cx); + let ssl_client_key = input(&c.ssl.client_key_file, "客户端私钥 PEM 路径（可选）", window, cx); + + let group_new = input("", "新分组名（可选）", window, cx); + + let (will_topic, will_payload, will_enabled, will_retain, will_qos_idx, will_content_type, will_response_topic) = match &c.last_will { Some(w) => ( input(&w.topic, "遗嘱主题", window, cx), @@ -149,6 +188,8 @@ impl ConnectionForm { true, w.retain, w.qos as usize, + input(w.content_type.as_deref().unwrap_or(""), "内容类型（可选）", window, cx), + input(w.response_topic.as_deref().unwrap_or(""), "响应主题（可选）", window, cx), ), None => ( input("", "遗嘱主题", window, cx), @@ -156,6 +197,8 @@ impl ConnectionForm { false, false, 0, + input("", "内容类型（可选）", window, cx), + input("", "响应主题（可选）", window, cx), ), }; @@ -171,6 +214,16 @@ impl ConnectionForm { .unwrap_or(0); let transport = make_select(&TRANSPORTS, transport_idx, window, cx); let will_qos = make_select(&QOS, will_qos_idx, window, cx); + // 分组下拉：首项固定为「无分组」，其后是已有分组 + let mut group_labels = vec![NO_GROUP]; + group_labels.extend(groups.iter().map(|s| s.as_str())); + let group_idx = c + .group + .as_deref() + .and_then(|g| groups.iter().position(|x| x == g)) + .map(|i| i + 1) + .unwrap_or(0); + let group_select = make_select(&group_labels, group_idx, window, cx); // 切换传输方式时，把端口同步成该方式的默认端口，并刷新 WS 路径的按需显示。 let port_for_sub = port.clone(); @@ -203,6 +256,7 @@ impl ConnectionForm { engine, editing_id, created_at, + groups, name, host, port, @@ -211,23 +265,34 @@ impl ConnectionForm { username, password, keep_alive, + conn_timeout, + max_reconnect, session_expiry, receive_max, max_packet, topic_alias, + ssl_ca, + ssl_client_cert, + ssl_client_key, + group_new, will_topic, will_payload, + will_content_type, + will_response_topic, protocol, transport, will_qos, + group_select, clean_start: c.clean_start, auto_resubscribe: c.auto_resubscribe, auto_reconnect: c.auto_reconnect, will_enabled, will_retain, + ssl_ignore_ca: c.ssl.ignore_ca, show_auth: true, show_mqtt5: false, show_will: false, + show_ssl: false, testing: false, _subs: vec![sub_transport, sub_protocol], } @@ -298,6 +363,48 @@ impl ConnectionForm { .map(|v| u16::try_from(v).map_err(|_| "主题别名上限不能超过 65535".to_string())) .transpose()?; + let conn_timeout_raw = self.val(&self.conn_timeout, cx); + let connection_timeout_secs = if conn_timeout_raw.trim().is_empty() { + 10 + } else { + conn_timeout_raw + .trim() + .parse::() + .ok() + .filter(|v| *v >= 1) + .ok_or_else(|| "连接超时需要是 1~65535 的整数（秒）".to_string())? + }; + let max_reconnect_raw = self.val(&self.max_reconnect, cx); + let max_reconnect_times = if max_reconnect_raw.trim().is_empty() { + 0 + } else { + max_reconnect_raw + .trim() + .parse::() + .map_err(|_| "最大重连次数需要是非负整数（0=无限）".to_string())? + }; + + // 分组：新分组名非空时优先，否则取下拉所选（首项=无分组） + let group = { + let new_name = self.val(&self.group_new, cx); + let trimmed = new_name.trim(); + if !trimmed.is_empty() { + Some(trimmed.to_string()) + } else { + let idx = self + .group_select + .read(cx) + .selected_value() + .copied() + .unwrap_or(0); + if idx == 0 { + None + } else { + self.groups.get(idx - 1).cloned() + } + } + }; + let protocol = match self.protocol.read(cx).selected_value() { Some(1) => ProtocolVersion::V311, _ => ProtocolVersion::V5, @@ -309,18 +416,35 @@ impl ConnectionForm { .and_then(|i| TransportKind::ALL.get(*i).copied()) .unwrap_or_default(); + let ssl = SslConfig { + ca_file: self.val(&self.ssl_ca, cx).trim().to_string(), + client_cert_file: self.val(&self.ssl_client_cert, cx).trim().to_string(), + client_key_file: self.val(&self.ssl_client_key, cx).trim().to_string(), + ignore_ca: self.ssl_ignore_ca, + }; + // 只填证书或只填私钥必然握手失败；仅 TLS/WSS 下 SSL 生效，按生效传输校验 + if transport.is_tls() + && ssl.client_cert_file.is_empty() != ssl.client_key_file.is_empty() + { + return Err("客户端证书与客户端密钥必须同时填写".into()); + } + let last_will = if self.will_enabled { let topic = self.val(&self.will_topic, cx); if topic.trim().is_empty() { return Err("遗嘱主题不能为空".into()); } + let opt_str = |v: String| { + let t = v.trim().to_string(); + if t.is_empty() { None } else { Some(t) } + }; Some(LastWill { topic, payload: self.val(&self.will_payload, cx), qos: self.will_qos.read(cx).selected_value().copied().unwrap_or(0) as u8, retain: self.will_retain, - content_type: None, - response_topic: None, + content_type: opt_str(self.val(&self.will_content_type, cx)), + response_topic: opt_str(self.val(&self.will_response_topic, cx)), }) } else { None @@ -348,6 +472,10 @@ impl ConnectionForm { auto_reconnect: self.auto_reconnect, last_will, created_at: self.created_at, + group, + connection_timeout_secs, + max_reconnect_times, + ssl, }) } @@ -406,6 +534,7 @@ impl ConnectionForm { "auto_reconnect" => this.auto_reconnect = v, "will_enabled" => this.will_enabled = v, "will_retain" => this.will_retain = v, + "ssl_ignore_ca" => this.ssl_ignore_ca = v, _ => {} } cx.notify(); @@ -490,6 +619,11 @@ impl Render for ConnectionForm { self.transport.read(cx).selected_value(), Some(2) | Some(3) ); + // SSL/TLS 区块只对 TLS/WSS 有意义 + let is_tls = matches!( + self.transport.read(cx).selected_value(), + Some(1) | Some(3) + ); v_flex() .gap_3() .max_h(px(560.)) @@ -508,7 +642,24 @@ impl Render for ConnectionForm { .child(div().flex_1().min_w(px(0.)).child(field("传输", Select::new(&self.transport)))) ) .child(h_flex().gap_2().w_full() + .child(div().flex_1().min_w(px(0.)).child(field("分组", Select::new(&self.group_select)))) + .child(div().flex_1().min_w(px(0.)).child(field("新建分组（优先于左侧选择）", Input::new(&self.group_new)))) + ) + .child(h_flex().gap_2().w_full().items_end() .child(div().flex_1().min_w(px(0.)).child(field("Client ID", Input::new(&self.client_id)))) + // 一键换一个随机 Client ID，与 ConnectionConfig::new 的生成逻辑一致 + .child( + Button::new("cid-regen") + .icon(IconName::RefreshCw) + .ghost() + .tooltip("重新生成 Client ID") + .on_click(cx.listener(|this, _, window, cx| { + let cid = crate::model::generate_client_id(); + this.client_id + .update(cx, |s, cx| s.set_value(cid, window, cx)); + cx.notify(); + })), + ) // WS 路径仅对 WebSocket/WSS 有意义，按需显示 .when(is_ws, |w| w.child( div().flex_1().min_w(px(0.)).child(field("WS 路径", Input::new(&self.path))) @@ -525,11 +676,27 @@ impl Render for ConnectionForm { .child(div().flex_1().min_w(px(0.)).child(field("密码", Input::new(&self.password).mask_toggle()))) .child(div().w(px(110.)).child(field("Keep Alive (秒)", Input::new(&self.keep_alive)))) ) + .child(h_flex().gap_2().w_full() + .child(div().w(px(150.)).child(field("连接超时 (秒)", Input::new(&self.conn_timeout)))) + .child(div().w(px(170.)).child(field("最大重连次数 (0=无限)", Input::new(&self.max_reconnect)))) + ) .child(self.switch_row("clean_start", "Clean Start / Clean Session", self.clean_start, cx)) .child(self.switch_row("auto_reconnect", "断线自动重连", self.auto_reconnect, cx)) .child(self.switch_row("auto_resubscribe", "连接后自动恢复订阅", self.auto_resubscribe, cx)), cx, )) + // ── SSL/TLS（仅 TLS/WSS 显示） ── + .when(is_tls, |w| w.child(self.collapsible_section( + "sec-ssl", "SSL/TLS", None, false, self.show_ssl, + |f| f.show_ssl = !f.show_ssl, border, card_bg, + v_flex().gap_2() + // Phase 2 或后续接原生文件对话框；GPUI 暂无内置文件选择器，先用路径文本输入 + .child(field("CA 证书 (PEM)", Input::new(&self.ssl_ca))) + .child(field("客户端证书 (PEM)", Input::new(&self.ssl_client_cert))) + .child(field("客户端密钥 (PEM)", Input::new(&self.ssl_client_key))) + .child(self.switch_row("ssl_ignore_ca", "忽略 CA 校验（不验证服务器证书）", self.ssl_ignore_ca, cx)), + cx, + ))) // ── MQTT 5 高级属性（仅 v5 协议显示） ── .when(is_v5, |w| w.child(self.collapsible_section( "sec-mqtt5", "MQTT 5 属性", None, false, self.show_mqtt5, @@ -554,6 +721,11 @@ impl Render for ConnectionForm { .child(div().w(px(130.)).child(field("QoS", Select::new(&self.will_qos)))) ) .child(field("遗嘱内容", Input::new(&self.will_payload))) + // v5 遗嘱属性；3.1.1 连接会忽略这两项 + .child(h_flex().gap_2().w_full() + .child(div().flex_1().min_w(px(0.)).child(field("内容类型 (Content Type)", Input::new(&self.will_content_type)))) + .child(div().flex_1().min_w(px(0.)).child(field("响应主题 (Response Topic)", Input::new(&self.will_response_topic)))) + ) .child(self.switch_row("will_retain", "Retain", self.will_retain, cx)) ), cx, diff --git a/gpui-app/src/ui/connection_view.rs b/gpui-app/src/ui/connection_view.rs index fb68896..53d0c42 100644 --- a/gpui-app/src/ui/connection_view.rs +++ b/gpui-app/src/ui/connection_view.rs @@ -2,7 +2,7 @@ use std::sync::Arc; -use gpui_kit::component::button::{Button, ButtonVariants as _}; +use gpui_kit::component::button::{Button, ButtonGroup, ButtonVariants as _}; use gpui_kit::component::scroll::ScrollableElement as _; use gpui_kit::component::input::{Input, InputState, Textarea, TextareaState}; use gpui_kit::component::menu::{DropdownMenu as _, PopupMenuItem}; @@ -16,14 +16,15 @@ use gpui_kit::component::{ use gpui_kit::prelude::FluentBuilder as _; use crate::ui::IconName; use gpui_kit::{ - div, px, App, AppContext as _, Context, Entity, InteractiveElement as _, StatefulInteractiveElement as _, - IntoElement, ParentElement as _, Render, SharedString, Styled as _, Window, + div, hsla, px, App, AppContext as _, ClipboardItem, Context, Entity, Hsla, + InteractiveElement as _, IntoElement, ParentElement as _, Render, SharedString, + StatefulInteractiveElement as _, Styled as _, Window, }; use gpui_kit::component::IndexPath; use crate::model::{ render_template, ConnectionConfig, ConnectionStatus, Direction, LogLevel, MqttRecord, - PayloadFormat, PublishParams, + PayloadFormat, PublishParams, SubscribeOptions, Subscription, }; use crate::mqtt::MqttEngine; use crate::ui::app::MqttXApp; @@ -31,7 +32,22 @@ use crate::ui::widgets::{field, format_time, make_select, KvEditor, OptionDelega const QOS: [&str; 3] = ["QoS 0", "QoS 1", "QoS 2"]; const PAYLOAD_FORMATS: [&str; 4] = ["Plaintext", "JSON", "Base64", "Hex"]; +/// Retain Handling 0/1/2 的下拉文案 +const RETAIN_HANDLING: [&str; 3] = ["0 每次发送", "1 仅新订阅", "2 不发送"]; const MAX_RENDERED_MESSAGES: usize = 300; +/// 订阅色板预设：色相（度）+ 名称 +const PRESET_HUES: [(f32, &str); 10] = [ + (0., "红"), + (30., "橙"), + (60., "黄"), + (120., "绿"), + (160., "青"), + (200., "蓝"), + (240., "靛"), + (270., "紫"), + (300., "品红"), + (330., "玫红"), +]; #[derive(Clone, Copy, PartialEq, Eq)] enum Panel { @@ -39,14 +55,98 @@ enum Panel { Logs, } +/// 消息流方向过滤（与搜索、订阅过滤叠加生效） +#[derive(Clone, Copy, PartialEq, Eq)] +enum DirFilter { + All, + Received, + Published, +} + +impl DirFilter { + fn label(self) -> &'static str { + match self { + Self::All => "全部", + Self::Received => "接收", + Self::Published => "发布", + } + } + + fn index(self) -> usize { + match self { + Self::All => 0, + Self::Received => 1, + Self::Published => 2, + } + } + + fn from_index(i: usize) -> Self { + match i { + 1 => Self::Received, + 2 => Self::Published, + _ => Self::All, + } + } +} + +/// 展开消息详情里的 payload 展示格式。任一时刻只展开一条消息， +/// 视图级单值即可视作「逐消息」选择。 +#[derive(Clone, Copy, PartialEq, Eq)] +enum DetailFormat { + Auto, + Text, + Hex, + Base64, +} + +impl DetailFormat { + fn label(self) -> &'static str { + match self { + Self::Auto => "自动", + Self::Text => "文本", + Self::Hex => "Hex", + Self::Base64 => "Base64", + } + } + + fn index(self) -> usize { + match self { + Self::Auto => 0, + Self::Text => 1, + Self::Hex => 2, + Self::Base64 => 3, + } + } + + fn from_index(i: usize) -> Self { + match i { + 1 => Self::Text, + 2 => Self::Hex, + 3 => Self::Base64, + _ => Self::Auto, + } + } +} + pub struct ConnectionView { conn_id: String, app: gpui_kit::WeakEntity, engine: Arc, - // 订阅 + // ── 订阅 ── sub_topic: Entity, sub_qos: Entity>, + sub_alias: Entity, + // v5 高级订阅选项 + sub_identifier: Entity, + sub_no_local: bool, + sub_rap: bool, + sub_retain_handling: Entity>, + sub_show_advanced: bool, + /// 编辑中的旧主题：提交时用旧主题移除原订阅，避免改名过程丢数据 + editing: Option, + /// 点击订阅项激活的消息过滤主题（再点一次取消） + sub_filter: Option, // 发布 pub_topic: Entity, @@ -56,12 +156,17 @@ pub struct ConnectionView { retain: bool, content_type: Entity, user_props: Entity, + msg_expiry: Entity, + response_topic: Entity, + correlation_data: Entity, show_props: bool, // 过滤与面板 filter: Entity, panel: Panel, expanded: Option, + msg_dir: DirFilter, + detail_format: DetailFormat, // 预设 preset_name: Entity, @@ -77,9 +182,12 @@ impl ConnectionView { ) -> Self { let sub_topic = cx.new(|cx| { - InputState::new(window, cx).placeholder("订阅主题，支持通配符 # +") + InputState::new(window, cx).placeholder("订阅主题，支持 # +，多个用逗号/空格分隔") }); let sub_qos = make_select(&QOS, 0, window, cx); + let sub_alias = cx.new(|cx| InputState::new(window, cx).placeholder("别名（可选）")); + let sub_identifier = cx.new(|cx| InputState::new(window, cx).placeholder("订阅标识符")); + let sub_retain_handling = make_select(&RETAIN_HANDLING, 0, window, cx); let pub_topic = cx.new(|cx| InputState::new(window, cx).placeholder("发布主题")); let payload = cx.new(|cx| { TextareaState::new(window, cx) @@ -88,6 +196,11 @@ impl ConnectionView { let pub_qos = make_select(&QOS, 0, window, cx); let payload_format = make_select(&PAYLOAD_FORMATS, 0, window, cx); let content_type = cx.new(|cx| InputState::new(window, cx).placeholder("Content-Type（可选）")); + let msg_expiry = cx.new(|cx| InputState::new(window, cx).placeholder("秒，如 60")); + let response_topic = + cx.new(|cx| InputState::new(window, cx).placeholder("Response Topic（可选）")); + let correlation_data = + cx.new(|cx| InputState::new(window, cx).placeholder("Correlation Data（可选）")); let filter = cx.new(|cx| InputState::new(window, cx).placeholder("过滤主题或内容…")); let user_props = cx.new(|cx| KvEditor::new("pub", &[], window, cx)); let preset_name = @@ -99,6 +212,14 @@ impl ConnectionView { engine, sub_topic, sub_qos, + sub_alias, + sub_identifier, + sub_no_local: false, + sub_rap: false, + sub_retain_handling, + sub_show_advanced: false, + editing: None, + sub_filter: None, pub_topic, payload, pub_qos, @@ -106,10 +227,15 @@ impl ConnectionView { retain: false, content_type, user_props, + msg_expiry, + response_topic, + correlation_data, show_props: false, filter, panel: Panel::Messages, expanded: None, + msg_dir: DirFilter::All, + detail_format: DetailFormat::Auto, preset_name, } } @@ -126,82 +252,285 @@ impl ConnectionView { .flatten() } + fn is_v5(&self, cx: &App) -> bool { + self.config(cx).map(|c| c.protocol.is_v5()).unwrap_or(true) + } + + // ── 订阅数据操作 ──────────────────────────────────────────────────────── + + /// 订阅按 (连接, 主题) 唯一：先摘掉旧条目再走 `add_subscription`， + /// 借它入列 + `save_subscriptions` 落盘的逻辑完成 upsert（storage 字段是私有的）。 + /// 落盘顺序以 add 时为准，入列后移回原位以保持界面顺序稳定。 + fn upsert_subscription(&self, sub: Subscription, cx: &mut Context) { + let Some(app_entity) = self.app.upgrade() else { + return; + }; + app_entity.update(cx, |app, cx| { + let slot = app + .subscriptions + .iter() + .position(|s| s.connection_id == sub.connection_id && s.topic == sub.topic); + app.subscriptions.retain(|s| { + !(s.connection_id == sub.connection_id && s.topic == sub.topic) + }); + let new_id = sub.id.clone(); + app.add_subscription(sub); + if let Some(pos) = slot + && let Some(from) = app.subscriptions.iter().position(|s| s.id == new_id) + { + let moved = app.subscriptions.remove(from); + app.subscriptions.insert(pos.min(app.subscriptions.len()), moved); + } + cx.notify(); + }); + cx.notify(); + } + + fn set_subscription_color(&mut self, sub_id: &str, hue: Option, cx: &mut Context) { + let sub = self + .with_app(cx, |app| { + app.subscriptions + .iter() + .find(|s| s.id == sub_id) + .cloned() + }) + .flatten(); + if let Some(mut sub) = sub { + sub.color = hue; + self.upsert_subscription(sub, cx); + } + } + + /// 启停订阅：置灰/恢复行显示，并在已连接时对引擎退订/重订。 + fn toggle_subscription_enabled(&mut self, sub_id: &str, cx: &mut Context) { + let Some(mut sub) = self + .with_app(cx, |app| { + app.subscriptions + .iter() + .find(|s| s.id == sub_id) + .cloned() + }) + .flatten() + else { + return; + }; + sub.enabled = !sub.enabled; + let enabled = sub.enabled; + let topic = sub.topic.clone(); + let conn = self.conn_id.clone(); + self.upsert_subscription(sub.clone(), cx); + if self.engine.is_connected(&conn) { + if enabled { + self.engine + .subscribe_with_options(conn, sub.clone(), SubscribeOptions::from(&sub)); + } else { + self.engine.unsubscribe(conn, topic); + } + } + } + + /// 回填输入区进入编辑态；旧订阅保留到提交时才替换，中途放弃不丢数据。 + fn start_edit_subscription(&mut self, sub: &Subscription, window: &mut Window, cx: &mut Context) { + self.editing = Some(sub.topic.clone()); + self.sub_topic + .update(cx, |s, cx| s.set_value(sub.topic.clone(), window, cx)); + let qos_idx = (sub.qos.min(2)) as usize; + self.sub_qos.update(cx, |s, cx| { + s.set_selected_index(Some(IndexPath::new(qos_idx)), window, cx) + }); + let alias = sub.alias.clone().unwrap_or_default(); + self.sub_alias + .update(cx, |s, cx| s.set_value(alias, window, cx)); + let ident = sub + .sub_identifier + .map(|v| v.to_string()) + .unwrap_or_default(); + self.sub_identifier + .update(cx, |s, cx| s.set_value(ident, window, cx)); + self.sub_no_local = sub.no_local; + self.sub_rap = sub.retain_as_published; + let rh_idx = sub.retain_handling.min(2) as usize; + self.sub_retain_handling.update(cx, |s, cx| { + s.set_selected_index(Some(IndexPath::new(rh_idx)), window, cx) + }); + self.sub_show_advanced = true; + cx.notify(); + } + + // ── 订阅 / 发布动作 ───────────────────────────────────────────────────── + fn do_subscribe(&mut self, window: &mut Window, cx: &mut Context) { - let topic = self.sub_topic.read(cx).value().to_string(); - if topic.trim().is_empty() { + let raw = self.sub_topic.read(cx).value().to_string(); + let topics = split_topic_list(&raw); + if topics.is_empty() { window.push_notification(Notification::warning("订阅主题不能为空"), cx); return; } let qos = self.sub_qos.read(cx).selected_value().copied().unwrap_or(0) as u8; - let sub = crate::model::Subscription::new(self.conn_id.clone(), topic, qos); - self.engine.subscribe(self.conn_id.clone(), sub); - self.sub_topic.update(cx, |s, cx| s.set_value("", window, cx)); + let alias_raw = self.sub_alias.read(cx).value().to_string(); + let alias = { + let t = alias_raw.trim(); + (!t.is_empty()).then(|| t.to_string()) + }; + let is_v5 = self.is_v5(cx); + // v5 选项只对 v5 连接生效，v4 下不读取（避免无效输入阻断订阅） + let mut sub_identifier = None; + if is_v5 { + let ident = self.sub_identifier.read(cx).value().to_string(); + let t = ident.trim(); + if !t.is_empty() { + match t.parse::() { + Ok(v) if v > 0 => sub_identifier = Some(v), + _ => { + window.push_notification( + Notification::warning("订阅标识符须为正整数"), + cx, + ); + return; + } + } + } + } + let no_local = self.sub_no_local; + let rap = self.sub_rap; + let retain_handling = self + .sub_retain_handling + .read(cx) + .selected_value() + .copied() + .unwrap_or(0) as u8; + + let connected = self.engine.is_connected(&self.conn_id); + let conn = self.conn_id.clone(); + + // 编辑提交：旧主题不在新列表里时先移除（含引擎退订），新主题由下方 upsert 覆盖 + if let Some(old_topic) = self.editing.take() + && !topics.contains(&old_topic) + { + if self.sub_filter.as_deref() == Some(old_topic.as_str()) { + self.sub_filter = None; + } + if let Some(app_entity) = self.app.upgrade() { + app_entity.update(cx, |app, cx| { + app.remove_subscription(&conn, &old_topic); + cx.notify(); + }); + } + } + + for topic in topics { + let mut sub = Subscription::new(conn.clone(), topic, qos); + sub.alias = alias.clone(); + if is_v5 { + sub.sub_identifier = sub_identifier; + sub.no_local = no_local; + sub.retain_as_published = rap; + sub.retain_handling = retain_handling; + } + self.upsert_subscription(sub.clone(), cx); + if connected { + self.engine.subscribe_with_options( + conn.clone(), + sub.clone(), + SubscribeOptions::from(&sub), + ); + } + } + if !connected { + let hint = if self + .config(cx) + .map(|c| c.auto_resubscribe) + .unwrap_or(true) + { + "未连接：订阅已保存，连接后自动恢复" + } else { + "未连接：订阅已保存" + }; + window.push_notification(Notification::warning(hint), cx); + } + + self.sub_topic + .update(cx, |s, cx| s.set_value("", window, cx)); + self.sub_alias + .update(cx, |s, cx| s.set_value("", window, cx)); + cx.notify(); } fn do_publish(&mut self, window: &mut Window, cx: &mut Context) { - let topic = self.pub_topic.read(cx).value().to_string(); - if topic.trim().is_empty() { + let mut params = self.collect_publish_params(cx); + if params.topic.trim().is_empty() { window.push_notification(Notification::warning("发布主题不能为空"), cx); return; } - let raw_payload = self.payload.read(cx).value().to_string(); + // 过期秒数输入非法时 collect 会静默得到 None，这里显式拦截避免误发 + let expiry_raw = self.msg_expiry.read(cx).value().to_string(); + if !expiry_raw.trim().is_empty() && params.message_expiry_interval.is_none() { + window.push_notification(Notification::warning("消息过期须为非负整数（秒）"), cx); + return; + } let vars = self.with_app(cx, |app| app.variables.clone()).unwrap_or_default(); - let topic = render_template(&topic, &vars); - let rendered = render_template(&raw_payload, &vars); + params.topic = render_template(¶ms.topic, &vars); + params.payload = render_template(¶ms.payload, &vars); + // {{变量}} 同样作用于用户属性的 key/value + params.user_properties = params + .user_properties + .into_iter() + .map(|(k, v)| (render_template(&k, &vars), render_template(&v, &vars))) + .collect(); - let format = PayloadFormat::ALL[ - self.payload_format.read(cx).selected_value().copied().unwrap_or(0) - ]; - if let PayloadFormat::Json = format - && let Err(e) = serde_json::from_str::(&rendered) { + if let PayloadFormat::Json = params.payload_format + && let Err(e) = serde_json::from_str::(¶ms.payload) { window.push_notification( Notification::warning(format!("JSON 格式无效: {e}")), cx, ); return; } - let encoded = match format.encode(&rendered) { - Ok(bytes) => bytes, + match params.payload_format.encode(¶ms.payload) { + Ok(bytes) => params.raw_bytes = Some(bytes), Err(e) => { window.push_notification(Notification::warning(e), cx); return; } - }; + } - let is_v5 = self.config(cx).map(|c| c.protocol.is_v5()).unwrap_or(true); - let mut params = PublishParams { - topic, - payload: rendered, - payload_format: format, - qos: self.pub_qos.read(cx).selected_value().copied().unwrap_or(0) as u8, - retain: self.retain, - raw_bytes: Some(encoded), - ..Default::default() - }; - if is_v5 { - let ct = self.content_type.read(cx).value().to_string(); - params.content_type = (!ct.is_empty()).then_some(ct); - params.user_properties = self.user_props.read(cx).pairs(cx); + // v3.1.1 忽略 v5 属性，避免残留输入带进发布报文 + if !self.is_v5(cx) { + params.user_properties.clear(); + params.content_type = None; + params.message_expiry_interval = None; + params.response_topic = None; + params.correlation_data = None; } self.engine.publish(self.conn_id.clone(), params); } - /// 读取当前发布面板上的参数（模板渲染前的原始值）。 + /// 读取当前发布面板上的参数（模板渲染前的原始值）；保存预设与发布共用， + /// 须覆盖全部 v5 属性与用户属性。 fn collect_publish_params(&self, cx: &App) -> PublishParams { let qos = self.pub_qos.read(cx).selected_value().copied().unwrap_or(0) as u8; let format = PayloadFormat::ALL [self.payload_format.read(cx).selected_value().copied().unwrap_or(0)]; + let opt = |s: String| { + let t = s.trim(); + (!t.is_empty()).then(|| t.to_string()) + }; PublishParams { topic: self.pub_topic.read(cx).value().to_string(), payload: self.payload.read(cx).value().to_string(), payload_format: format, qos, retain: self.retain, - content_type: { - let ct = self.content_type.read(cx).value().to_string(); - (!ct.is_empty()).then_some(ct) + user_properties: self.user_props.read(cx).pairs(cx), + content_type: opt(self.content_type.read(cx).value().to_string()), + message_expiry_interval: { + let t = self.msg_expiry.read(cx).value().to_string(); + let t = t.trim(); + if t.is_empty() { None } else { t.parse::().ok() } }, + response_topic: opt(self.response_topic.read(cx).value().to_string()), + correlation_data: opt(self.correlation_data.read(cx).value().to_string()), ..Default::default() } } @@ -221,6 +550,20 @@ impl ConnectionView { let ct = params.content_type.clone().unwrap_or_default(); self.content_type .update(cx, |s, cx| s.set_value(ct, window, cx)); + let expiry = params + .message_expiry_interval + .map(|v| v.to_string()) + .unwrap_or_default(); + self.msg_expiry + .update(cx, |s, cx| s.set_value(expiry, window, cx)); + let rt = params.response_topic.clone().unwrap_or_default(); + self.response_topic + .update(cx, |s, cx| s.set_value(rt, window, cx)); + let cd = params.correlation_data.clone().unwrap_or_default(); + self.correlation_data + .update(cx, |s, cx| s.set_value(cd, window, cx)); + // KvEditor 不支持运行时整体替换，直接重建实体回填用户属性 + self.user_props = cx.new(|cx| KvEditor::new("pub", ¶ms.user_properties, window, cx)); let qos_idx = params.qos.min(2) as usize; self.pub_qos.update(cx, |s, cx| { s.set_selected_index(Some(IndexPath::new(qos_idx)), window, cx) @@ -235,9 +578,11 @@ impl ConnectionView { self.retain = params.retain; cx.notify(); } +} - // ── 渲染 ────────────────────────────────────────────────────────────── +// ── 渲染 ────────────────────────────────────────────────────────────────────── +impl ConnectionView { fn render_top_bar(&self, cx: &mut Context) -> impl IntoElement { let id = self.conn_id.clone(); let (name, status, is_v5) = self @@ -315,30 +660,120 @@ impl ConnectionView { fn render_subscribe_bar(&mut self, cx: &mut Context) -> impl IntoElement { let sub_topic = self.sub_topic.clone(); let sub_qos = self.sub_qos.clone(); - h_flex() - .gap_2() - .px_3() - .h_12() - .items_center() + let sub_alias = self.sub_alias.clone(); + let is_v5 = self.is_v5(cx); + let editing = self.editing.is_some(); + let show_advanced = self.sub_show_advanced && is_v5; + + let mut bar = v_flex() .border_b_1() .border_color(cx.theme().border) - .bg(cx.theme().secondary) - .child(div().flex_1().min_w(px(0.)).child(Input::new(&sub_topic).small())) - .child(div().w(px(96.)).child(Select::new(&sub_qos).small())) - .child( - Button::new(SharedString::from(format!("subscribe-{}", self.conn_id))) - .icon(IconName::Plus) - .label("订阅") - .primary() - .small() - .on_click(cx.listener(|this, _, window, cx| this.do_subscribe(window, cx))), - ) + .bg(cx.theme().secondary); + bar = bar.child( + h_flex() + .gap_2() + .px_3() + .h_12() + .items_center() + .child(div().flex_1().min_w(px(0.)).child(Input::new(&sub_topic).small())) + .child(div().w(px(96.)).child(Select::new(&sub_qos).small())) + .child(div().w(px(112.)).child(Input::new(&sub_alias).small())) + .when(is_v5, |h| { + h.child( + Button::new(SharedString::from(format!("sub-adv-{}", self.conn_id))) + .icon(IconName::SlidersHorizontal) + .label("高级") + .ghost() + .small() + .when(show_advanced, |b| b.selected(true)) + .tooltip("MQTT 5 订阅选项") + .on_click(cx.listener(|this, _, _, cx| { + this.sub_show_advanced = !this.sub_show_advanced; + cx.notify(); + })), + ) + }) + .child( + Button::new(SharedString::from(format!("subscribe-{}", self.conn_id))) + .icon(if editing { IconName::Check } else { IconName::Plus }) + .label(if editing { "更新" } else { "订阅" }) + .primary() + .small() + .on_click(cx.listener(|this, _, window, cx| this.do_subscribe(window, cx))), + ) + .when(editing, |h| { + h.child( + Button::new(SharedString::from(format!("sub-edit-cancel-{}", self.conn_id))) + .icon(IconName::Close) + .ghost() + .small() + .tooltip("取消编辑") + .on_click(cx.listener(|this, _, _, cx| { + this.editing = None; + cx.notify(); + })), + ) + }), + ); + + if show_advanced { + let sub_identifier = self.sub_identifier.clone(); + let sub_no_local = self.sub_no_local; + let sub_rap = self.sub_rap; + let sub_retain_handling = self.sub_retain_handling.clone(); + bar = bar.child( + h_flex() + .gap_4() + .px_3() + .pb_2() + .items_start() + .child( + div() + .w(px(150.)) + .child(field("订阅标识符", Input::new(&sub_identifier).small())), + ) + .child( + h_flex() + .gap_1() + .items_center() + .pt_4() + .child(Switch::new("sub-no-local").checked(sub_no_local).on_change( + cx.listener(|this, v, _, cx| { + this.sub_no_local = *v; + cx.notify(); + }), + )) + .child(div().text_xs().child("No Local")), + ) + .child( + h_flex() + .gap_1() + .items_center() + .pt_4() + .child(Switch::new("sub-rap").checked(sub_rap).on_change( + cx.listener(|this, v, _, cx| { + this.sub_rap = *v; + cx.notify(); + }), + )) + .child(div().text_xs().child("Retain As Published")), + ) + .child( + div().w(px(150.)).child(field( + "Retain Handling", + Select::new(&sub_retain_handling).small(), + )), + ), + ); + } + bar } fn render_subscriptions(&self, cx: &mut Context) -> impl IntoElement { let hover_bg = cx.theme().muted; + let muted_fg = cx.theme().muted_foreground; let conn = self.conn_id.clone(); - let subs: Vec<_> = self + let subs: Vec = self .with_app(cx, |app| { app.subscriptions .iter() @@ -347,6 +782,7 @@ impl ConnectionView { .collect() }) .unwrap_or_default(); + let active_filter = self.sub_filter.clone(); let mut list = v_flex().gap_0p5(); if subs.is_empty() { @@ -369,9 +805,201 @@ impl ConnectionView { ); } for sub in subs { + let enabled = sub.enabled; + let is_active = enabled && active_filter.as_deref() == Some(sub.topic.as_str()); let topic = sub.topic.clone(); - let topic_for_click = topic.clone(); - let cid = self.conn_id.clone(); + let sub_id = sub.id.clone(); + let alias = sub.alias.clone().filter(|a| !a.trim().is_empty()); + let primary = alias.clone().unwrap_or_else(|| topic.clone()); + + // 颜色按钮：已设色显示色点，未设显示灰色「+」 + let color_btn = { + let weak_menu = cx.weak_entity(); + let sid = sub_id.clone(); + let trigger_id = SharedString::from(format!("sub-color-{}", sub_id)); + let trigger = if let Some(hue) = sub.color { + Button::new(trigger_id) + .ghost() + .xsmall() + .tooltip("订阅颜色") + .child( + div() + .size(px(11.)) + .rounded_full() + .bg(hue_color(hue)) + .border_1() + .border_color(cx.theme().border), + ) + } else { + Button::new(trigger_id) + .ghost() + .xsmall() + .tooltip("订阅颜色") + .child( + div() + .size(px(11.)) + .flex() + .items_center() + .justify_center() + .rounded_full() + .bg(cx.theme().muted) + .text_color(muted_fg) + .text_xs() + .child("+"), + ) + }; + trigger.dropdown_menu(move |mut menu, _, _| { + for (hue, name) in PRESET_HUES { + let weak = weak_menu.clone(); + let sid = sid.clone(); + let swatch = hue_color(hue); + let label = SharedString::from(format!("{name} {hue:.0}°")); + menu = menu.item( + PopupMenuItem::element(move |_, _| { + h_flex() + .gap_2() + .items_center() + .child( + div() + .size(px(10.)) + .rounded_full() + .bg(swatch) + .border_1() + .border_color(hsla(0., 0., 1., 0.6)), + ) + .child(div().text_sm().child(label.clone())) + }) + .on_click(move |_, _, cx| { + weak + .update(cx, |view, cx| { + view.set_subscription_color(&sid, Some(hue), cx); + }) + .ok(); + }), + ); + } + menu = menu.separator(); + let weak_rand = weak_menu.clone(); + let sid_rand = sid.clone(); + menu = menu.item(PopupMenuItem::new("随机颜色").on_click(move |_, _, cx| { + let hue = (uuid::Uuid::new_v4().as_u128() % 360) as f32; + weak_rand + .update(cx, |view, cx| { + view.set_subscription_color(&sid_rand, Some(hue), cx); + }) + .ok(); + })); + let weak_clear = weak_menu.clone(); + let sid_clear = sid.clone(); + menu = menu.item(PopupMenuItem::new("清除颜色").on_click(move |_, _, cx| { + weak_clear + .update(cx, |view, cx| { + view.set_subscription_color(&sid_clear, None, cx); + }) + .ok(); + })); + menu + }) + }; + + let eye_sid = sub_id.clone(); + let eye_btn = Button::new(SharedString::from(format!("sub-eye-{}", sub_id))) + .icon(if enabled { IconName::Eye } else { IconName::EyeOff }) + .ghost() + .xsmall() + .tooltip(if enabled { "停用订阅" } else { "启用订阅" }) + .on_click(cx.listener(move |this, _, _, cx| { + this.toggle_subscription_enabled(&eye_sid, cx); + })); + + let edit_sid = sub_id.clone(); + let edit_btn = Button::new(SharedString::from(format!("sub-edit-{}", sub_id))) + .icon(IconName::Pencil) + .ghost() + .xsmall() + .tooltip("编辑订阅") + .on_click(cx.listener(move |this, _, window, cx| { + let sub = this + .with_app(cx, |app| { + app.subscriptions + .iter() + .find(|s| s.id == edit_sid) + .cloned() + }) + .flatten(); + if let Some(sub) = sub { + this.start_edit_subscription(&sub, window, cx); + } + })); + + let del_cid = self.conn_id.clone(); + let del_topic = topic.clone(); + let del_weak_filter = self.sub_filter.clone(); + let del_btn = Button::new(SharedString::from(format!("sub-del-{}", sub_id))) + .icon(IconName::Close) + .ghost() + .xsmall() + .tooltip("取消订阅") + .on_click(cx.listener(move |this, _, _, cx| { + // 若删除的是当前过滤主题，一并清掉消息过滤 + if del_weak_filter.as_deref() == Some(del_topic.as_str()) { + this.sub_filter = None; + } + if let Some(app) = this.app.upgrade() { + app.update(cx, |app, cx| { + app.remove_subscription(&del_cid, &del_topic); + cx.notify(); + }); + } + this.editing = None; + cx.notify(); + })); + + // 点击区（非按钮）：切换按该主题过滤消息流；禁用的订阅不参与 + let filter_topic = topic.clone(); + let click_zone = h_flex() + .id(SharedString::from(format!("sub-click-{}", sub_id))) + .flex_1() + .min_w(px(0.)) + .when(enabled, |z| { + z.cursor_pointer().on_click(cx.listener( + move |this, _, _, cx| { + this.sub_filter = if this.sub_filter.as_deref() + == Some(filter_topic.as_str()) + { + None + } else { + Some(filter_topic.clone()) + }; + cx.notify(); + }, + )) + }) + .child( + v_flex() + .flex_1() + .min_w(px(0.)) + .gap_0p5() + .child( + div() + .text_sm() + .overflow_hidden() + .text_ellipsis() + .when(!enabled, |t| t.text_color(muted_fg)) + .child(primary), + ) + .when_some(alias, |v, _| { + v.child( + div() + .text_xs() + .text_color(muted_fg) + .overflow_hidden() + .text_ellipsis() + .child(topic.clone()), + ) + }), + ); + list = list.child( h_flex() .id(SharedString::from(format!("sub-{}-{}", self.conn_id, sub.id))) @@ -380,50 +1008,28 @@ impl ConnectionView { .px_2() .py_1p5() .rounded_md() - .hover(move |this| this.bg(hover_bg)) - .child( - gpui_kit::component::Icon::new(IconName::Hash) - .xsmall() - .text_color(cx.theme().muted_foreground), - ) - .child( - div() - .flex_1() - .min_w(px(0.)) - .text_sm() - .overflow_hidden() - .text_ellipsis() - .child(topic), - ) + .when(is_active, |t| t.bg(cx.theme().secondary)) + .when(!is_active, |t| t.hover(move |this| this.bg(hover_bg))) + .when(!enabled, |t| t.opacity(0.55)) + .child(color_btn) + .child(click_zone) .child( div() .text_xs() .px_1() .rounded_sm() .bg(cx.theme().muted) - .text_color(cx.theme().muted_foreground) + .text_color(muted_fg) .child(format!("Q{}", sub.qos)), ) - .child( - Button::new(SharedString::from(format!("sub-del-{}", sub.id))) - .icon(IconName::Close) - .ghost() - .xsmall() - .tooltip("取消订阅") - .on_click(cx.listener(move |this, _, _, cx| { - if let Some(app) = this.app.upgrade() { - app.update(cx, |app, cx| { - app.remove_subscription(&cid, &topic_for_click); - cx.notify(); - }); - } - })), - ), + .child(eye_btn) + .child(edit_btn) + .child(del_btn), ); } v_flex() - .w(px(220.)) + .w(px(248.)) .flex_shrink_0() .h_full() .bg(cx.theme().sidebar) @@ -438,7 +1044,12 @@ impl ConnectionView { .child(div().flex_1().overflow_y_scrollbar().px_2().child(list)) } - fn render_message_row(&self, record: &MqttRecord, cx: &mut Context) -> impl IntoElement { + fn render_message_row( + &self, + record: &MqttRecord, + sub_color: Option, + cx: &mut Context, + ) -> impl IntoElement { let seq = record.seq; let expanded = self.expanded == Some(seq); let received = matches!(record.direction, Direction::Received); @@ -452,112 +1063,184 @@ impl ConnectionView { let muted = cx.theme().muted_foreground; let mono = cx.theme().mono_font_family.clone(); let direction_label = if received { "接收" } else { "发送" }; + let show_millis = self + .with_app(cx, |app| app.settings.show_millis) + .unwrap_or(true); - let mut row = v_flex() - .id(SharedString::from(format!( - "msg-{}-{}", - record.connection_id, record.seq - ))) - .w_full() - .px_3() - .py_2() - .border_b_1() - .border_color(border) - .hover(move |this| this.bg(hover_bg)) - .cursor_pointer() + let header = h_flex() + .gap_2() + .items_center() .child( - h_flex() - .gap_2() + div() + .size(px(18.)) + .flex() .items_center() + .justify_center() + .rounded_full() + .bg(accent.alpha(0.14)) .child( - div() - .size(px(18.)) - .flex() - .items_center() - .justify_center() - .rounded_full() - .bg(accent.alpha(0.14)) - .child( - gpui_kit::component::Icon::new(if received { - IconName::ArrowDown - } else { - IconName::ArrowUp - }) - .size_3() - .text_color(accent), - ), - ) - .child( - div() - .flex_1() - .min_w(px(0.)) - .text_sm() - .font_medium() - .overflow_hidden() - .text_ellipsis() - .font_family(mono.clone()) - .child(record.topic.clone()), - ) - .child( - div() - .text_xs() - .text_color(muted) - .font_family(mono.clone()) - .child(direction_label), - ) - .child( - div() - .text_xs() - .px_1() - .rounded_sm() - .bg(cx.theme().muted) - .text_color(muted) - .font_family(mono.clone()) - .child(format!("Q{}", record.qos)), - ) - .when(record.retain, |t| { - t.child( - div() - .text_xs() - .px_1() - .rounded_sm() - .bg(cx.theme().warning.alpha(0.15)) - .text_color(cx.theme().warning) - .child("retain"), - ) - }) - .child( - div() - .text_xs() - .text_color(muted) - .font_family(mono.clone()) - .child(format_time(record.timestamp, true)), + gpui_kit::component::Icon::new(if received { + IconName::ArrowDown + } else { + IconName::ArrowUp + }) + .size_3() + .text_color(accent), ), ) + .child( + div() + .flex_1() + .min_w(px(0.)) + .text_sm() + .font_medium() + .overflow_hidden() + .text_ellipsis() + .font_family(mono.clone()) + // 匹配到订阅色时主题文字着色，否则保持默认前景色 + .when_some(sub_color, |t, hue| t.text_color(hue_color(hue))) + .child(record.topic.clone()), + ) + .child( + div() + .text_xs() + .text_color(muted) + .font_family(mono.clone()) + .child(direction_label), + ) + .child( + div() + .text_xs() + .px_1() + .rounded_sm() + .bg(cx.theme().muted) + .text_color(muted) + .font_family(mono.clone()) + .child(format!("Q{}", record.qos)), + ) + .when(record.retain, |t| { + t.child( + div() + .text_xs() + .px_1() + .rounded_sm() + .bg(cx.theme().warning.alpha(0.15)) + .text_color(cx.theme().warning) + .child("retain"), + ) + }) + .child( + div() + .text_xs() + .text_color(muted) + .font_family(mono.clone()) + .child(format_time(record.timestamp, show_millis)), + ); + + let mut body = v_flex() + .flex_1() + .min_w(px(0.)) + .px_3() + .py_2() + .child(header) .child( div() .text_sm() .text_color(cx.theme().foreground) .pt_0p5() .font_family(mono.clone()) - .child(preview_payload(&record.payload, expanded)), + .child(preview_payload(&record.payload)), ); if expanded { + let fmt = self.detail_format; + let detail_text = format_payload_detail(&record.payload, fmt); let mut details = v_flex().gap_1p5().pt_1(); - if let Ok(v) = serde_json::from_str::(&record.payload) { - details = details.child( - div() - .font_family(mono.clone()) - .text_xs() - .p_2() - .rounded_md() - .bg(cx.theme().background) - .border_1() - .border_color(border) - .child(serde_json::to_string_pretty(&v).unwrap_or_default()), + + // 格式切换 + 复制按钮；stop_propagation 避免点击时又收起整行 + let fmt_group = ButtonGroup::new(SharedString::from(format!("msg-fmt-{}", seq))) + .compact() + .children( + [DetailFormat::Auto, DetailFormat::Text, DetailFormat::Hex, DetailFormat::Base64] + .into_iter() + .map(|f| { + Button::new(SharedString::from(format!( + "msg-fmt-btn-{}-{}", + seq, + f.index() + ))) + .label(f.label()) + .small() + .selected(f == fmt) + }), + ) + .on_click(cx.listener(|this, ixs: &Vec, _, cx| { + cx.stop_propagation(); + if let Some(&i) = ixs.first() { + this.detail_format = DetailFormat::from_index(i); + cx.notify(); + } + })); + + let copy_topic = record.topic.clone(); + let copy_payload = record.payload.clone(); + let copy_detail = details_copy_text(record, show_millis); + let detail_toolbar = h_flex() + .gap_1() + .items_center() + .child(fmt_group) + .child(div().flex_1()) + .child( + Button::new(SharedString::from(format!("msg-copy-topic-{}", seq))) + .icon(IconName::Copy) + .label("主题") + .ghost() + .xsmall() + .tooltip("复制主题") + .on_click(move |_, window, cx| { + cx.stop_propagation(); + cx.write_to_clipboard(ClipboardItem::new_string(copy_topic.clone())); + window.push_notification(Notification::success("已复制主题"), cx); + }), + ) + .child( + Button::new(SharedString::from(format!("msg-copy-payload-{}", seq))) + .icon(IconName::Copy) + .label("负载") + .ghost() + .xsmall() + .tooltip("复制负载原文") + .on_click(move |_, window, cx| { + cx.stop_propagation(); + cx.write_to_clipboard(ClipboardItem::new_string(copy_payload.clone())); + window.push_notification(Notification::success("已复制负载"), cx); + }), + ) + .child( + Button::new(SharedString::from(format!("msg-copy-detail-{}", seq))) + .icon(IconName::Copy) + .label("详情") + .ghost() + .xsmall() + .tooltip("复制完整详情（含 v5 属性）") + .on_click(move |_, window, cx| { + cx.stop_propagation(); + cx.write_to_clipboard(ClipboardItem::new_string(copy_detail.clone())); + window.push_notification(Notification::success("已复制详情"), cx); + }), ); - } + details = details.child(detail_toolbar).child( + div() + .font_family(mono.clone()) + .text_xs() + .p_2() + .rounded_md() + .bg(cx.theme().background) + .border_1() + .border_color(border) + .child(detail_text), + ); + if !record.user_properties.is_empty() { let props: String = record .user_properties @@ -578,6 +1261,16 @@ impl ConnectionView { ("Content-Type", record.content_type.clone()), ("Response Topic", record.response_topic.clone()), ("Correlation Data", record.correlation_data.clone()), + ( + "消息过期", + record + .message_expiry_interval + .map(|v| format!("{v} 秒")), + ), + ( + "订阅标识符", + record.subscription_identifier.map(|v| v.to_string()), + ), ] { if let Some(v) = value { details = details.child( @@ -589,7 +1282,7 @@ impl ConnectionView { ); } } - row = row.child( + body = body.child( div() .mt_1() .p_2() @@ -598,41 +1291,88 @@ impl ConnectionView { .child(details), ); } - row.on_click(cx.listener(move |this, _, _, cx| { - this.expanded = if this.expanded == Some(seq) { - None - } else { - Some(seq) - }; - cx.notify(); - })) + + h_flex() + .id(SharedString::from(format!( + "msg-{}-{}", + record.connection_id, record.seq + ))) + .w_full() + .border_b_1() + .border_color(border) + .hover(move |this| this.bg(hover_bg)) + .cursor_pointer() + // 左侧色条：有订阅色时显示，否则透明以保持行内对齐 + .child( + div() + .w(px(3.)) + .self_stretch() + .rounded_full() + .bg(sub_color.map(hue_color).unwrap_or(hsla(0., 0., 0., 0.))), + ) + .child(body) + .on_click(cx.listener(move |this, _, _, cx| { + this.expanded = if this.expanded == Some(seq) { + None + } else { + Some(seq) + }; + cx.notify(); + })) } fn render_messages(&self, cx: &mut Context) -> impl IntoElement { let query = self.filter.read(cx).value().to_lowercase(); let conn = self.conn_id.clone(); - let records: Vec = self + let dir = self.msg_dir; + let sub_filter = self.sub_filter.clone(); + let (records, subs): (Vec, Vec) = self .with_app(cx, |app| { - app.messages + let subs: Vec = app + .subscriptions + .iter() + .filter(|s| s.connection_id == conn) + .cloned() + .collect(); + let records: Vec = app + .messages .get(&conn) .map(|m| { m.iter() .rev() - .take(MAX_RENDERED_MESSAGES) + // 方向 / 搜索 / 订阅过滤叠加，过滤后再截断保证最新消息优先 .filter(|r| { - query.is_empty() + let dir_ok = match dir { + DirFilter::All => true, + DirFilter::Received => { + r.direction == Direction::Received + } + DirFilter::Published => { + r.direction == Direction::Published + } + }; + let query_ok = query.is_empty() || r.topic.to_lowercase().contains(&query) - || r.payload.to_lowercase().contains(&query) + || r.payload.to_lowercase().contains(&query); + let sub_ok = sub_filter + .as_deref() + .map_or(true, |f| topic_matches(f, &r.topic)); + dir_ok && query_ok && sub_ok }) + .take(MAX_RENDERED_MESSAGES) .cloned() .collect::>() }) - .unwrap_or_default() + .unwrap_or_default(); + (records, subs) }) .unwrap_or_default(); let mut body = v_flex().flex_1().overflow_y_scrollbar(); if records.is_empty() { + let filtering = !query.is_empty() + || sub_filter.is_some() + || dir != DirFilter::All; body = body.child( v_flex() .h_full() @@ -658,20 +1398,84 @@ impl ConnectionView { div() .text_sm() .text_color(cx.theme().muted_foreground) - .child(if query.is_empty() { - "暂无消息，订阅主题后消息会显示在这里" - } else { + .child(if filtering { "没有匹配的消息" + } else { + "暂无消息，订阅主题后消息会显示在这里" }), ), ); } else { // 最新消息在最上方（与 MQTTX 一致，避免长列表需要手动滚底） for r in &records { - body = body.child(self.render_message_row(r, cx)); + let color = pick_subscription_color(&subs, &r.topic); + body = body.child(self.render_message_row(r, color, cx)); } } + // 方向过滤 seg + let dir_group = ButtonGroup::new(SharedString::from(format!( + "msg-dir-{}", + self.conn_id + ))) + .compact() + .children([DirFilter::All, DirFilter::Received, DirFilter::Published].into_iter().map( + |d| { + Button::new(SharedString::from(format!( + "msg-dir-btn-{}-{}", + self.conn_id, + d.index() + ))) + .label(d.label()) + .small() + .selected(d == dir) + }, + )) + .on_click(cx.listener(|this, ixs: &Vec, _, cx| { + if let Some(&i) = ixs.first() { + this.msg_dir = DirFilter::from_index(i); + cx.notify(); + } + })); + + // 订阅过滤 chip（点击订阅项激活） + let filter_chip = self.sub_filter.clone().map(|topic| { + h_flex() + .gap_1() + .items_center() + .px_1p5() + .py_0p5() + .rounded_md() + .bg(cx.theme().secondary) + .max_w(px(180.)) + .child( + gpui_kit::component::Icon::new(IconName::ListFilter) + .xsmall() + .text_color(cx.theme().muted_foreground), + ) + .child( + div() + .text_xs() + .overflow_hidden() + .text_ellipsis() + .child(topic), + ) + .child( + Button::new(SharedString::from(format!( + "msg-filter-clear-{}", + self.conn_id + ))) + .icon(IconName::Close) + .ghost() + .xsmall() + .tooltip("清除订阅过滤") + .on_click(cx.listener(|this, _, _, cx| { + this.sub_filter = None; + cx.notify(); + })), + ) + }); + v_flex() .flex_1() .min_w(px(0.)) @@ -685,6 +1489,7 @@ impl ConnectionView { .items_center() .border_b_1() .border_color(cx.theme().border) + .child(dir_group) .child( div() .flex_1() @@ -693,12 +1498,32 @@ impl ConnectionView { gpui_kit::component::Icon::new(IconName::Search).small(), )), ) + .children(filter_chip) .child( div() .text_xs() .px_1p5() .text_color(cx.theme().muted_foreground) .child(format!("{} 条", records.len())), + ) + .child( + Button::new(SharedString::from(format!("msg-clear-{}", self.conn_id))) + .icon(IconName::Eraser) + .label("清空") + .ghost() + .small() + .tooltip("清空当前连接的消息") + .on_click(cx.listener(|this, _, _, cx| { + let id = this.conn_id.clone(); + this.expanded = None; + if let Some(app) = this.app.upgrade() { + app.update(cx, |app, cx| { + app.messages.remove(&id); + cx.notify(); + }); + } + cx.notify(); + })), ), ) .child(body) @@ -707,6 +1532,9 @@ impl ConnectionView { fn render_logs(&self, cx: &mut Context) -> impl IntoElement { let muted = cx.theme().muted_foreground; let conn = self.conn_id.clone(); + let show_millis = self + .with_app(cx, |app| app.settings.show_millis) + .unwrap_or(true); let logs: Vec<_> = self .with_app(cx, |app| { app.logs @@ -758,7 +1586,7 @@ impl ConnectionView { .text_xs() .text_color(muted) .font_family(cx.theme().mono_font_family.clone()) - .child(format_time(l.timestamp, true)), + .child(format_time(l.timestamp, show_millis)), ) .child( div() @@ -800,20 +1628,19 @@ impl ConnectionView { ) .child(body) } - fn render_publish_bar(&mut self, cx: &mut Context) -> impl IntoElement { let pub_topic = self.pub_topic.clone(); let payload = self.payload.clone(); let pub_qos = self.pub_qos.clone(); let payload_format = self.payload_format.clone(); let content_type = self.content_type.clone(); + let msg_expiry = self.msg_expiry.clone(); + let response_topic = self.response_topic.clone(); + let correlation_data = self.correlation_data.clone(); let user_props = self.user_props.clone(); let retain = self.retain; let show_props = self.show_props; - let is_v5 = self - .config(cx) - .map(|c| c.protocol.is_v5()) - .unwrap_or(true); + let is_v5 = self.is_v5(cx); // 发布预设 let presets = self.with_app(cx, |app| app.presets.clone()).unwrap_or_default(); @@ -831,7 +1658,22 @@ impl ConnectionView { h_flex() .gap_2() .items_center() - .child(div().flex_1().min_w(px(0.)).child(Input::new(&pub_topic).small())) + // Ctrl+Enter 快捷发送（焦点在主题输入上时） + .child( + div() + .id(SharedString::from(format!("pub-topic-wrap-{}", self.conn_id))) + .flex_1() + .min_w(px(0.)) + .on_key_down(cx.listener( + |this, ev: &gpui_kit::KeyDownEvent, window, cx| { + if ev.keystroke.modifiers.control && ev.keystroke.key == "enter" { + cx.stop_propagation(); + this.do_publish(window, cx); + } + }, + )) + .child(Input::new(&pub_topic).small()), + ) .child(div().w(px(92.)).child(Select::new(&pub_qos).small())) .child(div().w(px(124.)).child(Select::new(&payload_format).small())) .child(self.render_preset_menu(presets, cx)) @@ -873,26 +1715,60 @@ impl ConnectionView { card = card.child( div() + .id(SharedString::from(format!("payload-wrap-{}", self.conn_id))) .h(px(110.)) .rounded_md() .border_1() .border_color(cx.theme().border) .bg(cx.theme().background) .p_1() + // Ctrl+Enter 快捷发送（焦点在负载输入上时） + .on_key_down(cx.listener( + |this, ev: &gpui_kit::KeyDownEvent, window, cx| { + if ev.keystroke.modifiers.control && ev.keystroke.key == "enter" { + cx.stop_propagation(); + this.do_publish(window, cx); + } + }, + )) .child(Textarea::new(&payload)), ); if show_props && is_v5 { card = card.child( - h_flex() + v_flex() .gap_2() .w_full() .child( - div() - .w(px(240.)) - .child(field("Content-Type", Input::new(&content_type).small())), + h_flex() + .gap_2() + .w_full() + .child( + div().w(px(220.)).child(field( + "Content-Type", + Input::new(&content_type).small(), + )), + ) + .child( + div().w(px(120.)).child(field( + "消息过期(秒)", + Input::new(&msg_expiry).small(), + )), + ) + .child( + div().flex_1().min_w(px(0.)).child(field( + "Response Topic", + Input::new(&response_topic).small(), + )), + ) + .child( + div().flex_1().min_w(px(0.)).child(field( + "Correlation Data", + Input::new(&correlation_data).small(), + )), + ), ) - .child(div().flex_1().min_w(px(0.)).child(user_props.clone())), + .child(user_props.clone()), ); } card @@ -1039,6 +1915,7 @@ impl ConnectionView { ); return; } + // collect 已覆盖用户属性与全部 v5 字段 let params = weak .read_with(cx, |view, cx| { view.collect_publish_params(cx) @@ -1116,15 +1993,123 @@ impl Render for ConnectionView { } } -fn preview_payload(payload: &str, expanded: bool) -> String { - if expanded { - payload.to_string() - } else { - let one_line: String = payload.chars().take(200).collect(); - if payload.len() > 200 { - format!("{one_line}…") - } else { - one_line +// ─── 独立工具函数 ───────────────────────────────────────────────────────────── + +/// 色相（度）→ gpui Hsla（h 分量为 0..=1 的整圆比例） +fn hue_color(hue: f32) -> Hsla { + hsla(hue.rem_euclid(360.0) / 360.0, 0.72, 0.52, 1.0) +} + +/// 逗号 / 空白 / 换行分隔的多主题输入 → 去重后的主题列表（批量订阅） +fn split_topic_list(raw: &str) -> Vec { + let mut out: Vec = Vec::new(); + for part in raw.split(|c: char| c == ',' || c.is_whitespace()) { + let t = part.trim(); + if !t.is_empty() && !out.iter().any(|x| x == t) { + out.push(t.to_string()); + } + } + out +} + +/// MQTT 主题过滤器匹配：`+` 匹配单层，`#` 匹配本层及以下（可匹配父级本身）。 +fn topic_matches(filter: &str, topic: &str) -> bool { + let f: Vec<&str> = filter.split('/').collect(); + let t: Vec<&str> = topic.split('/').collect(); + for (i, part) in f.iter().enumerate() { + if *part == "#" { + // "a/#" 需要匹配 "a" 自身：剩余层级数 >= 过滤器已消费层数 + return i <= t.len(); + } + if i >= t.len() { + return false; + } + if *part != "+" && *part != t[i] { + return false; + } + } + f.len() == t.len() +} + +/// 挑选消息着色用的订阅色：先看精确匹配，再看通配匹配；只考虑启用且已设色的订阅。 +fn pick_subscription_color(subs: &[Subscription], topic: &str) -> Option { + let has_wild = |t: &str| t.contains('+') || t.contains('#'); + for s in subs.iter().filter(|s| s.enabled) { + if s.topic == topic && !has_wild(&s.topic) && s.color.is_some() { + return s.color; + } + } + for s in subs.iter().filter(|s| s.enabled) { + if has_wild(&s.topic) && topic_matches(&s.topic, topic) && s.color.is_some() { + return s.color; + } + } + None +} + +/// 展开详情里的 payload 渲染：自动 = 合法 JSON 美化，否则原文。 +fn format_payload_detail(payload: &str, 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() + .iter() + .map(|b| format!("{b:02x}")) + .collect::>() + .join(" "), + DetailFormat::Base64 => { + use base64::Engine as _; + base64::engine::general_purpose::STANDARD.encode(payload.as_bytes()) } } } + +/// 「复制详情」的可读文本：元数据 + v5 属性 + 用户属性 + 负载。 +fn details_copy_text(record: &MqttRecord, show_millis: bool) -> String { + let mut out = String::new(); + out.push_str(&format!("Topic: {}\n", record.topic)); + out.push_str(&format!("Direction: {}\n", record.direction.label())); + out.push_str(&format!("QoS: {}\n", record.qos)); + out.push_str(&format!("Retain: {}\n", record.retain)); + out.push_str(&format!( + "Time: {}\n", + format_time(record.timestamp, show_millis) + )); + if let Some(v) = record.content_type.as_deref() { + out.push_str(&format!("Content-Type: {v}\n")); + } + if let Some(v) = record.response_topic.as_deref() { + out.push_str(&format!("Response Topic: {v}\n")); + } + if let Some(v) = record.correlation_data.as_deref() { + out.push_str(&format!("Correlation Data: {v}\n")); + } + if let Some(v) = record.message_expiry_interval { + out.push_str(&format!("Message Expiry Interval: {v}\n")); + } + if let Some(v) = record.subscription_identifier { + out.push_str(&format!("Subscription Identifier: {v}\n")); + } + if !record.user_properties.is_empty() { + out.push_str("User Properties:\n"); + for (k, v) in &record.user_properties { + out.push_str(&format!(" {k} = {v}\n")); + } + } + out.push_str(&format!("Payload:\n{}", record.payload)); + out +} + +/// 折叠态只显示单行预览；完整内容由展开详情按所选格式渲染。 +fn preview_payload(payload: &str) -> String { + let one_line: String = payload.chars().take(200).collect(); + if payload.chars().count() > 200 { + format!("{one_line}…") + } else { + one_line + } +} diff --git a/gpui-app/src/ui/settings_dialog.rs b/gpui-app/src/ui/settings_dialog.rs index dbf3a21..a6e96e1 100644 --- a/gpui-app/src/ui/settings_dialog.rs +++ b/gpui-app/src/ui/settings_dialog.rs @@ -1,11 +1,15 @@ -//! 应用设置对话框：主题、消息缓存条数、时间戳显示。 +//! 应用设置对话框：主题、消息缓存、时间戳、自动检查更新、数据/日志目录、关于。 + +use std::path::{Path, PathBuf}; use gpui_kit::component::button::{Button, ButtonVariants as _}; +use gpui_kit::component::clipboard::Clipboard; use gpui_kit::component::input::{Input, InputState}; use gpui_kit::component::select::{Select, SelectState}; use gpui_kit::component::switch::Switch; use gpui_kit::component::{ - h_flex, notification::Notification, v_flex, WindowExt as _, + h_flex, notification::Notification, v_flex, ActiveTheme as _, Sizable as _, StyledExt as _, + WindowExt as _, }; use gpui_kit::{ div, px, App, AppContext as _, Context, Entity, IntoElement, ParentElement as _, Render, @@ -15,17 +19,30 @@ use gpui_kit::{ use crate::model::{AppSettings, ThemeModePref}; use crate::ui::app::MqttXApp; use crate::ui::widgets::{field, make_select, OptionDelegate}; +use crate::ui::IconName; const THEMES: [&str; 3] = ["跟随系统", "浅色", "深色"]; +const SITE_URL: &str = "https://mqttx.app"; +const REPO_URL: &str = "https://gitea.heavenlybook.cn/JoeAllen/mqttx"; + struct SettingsDialog { theme: Entity>, max_messages: Entity, show_millis: bool, + auto_check_update: bool, + data_dir: PathBuf, + log_dir: PathBuf, } impl SettingsDialog { - fn new(initial: AppSettings, window: &mut Window, cx: &mut Context) -> Self { + fn new( + initial: AppSettings, + data_dir: PathBuf, + log_dir: PathBuf, + window: &mut Window, + cx: &mut Context, + ) -> Self { let theme_idx = match initial.theme { ThemeModePref::System => 0, ThemeModePref::Light => 1, @@ -37,6 +54,9 @@ impl SettingsDialog { InputState::new(window, cx).default_value(initial.max_messages.to_string()) }), show_millis: initial.show_millis, + auto_check_update: initial.auto_check_update, + data_dir, + log_dir, } } @@ -54,20 +74,41 @@ impl SettingsDialog { .parse::() .map_err(|_| "消息缓存条数必须是正整数".to_string())? .max(100); + // 四个字段全部显式写出，避免 ..Default::default() 把开关项重置为默认值 Ok(AppSettings { theme, max_messages, show_millis: self.show_millis, - ..Default::default() + auto_check_update: self.auto_check_update, }) } } +/// 用系统文件管理器打开目录；失败弹 toast（不写日志，避免与日志落盘互相触发）。 +fn open_in_file_manager(path: &Path, window: &mut Window, cx: &mut App) { + #[cfg(target_os = "windows")] + let mut cmd = std::process::Command::new("explorer"); + #[cfg(target_os = "macos")] + let mut cmd = std::process::Command::new("open"); + #[cfg(all(unix, not(target_os = "macos")))] + let mut cmd = std::process::Command::new("xdg-open"); + + cmd.arg(path); + if let Err(e) = cmd.spawn() { + window.push_notification(Notification::error(format!("打开目录失败: {e}")), cx); + } +} + impl Render for SettingsDialog { fn render(&mut self, _window: &mut Window, cx: &mut Context) -> impl IntoElement { + let data_dir = self.data_dir.to_string_lossy().into_owned(); + let log_dir = self.log_dir.to_string_lossy().into_owned(); + let data_dir_for_open = self.data_dir.clone(); + let log_dir_for_open = self.log_dir.clone(); + v_flex() .gap_4() - .w(px(420.)) + .w(px(500.)) .child(field("主题", Select::new(&self.theme))) .child(field( "每条连接内存中保留的消息条数", @@ -86,20 +127,140 @@ impl Render for SettingsDialog { })), ), ) + .child( + h_flex() + .justify_between() + .child(div().text_sm().child("自动检查更新")) + .child( + Switch::new("auto-check-update") + .checked(self.auto_check_update) + .on_change(cx.listener(|this, v, _, cx| { + this.auto_check_update = *v; + cx.notify(); + })), + ), + ) + .child(div().h(px(1.)).bg(cx.theme().border).w_full()) + .child(field( + "数据目录", + h_flex() + .gap_1() + .w_full() + .child( + div() + .flex_1() + .min_w(px(0.)) + .text_xs() + .whitespace_nowrap() + .overflow_hidden() + .text_ellipsis() + .text_color(cx.theme().muted_foreground) + .child(data_dir.clone()), + ) + .child(Clipboard::new("copy-data-dir").value(data_dir.clone())) + .child( + Button::new("open-data-dir") + .icon(IconName::FolderOpen) + .label("打开目录") + .outline() + .xsmall() + .on_click(move |_, window, cx| { + open_in_file_manager(&data_dir_for_open, window, cx); + }), + ), + )) + .child(field( + "日志目录", + h_flex() + .gap_1() + .w_full() + .child( + div() + .flex_1() + .min_w(px(0.)) + .text_xs() + .whitespace_nowrap() + .overflow_hidden() + .text_ellipsis() + .text_color(cx.theme().muted_foreground) + .child(log_dir.clone()), + ) + .child( + Button::new("open-log-dir") + .icon(IconName::FolderOpen) + .label("打开日志目录") + .outline() + .xsmall() + .on_click(move |_, window, cx| { + // 日志目录可能尚未创建（从未产生过日志） + if let Err(e) = std::fs::create_dir_all(&log_dir_for_open) { + window.push_notification( + Notification::error(format!("创建日志目录失败: {e}")), + cx, + ); + return; + } + open_in_file_manager(&log_dir_for_open, window, cx); + }), + ), + )) + .child(div().h(px(1.)).bg(cx.theme().border).w_full()) + .child( + v_flex().gap_1().child( + v_flex() + .gap_1() + .child(div().text_sm().font_semibold().child("关于")) + .child( + h_flex() + .gap_2() + .text_xs() + .text_color(cx.theme().muted_foreground) + .child(format!("版本 {}", env!("CARGO_PKG_VERSION"))) + .child("Apache-2.0 许可证"), + ) + .child( + div() + .text_xs() + .text_color(cx.theme().muted_foreground) + .child("对标 MQTTX 的 GPUI 跨平台 MQTT 调试客户端"), + ) + .child( + h_flex().gap_1().child( + Button::new("about-site") + .label("mqttx.app") + .ghost() + .xsmall() + .tooltip(SITE_URL) + .on_click(|_, _, cx| cx.open_url(SITE_URL)), + ).child( + Button::new("about-repo") + .label("gitea.heavenlybook.cn/JoeAllen/mqttx") + .ghost() + .xsmall() + .tooltip(REPO_URL) + .on_click(|_, _, cx| cx.open_url(REPO_URL)), + ), + ), + ), + ) } } pub fn open(app: Entity, window: &mut Window, cx: &mut App) { - let initial = app.read(cx).settings.clone(); - let dialog_view: Entity = - cx.new(|cx| SettingsDialog::new(initial, window, cx)); + let (initial, data_dir, log_dir) = { + let state = app.read(cx); + (state.settings.clone(), state.data_dir(), state.log_dir()) + }; + let dialog_view: Entity = cx.new(|cx| { + SettingsDialog::new(initial, data_dir, log_dir, window, cx) + }); let app_save = app.clone(); window.open_dialog(cx, move |dialog, _window, _cx| { let body = dialog_view.clone(); let for_collect = dialog_view.clone(); let app_for_save = app_save.clone(); dialog - .w(px(480.)) + .w(px(560.)) .title("设置") .child(body) .footer(