From 0152a10ee6506652884c62b998cfdc7cac87cf85 Mon Sep 17 00:00:00 2001 From: JoeAllen <2494136784@qq.com> Date: Fri, 4 Sep 2026 01:45:19 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E9=98=BF=E9=87=8C=E4=BA=91=20MQTT=20?= =?UTF-8?q?=E8=BF=9E=E6=8E=A5=E9=A2=84=E8=AE=BE=E3=80=81=E8=BF=9E=E6=8E=A5?= =?UTF-8?q?=E5=88=86=E7=BB=84=E4=B8=8E=20MQTT=203.1.1=20=E6=94=AF=E6=8C=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 后端引入 rumqttc v4 客户端，支持 MQTT 3.1.1 连接/发布/订阅，协议版本随连接下发 - 连接表单：阿里云预设管理（分组、导入导出、去抖自动保存），退出云模式需确认，mqtts 时端口跟随 8883 - 连接列表：分组列就地编辑（输入即创建），空分组提示优化 - 导入按内容去重并提示跳过数量，导出提示含明文密钥 - 调试日志密码脱敏；修复多处取消弹窗的 unhandled rejection --- src-tauri/Cargo.lock | 24 + src-tauri/Cargo.toml | 4 +- src-tauri/src/lib.rs | 700 +++++++++++++----- src/components/connections/ConnectionForm.vue | 593 +++++++++++++-- src/components/connections/ConnectionList.vue | 319 +++++--- src/components/dashboard/SendPanel.vue | 10 +- src/stores/useAliyunPresets.ts | 125 ++++ src/stores/useConnections.ts | 90 ++- src/stores/useMqttBridge.ts | 4 + src/types/mqtt.ts | 102 ++- 10 files changed, 1626 insertions(+), 345 deletions(-) create mode 100644 src/stores/useAliyunPresets.ts diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index 5060f9c..a9e2bc7 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -2248,6 +2248,7 @@ dependencies = [ name = "mqttx" version = "0.1.0" dependencies = [ + "rumqttc-v4-next", "rumqttc-v5-next", "serde", "serde_json", @@ -3107,6 +3108,29 @@ dependencies = [ "tokio-rustls", ] +[[package]] +name = "rumqttc-v4-next" +version = "0.34.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3767dde22a804ea806107d0f4331e33cb794d1b8468b11c86fb0c1366f07054" +dependencies = [ + "async-tungstenite", + "bytes", + "fixedbitset", + "flume", + "futures-util", + "log", + "mqttbytes-core-next", + "rumqttc-core-next", + "rustls-native-certs", + "rustls-pki-types", + "rustls-webpki", + "thiserror 2.0.20", + "tokio", + "tokio-rustls", + "tokio-util", +] + [[package]] name = "rumqttc-v5-next" version = "0.34.0" diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 49a0555..afd2890 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -27,7 +27,9 @@ tauri-plugin-process = "2" tauri-plugin-updater = "2" serde = { version = "1", features = ["derive"] } serde_json = "1" -rumqttc-v5-next = "0.34" +# rumqttc-v5-next 与 rumqttc-v4-next 的 lib 目标名都是 "rumqttc"，必须重命名依赖才能共存 +rumqttc = { package = "rumqttc-v5-next", version = "0.34" } +rumqttc-v4 = { package = "rumqttc-v4-next", version = "0.34" } tokio = { version = "1", features = ["sync"] } uuid = { version = "1", features = ["v4"] } diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 3705f5c..240ccd7 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -5,6 +5,7 @@ use rumqttc::{ AsyncClient, Broker, ConnectionError, Event, EventLoop, MqttOptions, Outgoing, PublishOptions, }; +use rumqttc_v4 as rv4; use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::path::PathBuf; @@ -13,6 +14,29 @@ use std::sync::Arc; use tauri::{AppHandle, Emitter, Manager}; use tokio::sync::{RwLock, watch}; +// ─── 调试日志 ──────────────────────────────────────────────────────────────── + +fn write_debug_log(msg: &str) { + use std::io::Write; + if let Ok(mut f) = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open("mqttx_debug.log") + { + let _ = writeln!(f, "[{}] {}", chrono_now(), msg); + } +} + +fn chrono_now() -> String { + use std::time::{SystemTime, UNIX_EPOCH}; + let secs = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs(); + // 简单 UTC 时间 + let h = (secs / 3600) % 24; + let m = (secs / 60) % 60; + let s = secs % 60; + format!("{:02}:{:02}:{:02}", h, m, s) +} + // ─── DTO structs ─────────────────────────────────────────────────────────────── #[derive(Debug, Clone, Serialize, Deserialize)] @@ -24,6 +48,9 @@ pub struct MqttConnectionDto { pub username: String, pub password: String, pub client_id: String, + /// MQTT 协议版本: "3.1.1" | "5.0"，老数据缺省按 5.0 处理 + #[serde(default = "default_protocol_version")] + pub protocol_version: String, pub clean_start: bool, pub session_expiry_interval: u32, pub keep_alive: u64, @@ -33,6 +60,10 @@ pub struct MqttConnectionDto { pub topic_alias_maximum: Option, } +fn default_protocol_version() -> String { + "5.0".to_string() +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct PublishDto { pub connection_id: String, @@ -183,8 +214,13 @@ impl LogWriter { // ─── State management ────────────────────────────────────────────────────────── +enum ClientKind { + V5(AsyncClient), + V4(rv4::AsyncClient), +} + struct ClientHandle { - client: AsyncClient, + client: ClientKind, cancel_tx: watch::Sender, connected: Arc, } @@ -238,6 +274,22 @@ fn emit_log(app_handle: &AppHandle, connection_id: &str, level: &str, event: &st } } +fn make_status( + connection_id: &str, + status: &str, + error: Option, + session_present: Option, + reason_code: Option, +) -> MqttStatusEvent { + MqttStatusEvent { + connection_id: connection_id.to_string(), + status: status.to_string(), + error, + session_present, + reason_code, + } +} + fn extract_publish_info(publish: &Publish) -> MqttMessageEvent { let payload_str = String::from_utf8_lossy(&publish.payload).to_string(); let props = publish.properties.as_ref(); @@ -271,6 +323,26 @@ fn extract_publish_info(publish: &Publish) -> MqttMessageEvent { } } +/// MQTT 3.1.1 的 Publish 没有 v5 properties，v5 专属字段一律置空 +fn extract_publish_info_v4(publish: &rv4::Publish) -> MqttMessageEvent { + MqttMessageEvent { + connection_id: String::new(), // will be filled by caller + topic: String::from_utf8_lossy(&publish.topic).to_string(), + payload: String::from_utf8_lossy(&publish.payload).to_string(), + qos: qos_to_u8(publish.qos), + retain: publish.retain, + timestamp: now_timestamp(), + reason_code: None, + user_properties: Vec::new(), + content_type: None, + content_encoding: None, + response_topic: None, + correlation_data: None, + message_expiry_interval: None, + subscription_identifier: None, + } +} + fn describe_publish_properties(props: &Option) -> String { props.as_ref().map(|p| { let mut parts = Vec::new(); @@ -308,7 +380,7 @@ fn describe_publish_properties(props: &Option) -> String { // ─── Background event loop task ──────────────────────────────────────────────── -fn spawn_event_loop( +fn spawn_event_loop_v5( connection_id: String, mut eventloop: EventLoop, mut cancel_rx: watch::Receiver, @@ -361,13 +433,10 @@ fn spawn_event_loop( emit_log(&app_handle, &connection_id, "error", "connack", format!("连接被拒绝: {:?} (code: {})", connack.code, reason_code), None); } - let status_event = MqttStatusEvent { - connection_id: connection_id.clone(), - status: "connected".to_string(), - error: None, - session_present: Some(connack.session_present), - reason_code: Some(reason_code), - }; + let status_event = make_status( + &connection_id, "connected", None, + Some(connack.session_present), Some(reason_code), + ); let _ = app_handle.emit("mqtt:status", &status_event); } @@ -389,13 +458,11 @@ fn spawn_event_loop( format!("收到断开连接: {reason_desc} (code: {reason_code})"), Some(props_info).filter(|d| !d.is_empty()), ); - let status_event = MqttStatusEvent { - connection_id: connection_id.clone(), - status: "disconnected".to_string(), - error: Some(format!("Disconnect reason: {reason_desc}")), - session_present: None, - reason_code: Some(reason_code), - }; + let status_event = make_status( + &connection_id, "disconnected", + Some(format!("Disconnect reason: {reason_desc}")), + None, Some(reason_code), + ); let _ = app_handle.emit("mqtt:status", &status_event); break; } @@ -545,13 +612,7 @@ fn spawn_event_loop( connected.store(false, Ordering::SeqCst); emit_log(&app_handle, &connection_id, "info", "disconnect_sent", "已发送断开连接请求".to_string(), None); - let status_event = MqttStatusEvent { - connection_id: connection_id.clone(), - status: "disconnected".to_string(), - error: None, - session_present: None, - reason_code: None, - }; + let status_event = make_status(&connection_id, "disconnected", None, None, None); let _ = app_handle.emit("mqtt:status", &status_event); break; } @@ -560,13 +621,7 @@ fn spawn_event_loop( connected.store(false, Ordering::SeqCst); emit_log(&app_handle, &connection_id, "info", "requests_done", "所有请求已完成，连接关闭".to_string(), None); - let status_event = MqttStatusEvent { - connection_id: connection_id.clone(), - status: "disconnected".to_string(), - error: None, - session_present: None, - reason_code: None, - }; + let status_event = make_status(&connection_id, "disconnected", None, None, None); let _ = app_handle.emit("mqtt:status", &status_event); break; } @@ -575,13 +630,9 @@ fn spawn_event_loop( connected.store(false, Ordering::SeqCst); emit_log(&app_handle, &connection_id, "error", "connection_error", format!("连接错误: {e}"), None); - let status_event = MqttStatusEvent { - connection_id: connection_id.clone(), - status: "error".to_string(), - error: Some(e.to_string()), - session_present: None, - reason_code: None, - }; + let status_event = make_status( + &connection_id, "error", Some(e.to_string()), None, None, + ); let _ = app_handle.emit("mqtt:status", &status_event); break; } @@ -618,13 +669,210 @@ fn spawn_event_loop( connected.store(false, Ordering::SeqCst); emit_log(&app_handle, &connection_id, "info", "disconnect_cancelled", "用户取消连接".to_string(), None); - let status_event = MqttStatusEvent { - connection_id: connection_id.clone(), - status: "disconnected".to_string(), - error: None, - session_present: None, - reason_code: None, - }; + let status_event = make_status(&connection_id, "disconnected", None, None, None); + let _ = app_handle.emit("mqtt:status", &status_event); + break; + } + } + } + }); +} + +/// MQTT 3.1.1 事件循环：与 v5 平行，packet 类型/日志行为按 v4 协议对齐 +fn spawn_event_loop_v4( + connection_id: String, + mut eventloop: rv4::EventLoop, + mut cancel_rx: watch::Receiver, + connected: Arc, + app_handle: AppHandle, +) { + let conn_id = connection_id.clone(); + let ah = app_handle.clone(); + emit_log(&ah, &conn_id, "info", "event_loop_started", "MQTT 3.1.1 事件循环已启动".to_string(), None); + + tokio::spawn(async move { + loop { + tokio::select! { + event = eventloop.poll() => { + match event { + Ok(rv4::Event::Incoming(rv4::Packet::Publish(publish))) => { + let topic = String::from_utf8_lossy(&publish.topic).to_string(); + let payload_len = publish.payload.len(); + let payload_preview = String::from_utf8_lossy(&publish.payload).to_string().chars().take(200).collect::(); + emit_log(&app_handle, &connection_id, "info", "publish_received", + format!("收到消息: {topic} ({} 字节, QoS {})", payload_len, qos_to_u8(publish.qos)), + Some(format!("payload: {}{}", + payload_preview, + if payload_len > 200 { format!("... (共 {} 字节)", payload_len) } else { String::new() })), + ); + let mut msg = extract_publish_info_v4(&publish); + msg.connection_id = connection_id.clone(); + let _ = app_handle.emit("mqtt:message", &msg); + } + + Ok(rv4::Event::Incoming(rv4::Packet::ConnAck(connack))) => { + connected.store(true, Ordering::SeqCst); + let reason_code = connack.code as u8; + let is_success = connack.code == rv4::ConnectReturnCode::Success; + if is_success { + emit_log(&app_handle, &connection_id, "info", "connack", + format!("连接成功 (session_present: {})", connack.session_present), None); + } else { + emit_log(&app_handle, &connection_id, "error", "connack", + format!("连接被拒绝: {:?} (code: {})", connack.code, reason_code), None); + } + let status_event = make_status( + &connection_id, "connected", None, + Some(connack.session_present), Some(reason_code), + ); + let _ = app_handle.emit("mqtt:status", &status_event); + } + + Ok(rv4::Event::Incoming(rv4::Packet::Disconnect)) => { + connected.store(false, Ordering::SeqCst); + emit_log(&app_handle, &connection_id, "warn", "disconnect_received", + "收到断开连接 (MQTT 3.1.1)".to_string(), None); + let status_event = make_status( + &connection_id, "disconnected", + Some("收到服务端断开连接".to_string()), None, None, + ); + let _ = app_handle.emit("mqtt:status", &status_event); + break; + } + + Ok(rv4::Event::Incoming(rv4::Packet::SubAck(suback))) => { + let codes: Vec = suback.return_codes.iter().map(|rc| { + match rc { + rv4::SubscribeReasonCode::Success(qos) => format!("成功(QoS {})", qos_to_u8(*qos)), + rv4::SubscribeReasonCode::Failure => "失败(0x80)".to_string(), + } + }).collect(); + let any_failure = suback.return_codes.iter().any(|rc| !matches!(rc, rv4::SubscribeReasonCode::Success(_))); + let level = if any_failure { "warn" } else { "info" }; + emit_log(&app_handle, &connection_id, level, "suback", + format!("订阅确认 (packet_id: {}): {}", suback.pkid, codes.join(", ")), None); + } + + Ok(rv4::Event::Incoming(rv4::Packet::UnsubAck(unsuback))) => { + emit_log(&app_handle, &connection_id, "info", "unsuback", + format!("取消订阅确认 (packet_id: {})", unsuback.pkid), None); + } + + Ok(rv4::Event::Incoming(rv4::Packet::PubAck(puback))) => { + emit_log(&app_handle, &connection_id, "info", "puback", + format!("发布确认 (packet_id: {})", puback.pkid), None); + } + + Ok(rv4::Event::Incoming(rv4::Packet::PubRec(pubrec))) => { + emit_log(&app_handle, &connection_id, "info", "pubrec", + format!("QoS2 发布收到 (packet_id: {})", pubrec.pkid), None); + } + + Ok(rv4::Event::Incoming(rv4::Packet::PubRel(pubrel))) => { + emit_log(&app_handle, &connection_id, "info", "pubrel", + format!("QoS2 发布释放 (packet_id: {})", pubrel.pkid), None); + } + + Ok(rv4::Event::Incoming(rv4::Packet::PubComp(pubcomp))) => { + emit_log(&app_handle, &connection_id, "info", "pubcomp", + format!("QoS2 发布完成 (packet_id: {})", pubcomp.pkid), None); + } + + Ok(rv4::Event::Incoming(rv4::Packet::PingResp)) => { + // Ping responses are too noisy to log every time + } + + // ─── Outgoing events (packet IDs only) ────────────────────── + + Ok(rv4::Event::Outgoing(rv4::Outgoing::PingReq)) => { + // Ping requests are too noisy + } + + Ok(rv4::Event::Outgoing(rv4::Outgoing::Publish(pkid))) => { + emit_log(&app_handle, &connection_id, "info", "publish_sent", + format!("发送消息 (packet_id: {pkid})"), None); + } + + Ok(rv4::Event::Outgoing(rv4::Outgoing::Subscribe(pkid))) => { + emit_log(&app_handle, &connection_id, "info", "subscribe_sent", + format!("发送订阅请求 (packet_id: {pkid})"), None); + } + + Ok(rv4::Event::Outgoing(rv4::Outgoing::Unsubscribe(pkid))) => { + emit_log(&app_handle, &connection_id, "info", "unsubscribe_sent", + format!("发送取消订阅请求 (packet_id: {pkid})"), None); + } + + Ok(rv4::Event::Outgoing(rv4::Outgoing::PubAck(pkid))) => { + emit_log(&app_handle, &connection_id, "info", "puback_sent", + format!("发送发布确认 (packet_id: {pkid})"), None); + } + + Ok(rv4::Event::Outgoing(rv4::Outgoing::PubRec(pkid))) => { + emit_log(&app_handle, &connection_id, "info", "pubrec_sent", + format!("发送QoS2发布收到 (packet_id: {pkid})"), None); + } + + Ok(rv4::Event::Outgoing(rv4::Outgoing::PubRel(pkid))) => { + emit_log(&app_handle, &connection_id, "info", "pubrel_sent", + format!("发送QoS2发布释放 (packet_id: {pkid})"), None); + } + + Ok(rv4::Event::Outgoing(rv4::Outgoing::PubComp(pkid))) => { + emit_log(&app_handle, &connection_id, "info", "pubcomp_sent", + format!("发送QoS2发布完成 (packet_id: {pkid})"), None); + } + + Ok(rv4::Event::Outgoing(rv4::Outgoing::PingResp)) => { + // Ping responses are too noisy + } + + Ok(rv4::Event::Outgoing(rv4::Outgoing::AwaitAck(pkid))) => { + emit_log(&app_handle, &connection_id, "info", "await_ack", + format!("等待应答 (packet_id: {pkid})"), None); + } + + Ok(rv4::Event::Outgoing(rv4::Outgoing::Disconnect)) => { + connected.store(false, Ordering::SeqCst); + emit_log(&app_handle, &connection_id, "info", "disconnect_sent", + "已发送断开连接请求".to_string(), None); + let status_event = make_status(&connection_id, "disconnected", None, None, None); + let _ = app_handle.emit("mqtt:status", &status_event); + break; + } + + Err(rv4::ConnectionError::RequestsDone) => { + connected.store(false, Ordering::SeqCst); + emit_log(&app_handle, &connection_id, "info", "requests_done", + "所有请求已完成，连接关闭".to_string(), None); + let status_event = make_status(&connection_id, "disconnected", None, None, None); + let _ = app_handle.emit("mqtt:status", &status_event); + break; + } + + Err(e) => { + connected.store(false, Ordering::SeqCst); + emit_log(&app_handle, &connection_id, "error", "connection_error", + format!("连接错误: {e}"), None); + let status_event = make_status( + &connection_id, "error", Some(e.to_string()), None, None, + ); + let _ = app_handle.emit("mqtt:status", &status_event); + break; + } + + Ok(rv4::Event::Incoming(other)) => { + emit_log(&app_handle, &connection_id, "info", "incoming_other", + format!("未处理入站事件: {:?}", std::mem::discriminant(&other)), None); + } + } + } + + _ = cancel_rx.changed() => { + connected.store(false, Ordering::SeqCst); + emit_log(&app_handle, &connection_id, "info", "disconnect_cancelled", + "用户取消连接".to_string(), None); + let status_event = make_status(&connection_id, "disconnected", None, None, None); let _ = app_handle.emit("mqtt:status", &status_event); break; } @@ -641,28 +889,15 @@ async fn mqtt_connect( state: tauri::State<'_, AppState>, conn: MqttConnectionDto, ) -> Result<(), String> { - let broker = Broker::tcp(&conn.host, conn.port); - let mut mqttopts = MqttOptions::new(&conn.client_id, broker); - mqttopts.set_keep_alive(conn.keep_alive as u16); - mqttopts.set_clean_start(conn.clean_start); - - if conn.session_expiry_interval > 0 { - mqttopts.set_session_expiry_interval(Some(conn.session_expiry_interval)); - } - - if !conn.username.is_empty() { - mqttopts.set_credentials(conn.username.clone(), conn.password.clone()); - } - - if let Some(rm) = conn.receive_maximum { - mqttopts.set_receive_maximum(Some(rm)); - } - if let Some(mps) = conn.maximum_packet_size { - mqttopts.set_max_packet_size(Some(mps)); - } - if let Some(tam) = conn.topic_alias_maximum { - mqttopts.set_topic_alias_max(Some(tam)); - } + // 调试日志：打印连接参数（密码脱敏，避免明文落盘） + let debug_info = format!( + "MQTT连接参数:\n host: {}\n port: {}\n client_id: {}\n username: {}\n password: ***\n protocol: {}\n clean_start: {}", + conn.host, conn.port, conn.client_id, conn.username, + conn.protocol_version, conn.clean_start + ); + write_debug_log(&debug_info); + emit_log(&app_handle, &conn.id, "debug", "connect_params", debug_info.clone(), None); + eprintln!("[MQTT DEBUG] {}", debug_info); // Cancel existing connection if any { @@ -672,36 +907,74 @@ async fn mqtt_connect( } } - let (client, eventloop) = AsyncClient::builder(mqttopts).build(); let connected = Arc::new(AtomicBool::new(false)); let (cancel_tx, cancel_rx) = watch::channel(false); - let handle = ClientHandle { - client: client.clone(), - cancel_tx, - connected: connected.clone(), + let is_v4 = conn.protocol_version == "3.1.1"; + + let client_kind = if is_v4 { + let broker = rv4::Broker::tcp(&conn.host, conn.port); + let mut mqttopts = rv4::MqttOptions::new(&conn.client_id, broker); + mqttopts.set_keep_alive(conn.keep_alive as u16); + mqttopts.set_clean_session(conn.clean_start); + if !conn.username.is_empty() { + mqttopts.set_credentials(conn.username.clone(), conn.password.clone()); + } + + let (client, eventloop) = rv4::AsyncClient::builder(mqttopts).build(); + let handle = ClientHandle { + client: ClientKind::V4(client.clone()), + cancel_tx, + connected: connected.clone(), + }; + spawn_event_loop_v4(conn.id.clone(), eventloop, cancel_rx, connected.clone(), app_handle.clone()); + handle + } else { + let broker = Broker::tcp(&conn.host, conn.port); + let mut mqttopts = MqttOptions::new(&conn.client_id, broker); + mqttopts.set_keep_alive(conn.keep_alive as u16); + mqttopts.set_clean_start(conn.clean_start); + + if conn.session_expiry_interval > 0 { + mqttopts.set_session_expiry_interval(Some(conn.session_expiry_interval)); + } + + if !conn.username.is_empty() { + mqttopts.set_credentials(conn.username.clone(), conn.password.clone()); + } + + if let Some(rm) = conn.receive_maximum { + mqttopts.set_receive_maximum(Some(rm)); + } + if let Some(mps) = conn.maximum_packet_size { + mqttopts.set_max_packet_size(Some(mps)); + } + if let Some(tam) = conn.topic_alias_maximum { + mqttopts.set_topic_alias_max(Some(tam)); + } + + let (client, eventloop) = AsyncClient::builder(mqttopts).build(); + let handle = ClientHandle { + client: ClientKind::V5(client.clone()), + cancel_tx, + connected: connected.clone(), + }; + spawn_event_loop_v5(conn.id.clone(), eventloop, cancel_rx, connected.clone(), app_handle.clone()); + handle }; { let mut clients = state.clients.write().await; - clients.insert(conn.id.clone(), handle); + clients.insert(conn.id.clone(), client_kind); } // Emit connecting status - let status_event = MqttStatusEvent { - connection_id: conn.id.clone(), - status: "connecting".to_string(), - error: None, - session_present: None, - reason_code: None, - }; + let status_event = make_status(&conn.id, "connecting", None, None, None); let _ = app_handle.emit("mqtt:status", &status_event); emit_log(&app_handle, &conn.id, "info", "connect", - format!("正在连接 {0}:{1}", conn.host, conn.port), None); - - // Spawn background task - spawn_event_loop(conn.id, eventloop, cancel_rx, connected, app_handle); + format!("正在连接 {0}:{1} (MQTT {2})", conn.host, conn.port, conn.protocol_version), + Some(format!("protocol_version: {}", conn.protocol_version))); Ok(()) } @@ -710,45 +983,87 @@ async fn mqtt_connect( async fn mqtt_test_connection(conn: MqttConnectionDto) -> Result { use tokio::time::timeout; - let test_id = format!("{}_test_{}", conn.client_id, std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_millis()); + // 调试日志：打印测试连接参数（密码脱敏，避免明文落盘） + let debug_info = format!( + "MQTT测试连接参数:\n host: {}\n port: {}\n client_id: {}\n username: {}\n password: ***\n protocol: {}", + conn.host, conn.port, conn.client_id, conn.username, + conn.protocol_version + ); + write_debug_log(&debug_info); + eprintln!("[MQTT TEST DEBUG] {}", debug_info); - let broker = Broker::tcp(&conn.host, conn.port); - let mut mqttopts = MqttOptions::new(&test_id, broker); - mqttopts.set_keep_alive(10); - mqttopts.set_clean_start(true); + // 测试连接使用原始 client_id，不修改它 + // 因为 password 是基于 client_id 计算的签名 + let is_v4 = conn.protocol_version == "3.1.1"; - if !conn.username.is_empty() { - mqttopts.set_credentials(conn.username.clone(), conn.password.clone()); - } - - let (client, mut eventloop) = AsyncClient::builder(mqttopts).build(); - - let result = timeout(std::time::Duration::from_secs(5), async { - loop { - match eventloop.poll().await { - Ok(Event::Incoming(Packet::ConnAck(connack))) => { - if connack.code == ConnectReturnCode::Success { - return Ok("连接测试成功！".to_string()); - } else { - return Err(format!("连接被拒绝: {:?}", connack.code)); - } - } - Ok(Event::Incoming(Packet::Disconnect(disconnect))) => { - return Err(format!("连接被拒绝: {:?}", disconnect.reason_code)); - } - Err(e) => { - return Err(format!("连接失败: {}", e)); - } - _ => {} - } + let result = if is_v4 { + let broker = rv4::Broker::tcp(&conn.host, conn.port); + let mut mqttopts = rv4::MqttOptions::new(&conn.client_id, broker); + mqttopts.set_keep_alive(10); + mqttopts.set_clean_session(true); + if !conn.username.is_empty() { + mqttopts.set_credentials(conn.username.clone(), conn.password.clone()); } - }).await; - // Drop client to close the connection - drop(client); + let (_client, mut eventloop) = rv4::AsyncClient::builder(mqttopts).build(); + timeout(std::time::Duration::from_secs(5), async { + loop { + match eventloop.poll().await { + Ok(rv4::Event::Incoming(rv4::Packet::ConnAck(connack))) => { + if connack.code == rv4::ConnectReturnCode::Success { + write_debug_log("MQTT测试连接成功"); + return Ok("连接测试成功！".to_string()); + } else { + let err = format!("连接被拒绝: {:?}", connack.code); + write_debug_log(&err); + return Err(err); + } + } + Ok(rv4::Event::Incoming(rv4::Packet::Disconnect)) => { + let err = "连接被拒绝: 服务端主动断开".to_string(); + write_debug_log(&err); + return Err(err); + } + Err(e) => { + let err = format!("连接失败: {}", e); + write_debug_log(&err); + return Err(err); + } + _ => {} + } + } + }).await + } else { + let broker = Broker::tcp(&conn.host, conn.port); + let mut mqttopts = MqttOptions::new(&conn.client_id, broker); + mqttopts.set_keep_alive(10); + mqttopts.set_clean_start(true); + if !conn.username.is_empty() { + mqttopts.set_credentials(conn.username.clone(), conn.password.clone()); + } + + let (_client, mut eventloop) = AsyncClient::builder(mqttopts).build(); + timeout(std::time::Duration::from_secs(5), async { + loop { + match eventloop.poll().await { + Ok(Event::Incoming(Packet::ConnAck(connack))) => { + if connack.code == ConnectReturnCode::Success { + return Ok("连接测试成功！".to_string()); + } else { + return Err(format!("连接被拒绝: {:?}", connack.code)); + } + } + Ok(Event::Incoming(Packet::Disconnect(disconnect))) => { + return Err(format!("连接被拒绝: {:?}", disconnect.reason_code)); + } + Err(e) => { + return Err(format!("连接失败: {}", e)); + } + _ => {} + } + } + }).await + }; match result { Ok(r) => r, @@ -789,15 +1104,6 @@ async fn mqtt_publish( let qos = qos_from_u8(publish.qos); let payload = publish.payload.into_bytes(); - let mut props = PublishProperties::default(); - props.user_properties = publish.user_properties; - props.content_type = publish.content_type.clone(); - props.message_expiry_interval = publish.message_expiry_interval; - props.response_topic = publish.response_topic.clone(); - props.correlation_data = publish - .correlation_data - .map(|s| s.into_bytes().into()); - emit_log(&app_handle, &publish.connection_id, "info", "publish_sent", format!("发布消息到 {topic}: {size} 字节 (QoS {qos}, retain={retain})", topic = publish.topic, @@ -808,19 +1114,44 @@ async fn mqtt_publish( Some(format!("payload: {}", String::from_utf8_lossy(&payload).to_string().chars().take(200).collect::())), ); - let options = PublishOptions::new(qos) - .retain(publish.retain) - .properties(props); + match &handle.client { + ClientKind::V5(client) => { + let mut props = PublishProperties::default(); + props.user_properties = publish.user_properties; + props.content_type = publish.content_type.clone(); + props.message_expiry_interval = publish.message_expiry_interval; + props.response_topic = publish.response_topic.clone(); + props.correlation_data = publish + .correlation_data + .map(|s| s.into_bytes().into()); - handle - .client - .publish(publish.topic, payload, options) - .await - .map_err(|e| { - emit_log(&app_handle, &publish.connection_id, "error", "publish_error", - format!("发布失败: {e}"), None); - e.to_string() - })?; + let options = PublishOptions::new(qos) + .retain(publish.retain) + .properties(props); + + client + .publish(publish.topic, payload, options) + .await + .map_err(|e| { + emit_log(&app_handle, &publish.connection_id, "error", "publish_error", + format!("发布失败: {e}"), None); + e.to_string() + })?; + } + ClientKind::V4(client) => { + // MQTT 3.1.1 没有 v5 properties，直接忽略 + let options = rv4::PublishOptions::new(qos).retain(publish.retain); + + client + .publish(publish.topic, payload, options) + .await + .map_err(|e| { + emit_log(&app_handle, &publish.connection_id, "error", "publish_error", + format!("发布失败: {e}"), None); + e.to_string() + })?; + } + } Ok(()) } @@ -842,23 +1173,40 @@ async fn mqtt_subscribe( format!("订阅主题: {topic} (QoS {qos})", topic = subscribe.topic, qos = subscribe.qos), None); // Use tracked subscribe to wait for SubAck result - let notice = handle - .client - .subscribe_tracked(subscribe.topic, qos) - .await - .map_err(|e| { - emit_log(&app_handle, &subscribe.connection_id, "error", "subscribe_error", - format!("订阅请求失败: {e}"), None); - e.to_string() - })?; - - // Wait for SubAck and check return codes - notice.wait_completion_async().await.map_err(|e| { - let err_msg = format!("订阅被拒绝: {e}"); - emit_log(&app_handle, &subscribe.connection_id, "error", "subscribe_rejected", - err_msg.clone(), None); - err_msg - })?; + match &handle.client { + ClientKind::V5(client) => { + let notice = client + .subscribe_tracked(subscribe.topic, qos) + .await + .map_err(|e| { + emit_log(&app_handle, &subscribe.connection_id, "error", "subscribe_error", + format!("订阅请求失败: {e}"), None); + e.to_string() + })?; + notice.wait_completion_async().await.map_err(|e| { + let err_msg = format!("订阅被拒绝: {e}"); + emit_log(&app_handle, &subscribe.connection_id, "error", "subscribe_rejected", + err_msg.clone(), None); + err_msg + })?; + } + ClientKind::V4(client) => { + let notice = client + .subscribe_tracked(subscribe.topic, qos) + .await + .map_err(|e| { + emit_log(&app_handle, &subscribe.connection_id, "error", "subscribe_error", + format!("订阅请求失败: {e}"), None); + e.to_string() + })?; + notice.wait_completion_async().await.map_err(|e| { + let err_msg = format!("订阅被拒绝: {e}"); + emit_log(&app_handle, &subscribe.connection_id, "error", "subscribe_rejected", + err_msg.clone(), None); + err_msg + })?; + } + } Ok(()) } @@ -878,15 +1226,22 @@ async fn mqtt_unsubscribe( emit_log(&app_handle, &connection_id, "info", "unsubscribe", format!("取消订阅主题: {topic}"), None); - handle - .client - .unsubscribe(topic) - .await - .map_err(|e| { - emit_log(&app_handle, &connection_id, "error", "unsubscribe_error", - format!("取消订阅失败: {e}"), None); - e.to_string() - })?; + match &handle.client { + ClientKind::V5(client) => { + client.unsubscribe(topic).await.map_err(|e| { + emit_log(&app_handle, &connection_id, "error", "unsubscribe_error", + format!("取消订阅失败: {e}"), None); + e.to_string() + })?; + } + ClientKind::V4(client) => { + client.unsubscribe(topic).await.map_err(|e| { + emit_log(&app_handle, &connection_id, "error", "unsubscribe_error", + format!("取消订阅失败: {e}"), None); + e.to_string() + })?; + } + } Ok(()) } @@ -931,6 +1286,18 @@ fn mqtt_get_log_dir(app_handle: AppHandle) -> Result { Ok(dir.to_string_lossy().to_string()) } +#[tauri::command] +async fn mqtt_read_debug_log() -> Result { + let path = std::env::current_dir() + .map_err(|e| e.to_string())? + .join("mqttx_debug.log"); + if path.exists() { + std::fs::read_to_string(&path).map_err(|e| e.to_string()) + } else { + Ok("日志文件不存在".to_string()) + } +} + #[cfg_attr(mobile, tauri::mobile_entry_point)] pub fn run() { let log_writer = std::env::current_dir() @@ -960,7 +1327,8 @@ pub fn run() { mqtt_get_connection_status, mqtt_test_connection, mqtt_get_log_dir, + mqtt_read_debug_log, ]) .run(tauri::generate_context!()) .expect("error while running tauri application"); -} \ No newline at end of file +} diff --git a/src/components/connections/ConnectionForm.vue b/src/components/connections/ConnectionForm.vue index 1f7ace4..a5df9a0 100644 --- a/src/components/connections/ConnectionForm.vue +++ b/src/components/connections/ConnectionForm.vue @@ -1,11 +1,16 @@ \ No newline at end of file +.preset-tab:hover { background: var(--el-fill-color); } +.preset-tab.active { background: var(--el-color-primary-light-9); color: var(--el-color-primary); font-weight: 500; } +.preset-tab-menu { opacity: 0.5; font-size: 14px; } +.preset-tab-menu:hover { opacity: 1; } + diff --git a/src/components/connections/ConnectionList.vue b/src/components/connections/ConnectionList.vue index 266bc85..d4f1eb2 100644 --- a/src/components/connections/ConnectionList.vue +++ b/src/components/connections/ConnectionList.vue @@ -1,5 +1,5 @@ @@ -112,21 +204,78 @@ function statusType(status: string): string {

连接管理

- - + 新增连接 - +
+ 导入 + 导出 + + 新增连接 +
+
- + +
+
+ 全部 + {{ connections.length }} +
+
+ 未分组 + {{ groupCounts['__ungrouped__'] || 0 }} +
+
+ {{ g }} + {{ groupCounts[g] || 0 }} + + … + + +
+
+ + + + + + - -
- 暂无连接，点击上方按钮创建 +
+ {{ connections.length === 0 ? '暂无连接，点击上方按钮创建' : '当前分组下暂无连接' }}
\ No newline at end of file +.group-tab.active .badge { background: var(--el-color-primary); color: #fff; } +.group-menu { cursor: pointer; font-size: 14px; opacity: 0.5; } +.group-menu:hover { opacity: 1; } + +.empty-hint { text-align: center; padding: 40px; color: var(--el-text-color-placeholder); } +.status-indicator { display: inline-block; width: 6px; height: 6px; border-radius: 50%; margin-right: 4px; } +.status-indicator.connected { background: var(--el-color-success); } +.status-indicator.connecting { background: var(--el-color-warning); animation: pulse 1s infinite; } +.status-indicator.error { background: var(--el-color-danger); } +.status-indicator.disconnected { background: var(--el-text-color-disabled); } +@keyframes pulse { 0%, 100% { opacity: 1; } 50% { opacity: 0.4; } } + diff --git a/src/components/dashboard/SendPanel.vue b/src/components/dashboard/SendPanel.vue index 99c402b..6202358 100644 --- a/src/components/dashboard/SendPanel.vue +++ b/src/components/dashboard/SendPanel.vue @@ -60,6 +60,13 @@ const connectedConnections = computed(() => props.connections.filter((c) => c.status === 'connected'), ) +const selectedConnection = computed(() => + props.connections.find((c) => c.id === selectedConnectionId.value), +) + +/** MQTT 3.1.1 连接不支持 v5 属性 */ +const isV5Connection = computed(() => selectedConnection.value?.protocolVersion !== '3.1.1') + const activePreset = computed(() => props.presets.find((p) => p.id === activePresetId.value), ) @@ -440,6 +447,7 @@ function onJsonPayloadChange(value: JsonValue) { {{ presetVariables.length }} -
+
diff --git a/src/stores/useAliyunPresets.ts b/src/stores/useAliyunPresets.ts new file mode 100644 index 0000000..deb5a8a --- /dev/null +++ b/src/stores/useAliyunPresets.ts @@ -0,0 +1,125 @@ +import { ref, watch } from 'vue' +import type { AliyunPreset } from '../types/mqtt' +import { defaultAliyunPreset } from '../types/mqtt' + +const STORAGE_KEY = 'mqttx_aliyun_presets' + +function loadPresets(): AliyunPreset[] { + try { + const raw = localStorage.getItem(STORAGE_KEY) + if (!raw) return [] + const parsed = JSON.parse(raw) as AliyunPreset[] + return parsed.map((p) => ({ + ...defaultAliyunPreset(), + ...p, + id: p.id || crypto.randomUUID(), + group: p.group ?? '', + })) + } catch { + return [] + } +} + +function savePresets(presets: AliyunPreset[]) { + try { + localStorage.setItem(STORAGE_KEY, JSON.stringify(presets)) + } catch {} +} + +const presets = ref(loadPresets()) + +watch(presets, (val) => savePresets(val), { deep: true }) + +export function useAliyunPresets() { + function add(preset?: Partial): AliyunPreset { + const p: AliyunPreset = { ...defaultAliyunPreset(), ...preset } + presets.value.push(p) + return p + } + + function update(id: string, data: Partial) { + const idx = presets.value.findIndex((p) => p.id === id) + if (idx !== -1) { + presets.value[idx] = { ...presets.value[idx], ...data } + } + } + + function remove(id: string) { + const idx = presets.value.findIndex((p) => p.id === id) + if (idx !== -1) presets.value.splice(idx, 1) + } + + function getById(id: string): AliyunPreset | undefined { + return presets.value.find((p) => p.id === id) + } + + // ─── 分组 ──────────────────────────────────────────────────────────────── + + function getGroups(): string[] { + const set = new Set() + for (const p of presets.value) { + if (p.group) set.add(p.group) + } + return Array.from(set).sort() + } + + function renameGroup(oldName: string, newName: string) { + for (const p of presets.value) { + if (p.group === oldName) p.group = newName + } + } + + function deleteGroup(group: string) { + for (const p of presets.value) { + if (p.group === group) p.group = '' + } + } + + // ─── 导入 / 导出 ──────────────────────────────────────────────────────── + + function exportPresets(ids?: string[]): string { + const list = ids + ? presets.value.filter((p) => ids.includes(p.id)) + : presets.value + return JSON.stringify(list, null, 2) + } + + /** 按 实例ID+GroupID+设备ID+AK 去重，返回导入/跳过数量 */ + function importPresets(json: string): { imported: number; skipped: number } { + const parsed = JSON.parse(json) as Partial[] + const seen = new Set( + presets.value.map((p) => `${p.instanceId}|${p.groupId}|${p.deviceId}|${p.accessKeyId}`), + ) + let imported = 0 + let skipped = 0 + for (const item of parsed) { + const p: AliyunPreset = { + ...defaultAliyunPreset(), + ...item, + id: crypto.randomUUID(), + } + const key = `${p.instanceId}|${p.groupId}|${p.deviceId}|${p.accessKeyId}` + if (seen.has(key)) { + skipped++ + continue + } + seen.add(key) + presets.value.push(p) + imported++ + } + return { imported, skipped } + } + + return { + presets, + add, + update, + remove, + getById, + getGroups, + renameGroup, + deleteGroup, + exportPresets, + importPresets, + } +} diff --git a/src/stores/useConnections.ts b/src/stores/useConnections.ts index 196e34b..725f773 100644 --- a/src/stores/useConnections.ts +++ b/src/stores/useConnections.ts @@ -9,9 +9,10 @@ function loadConnections(): MqttConnection[] { const raw = localStorage.getItem(STORAGE_KEY) if (!raw) return [] const parsed = JSON.parse(raw) as MqttConnection[] - // Ensure each connection has runtime status fields return parsed.map((c) => ({ ...c, + protocolVersion: c.protocolVersion ?? '5.0', + group: c.group ?? '', status: 'disconnected' as const, lastError: '', })) @@ -22,17 +23,15 @@ function loadConnections(): MqttConnection[] { function saveConnections(connections: MqttConnection[]) { try { - // Strip runtime fields before persisting const toSave = connections.map(({ status: _, lastError: __, ...rest }) => rest) localStorage.setItem(STORAGE_KEY, JSON.stringify(toSave)) } catch { - // localStorage full or unavailable — silently ignore + // localStorage full or unavailable } } const connections = ref(loadConnections()) -// Auto-persist on changes watch( connections, (val) => { @@ -42,7 +41,7 @@ watch( ) export function useConnections() { - function add() { + function add(): MqttConnection { const conn = defaultConnection() connections.value.push(conn) return conn @@ -82,6 +81,79 @@ export function useConnections() { return connections.value.filter((c) => c.status === 'connected') } + // ─── 分组操作 ────────────────────────────────────────────────────────────── + + /** 获取所有分组名称（去重） */ + function getGroups(): string[] { + const set = new Set() + for (const c of connections.value) { + if (c.group) set.add(c.group) + } + return Array.from(set).sort() + } + + /** 按分组获取连接列表 */ + function getByGroup(group: string): MqttConnection[] { + return connections.value.filter((c) => c.group === group) + } + + /** 重命名分组 */ + function renameGroup(oldName: string, newName: string) { + for (const c of connections.value) { + if (c.group === oldName) { + c.group = newName + } + } + } + + /** 删除分组（将组内连接移到未分组） */ + function deleteGroup(group: string) { + for (const c of connections.value) { + if (c.group === group) { + c.group = '' + } + } + } + + // ─── 导入 / 导出 ────────────────────────────────────────────────────────── + + /** 导出为 JSON 字符串（去除运行时字段） */ + function exportConnections(ids?: string[]): string { + const list = ids + ? connections.value.filter((c) => ids.includes(c.id)) + : connections.value + const data = list.map(({ status: _, lastError: __, ...rest }) => rest) + return JSON.stringify(data, null, 2) + } + + /** 从 JSON 导入，按名称+地址+端口+ClientID 去重，返回导入/跳过数量 */ + function importConnections(json: string): { imported: number; skipped: number } { + const parsed = JSON.parse(json) as Partial[] + const seen = new Set( + connections.value.map((c) => `${c.name}|${c.host}|${c.port}|${c.clientId}`), + ) + let imported = 0 + let skipped = 0 + for (const item of parsed) { + const conn: MqttConnection = { + ...defaultConnection(), + ...item, + id: crypto.randomUUID(), // 始终生成新 ID 避免冲突 + status: 'disconnected', + lastError: '', + } + const key = `${conn.name}|${conn.host}|${conn.port}|${conn.clientId}` + if (seen.has(key)) { + skipped++ + continue + } + seen.add(key) + connections.value.push(conn) + imported++ + } + return { imported, skipped } + } + return { connections, add, @@ -90,5 +162,11 @@ export function useConnections() { getById, updateStatus, getConnected, + getGroups, + getByGroup, + renameGroup, + deleteGroup, + exportConnections, + importConnections, } -} \ No newline at end of file +} diff --git a/src/stores/useMqttBridge.ts b/src/stores/useMqttBridge.ts index 9e42842..5e02638 100644 --- a/src/stores/useMqttBridge.ts +++ b/src/stores/useMqttBridge.ts @@ -149,6 +149,10 @@ export async function mqttTestConnection(conn: MqttConnectionDto): Promise('mqtt_test_connection', { conn }) } +export async function mqttReadDebugLog(): Promise { + return await invoke('mqtt_read_debug_log') +} + export async function mqttGetLogDir(): Promise { return await invoke('mqtt_get_log_dir') } \ No newline at end of file diff --git a/src/types/mqtt.ts b/src/types/mqtt.ts index ce79800..3e304ff 100644 --- a/src/types/mqtt.ts +++ b/src/types/mqtt.ts @@ -7,6 +7,98 @@ export interface Subscription { qos: number } +// ─── Aliyun Preset ───────────────────────────────────────────────────────────── + +/** 阿里云鉴权模式 */ +export type AliyunAuthMode = 'token' | 'device_credential' + +export interface AliyunPreset { + /** 唯一 ID */ + id: string + /** 预设名称 */ + name: string + /** 分组 */ + group: string + /** 实例 ID，如 post-cn-4591dq2ra1i */ + instanceId: string + /** 鉴权模式 */ + authMode: AliyunAuthMode + /** Access Key ID（Token 模式必填） */ + accessKeyId: string + /** Access Key Secret（Token 模式必填） */ + accessKeySecret: string + /** Group ID，如 GID-prod */ + groupId: string + /** 设备 ID，用于拼接 ClientId */ + deviceId: string + /** 设备 Access Key ID（一机一密模式必填） */ + deviceAccessKeyId: string + /** 设备 Access Key Secret（一机一密模式必填） */ + deviceAccessKeySecret: string +} + +export function defaultAliyunPreset(): AliyunPreset { + return { + id: crypto.randomUUID(), + name: '', + group: '', + instanceId: '', + authMode: 'token', + accessKeyId: '', + accessKeySecret: '', + groupId: '', + deviceId: `device_${Math.random().toString(36).slice(2, 10)}`, + deviceAccessKeyId: '', + deviceAccessKeySecret: '', + } +} + +/** 计算阿里云 MQTT 签名: HMAC-SHA1(Secret, ClientId) -> Base64 (使用 Web Crypto API) */ +export async function calculateAliyunSignature(clientId: string, secret: string): Promise { + const encoder = new TextEncoder() + const key = await crypto.subtle.importKey( + 'raw', + encoder.encode(secret), + { name: 'HMAC', hash: 'SHA-1' }, + false, + ['sign'], + ) + const signature = await crypto.subtle.sign('HMAC', key, encoder.encode(clientId)) + return bytesToBase64(new Uint8Array(signature)) +} + +/** 根据阿里云预设生成连接参数 */ +export async function generateAliyunConnection(preset: AliyunPreset): Promise> { + const clientId = `${preset.groupId}@@@${preset.deviceId}` + const username = `Signature|${preset.accessKeyId}|${preset.instanceId}` + const password = await calculateAliyunSignature(clientId, preset.accessKeySecret) + + return { + host: `${preset.instanceId}.mqtt.aliyuncs.com`, + port: 1883, + username, + password, + clientId, + protocolVersion: '3.1.1', + cleanStart: true, + } +} + +function bytesToBase64(bytes: Uint8Array): string { + const chars = 'ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/' + let result = '' + for (let i = 0; i < bytes.length; i += 3) { + const b0 = bytes[i] + const b1 = i + 1 < bytes.length ? bytes[i + 1] : 0 + const b2 = i + 2 < bytes.length ? bytes[i + 2] : 0 + result += chars[(b0 >> 2) & 0x3f] + result += chars[((b0 << 4) | (b1 >> 4)) & 0x3f] + result += i + 1 < bytes.length ? chars[((b1 << 2) | (b2 >> 6)) & 0x3f] : '=' + result += i + 2 < bytes.length ? chars[b2 & 0x3f] : '=' + } + return result +} + // ─── Connection ──────────────────────────────────────────────────────────────── export interface MqttConnection { @@ -17,10 +109,14 @@ export interface MqttConnection { username: string password: string clientId: string + /** MQTT 协议版本：3.1.1 / 5.0 */ + protocolVersion: '3.1.1' | '5.0' cleanStart: boolean sessionExpiryInterval: number keepAlive: number ssl: boolean + /** 分组名称，空字符串表示未分组 */ + group: string receiveMaximum: number | null maximumPacketSize: number | null topicAliasMaximum: number | null @@ -38,10 +134,12 @@ export function defaultConnection(): MqttConnection { username: '', password: '', clientId: `mqttx_${Math.random().toString(36).slice(2, 10)}`, + protocolVersion: '5.0', cleanStart: true, sessionExpiryInterval: 0, keepAlive: 60, ssl: false, + group: '', receiveMaximum: null, maximumPacketSize: null, topicAliasMaximum: null, @@ -112,6 +210,7 @@ export interface MqttConnectionDto { username: string password: string client_id: string + protocol_version: string clean_start: boolean session_expiry_interval: number keep_alive: number @@ -130,6 +229,7 @@ export function toConnectionDto(conn: MqttConnection): MqttConnectionDto { username: conn.username, password: conn.password, client_id: conn.clientId, + protocol_version: conn.protocolVersion, clean_start: conn.cleanStart, session_expiry_interval: conn.sessionExpiryInterval, keep_alive: conn.keepAlive, @@ -197,4 +297,4 @@ export interface MqttLogEvent { event: string message: string details: string | null -} \ No newline at end of file +}