From ab586d121a5f8756c0872863ac92be853681c22f Mon Sep 17 00:00:00 2001 From: jinmo <49105556+Jinmoli@users.noreply.github.com> Date: Thu, 6 Aug 2026 23:15:17 +0800 Subject: [PATCH] fix(mqtt): persist topics and support advanced options --- .../connection/ConnectionDialog.vue | 43 ++++- .../src/components/mqtt/MqttAdminConsole.vue | 12 +- .../src/components/mqtt/MqttPublishDialog.vue | 2 +- apps/desktop/src/i18n/locales/en.ts | 7 +- apps/desktop/src/i18n/locales/es.ts | 7 +- apps/desktop/src/i18n/locales/it.ts | 7 +- apps/desktop/src/i18n/locales/ja.ts | 7 +- apps/desktop/src/i18n/locales/pt-BR.ts | 7 +- apps/desktop/src/i18n/locales/zh-CN.ts | 7 +- apps/desktop/src/i18n/locales/zh-TW.ts | 7 +- apps/desktop/src/lib/backend/mqtt-http.ts | 2 +- apps/desktop/src/lib/backend/mqtt-tauri.ts | 4 +- apps/desktop/src/types/mqtt.ts | 7 + crates/dbx-core/src/mqtt/client.rs | 158 ++++++++++++++---- crates/dbx-core/src/mqtt/service.rs | 4 +- crates/dbx-core/src/mqtt/types.rs | 27 ++- docs/mqtt-issues-fix-plan.md | 58 +++++++ packages/app-tests/mqttPayloadCodec.test.ts | 5 + src-tauri/src/commands/mqtt_cmd.rs | 27 ++- 19 files changed, 347 insertions(+), 51 deletions(-) create mode 100644 docs/mqtt-issues-fix-plan.md diff --git a/apps/desktop/src/components/connection/ConnectionDialog.vue b/apps/desktop/src/components/connection/ConnectionDialog.vue index dd7aca7a58..ffc6c1f798 100644 --- a/apps/desktop/src/components/connection/ConnectionDialog.vue +++ b/apps/desktop/src/components/connection/ConnectionDialog.vue @@ -710,14 +710,18 @@ const mqttClientId = ref(""); const mqttProtocolVersion = ref<"v3" | "v4" | "v5">("v5"); const mqttTransportMode = ref<"tcp" | "websocket">("tcp"); const mqttWsPath = ref("/mqtt"); -const mqttAuthKind = ref<"none" | "password">("none"); +const mqttAuthKind = ref<"none" | "password" | "certificate">("none"); const mqttUsername = ref(""); const mqttPassword = ref(""); +const mqttCaCertPath = ref(""); +const mqttClientCertPath = ref(""); +const mqttClientKeyPath = ref(""); const mqttTls = ref(false); const mqttTlsSkipVerify = ref(false); const mqttKeepAliveSecs = ref(60); const mqttConnectTimeoutSecs = ref(30); const mqttMaxPacketSizeBytes = ref(16 * 1024 * 1024); +const mqttSavedTopics = ref([]); const nacosPrimaryAddressPlaceholder = computed(() => { return "http://127.0.0.1:8848/nacos"; }); @@ -1231,15 +1235,29 @@ function resetMqttFields(config?: Partial) { mqttKeepAliveSecs.value = Math.max(1, config?.keepAliveSecs || 60); mqttConnectTimeoutSecs.value = Math.max(1, config?.connectTimeoutSecs || 30); mqttMaxPacketSizeBytes.value = Math.min(268435455, Math.max(1024, config?.maxPacketSizeBytes || 16 * 1024 * 1024)); + mqttSavedTopics.value = config?.savedTopics ?? []; const auth = config?.auth; if (auth && auth.kind === "password") { mqttAuthKind.value = "password"; mqttUsername.value = auth.username || ""; mqttPassword.value = auth.password || ""; + mqttCaCertPath.value = ""; + mqttClientCertPath.value = ""; + mqttClientKeyPath.value = ""; + } else if (auth && auth.kind === "certificate") { + mqttAuthKind.value = "certificate"; + mqttUsername.value = ""; + mqttPassword.value = ""; + mqttCaCertPath.value = auth.caCertPath || ""; + mqttClientCertPath.value = auth.clientCertPath || ""; + mqttClientKeyPath.value = auth.clientKeyPath || ""; } else { mqttAuthKind.value = "none"; mqttUsername.value = ""; mqttPassword.value = ""; + mqttCaCertPath.value = ""; + mqttClientCertPath.value = ""; + mqttClientKeyPath.value = ""; } } @@ -1252,7 +1270,12 @@ function hydrateMqttFields(value: unknown) { } function buildMqttExternalConfig(): MqttConnectionConfig { - const auth: MqttConnectionConfig["auth"] = mqttAuthKind.value === "password" ? { kind: "password", username: mqttUsername.value, password: mqttPassword.value } : { kind: "none" }; + const auth: MqttConnectionConfig["auth"] = + mqttAuthKind.value === "password" + ? { kind: "password", username: mqttUsername.value, password: mqttPassword.value } + : mqttAuthKind.value === "certificate" + ? { kind: "certificate", caCertPath: mqttCaCertPath.value || undefined, clientCertPath: mqttClientCertPath.value || undefined, clientKeyPath: mqttClientKeyPath.value || undefined } + : { kind: "none" }; return { host: mqttHost.value.trim(), @@ -1266,6 +1289,7 @@ function buildMqttExternalConfig(): MqttConnectionConfig { keepAliveSecs: Math.max(1, mqttKeepAliveSecs.value), connectTimeoutSecs: Math.max(1, mqttConnectTimeoutSecs.value), maxPacketSizeBytes: Math.min(268435455, Math.max(1024, mqttMaxPacketSizeBytes.value)), + savedTopics: mqttSavedTopics.value, wsPath: mqttTransportMode.value === "websocket" ? mqttWsPath.value || "/mqtt" : undefined, }; } @@ -6050,6 +6074,7 @@ function openExternalUrl(url: string) {
+
+
diff --git a/apps/desktop/src/components/mqtt/MqttAdminConsole.vue b/apps/desktop/src/components/mqtt/MqttAdminConsole.vue index a11e163e16..6845e777db 100644 --- a/apps/desktop/src/components/mqtt/MqttAdminConsole.vue +++ b/apps/desktop/src/components/mqtt/MqttAdminConsole.vue @@ -20,6 +20,7 @@ const brokerInfo = ref(null); const topicTree = ref(null); const messages = ref([]); const subscribedTopics = ref<[string, string][]>([]); +const noLocalSubscribe = ref(false); const selectedTopic = ref(props.initialTopic ?? ""); const loading = ref(true); const error = ref(null); @@ -52,9 +53,9 @@ async function refreshData() { } } -async function handleSubscribe(topic: string) { +async function handleSubscribe(topic: string, noLocal = noLocalSubscribe.value) { try { - await mqttSubscribe(props.connectionId, topic, "atmostonce"); + await mqttSubscribe(props.connectionId, topic, "atmostonce", noLocal); await refreshData(); } catch (e) { error.value = String(e); @@ -194,6 +195,7 @@ onUnmounted(() => { " > +
@@ -216,13 +218,13 @@ onUnmounted(() => {
-
+
{{ t("connection.mqttNoMessages") }}
diff --git a/apps/desktop/src/components/mqtt/MqttPublishDialog.vue b/apps/desktop/src/components/mqtt/MqttPublishDialog.vue index ddfcece43c..68d2373857 100644 --- a/apps/desktop/src/components/mqtt/MqttPublishDialog.vue +++ b/apps/desktop/src/components/mqtt/MqttPublishDialog.vue @@ -32,7 +32,7 @@ const qosHints = computed(() => [t("connection.mqttQosAtMostOnce"), t("connectio const payloadPlaceholder = computed(() => { switch (encoding.value) { case "json": - return t("connection.mqttPayloadPlaceholderJson"); + return t("connection.mqttPayloadPlaceholderJson", { example: '{"key": "value"}' }); case "base64": return t("connection.mqttPayloadPlaceholderBase64"); case "hex": diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts index ba4023a0ea..55c1322f5f 100644 --- a/apps/desktop/src/i18n/locales/en.ts +++ b/apps/desktop/src/i18n/locales/en.ts @@ -526,6 +526,10 @@ export default { mqttUsername: "Username", mqttUsernamePlaceholder: "Optional", mqttPassword: "Password", + mqttAuthCertificate: "Client certificate", + mqttCaCertPath: "CA certificate path", + mqttClientCertPath: "Client certificate path", + mqttClientKeyPath: "Client private key path", mqttPasswordPlaceholder: "Optional", mqttTls: "TLS", mqttTlsSkipVerify: "Skip certificate verification", @@ -542,6 +546,7 @@ export default { mqttMessages: "Messages", mqttPublish: "Publish Message", mqttSubscribe: "Subscribe", + mqttNoLocal: "No Local", mqttUnsubscribe: "Unsubscribe", mqttNoMessages: "No messages", mqttNoTopics: "No subscribed topics", @@ -571,7 +576,7 @@ export default { mqttPayloadEncoding: "Payload encoding", mqttPayload: "Payload", mqttPayloadPlaceholderPlaintext: "Enter message content...", - mqttPayloadPlaceholderJson: 'Enter JSON, for example {"key": "value"}', + mqttPayloadPlaceholderJson: "Enter JSON, for example {example}", mqttPayloadPlaceholderBase64: "Enter Base64-encoded data", mqttPayloadPlaceholderHex: "Enter hexadecimal data, for example 48 65 6C 6C 6F", mqttPayloadPlaceholderStructured: "Enter a JSON value; it will be encoded as {encoding}", diff --git a/apps/desktop/src/i18n/locales/es.ts b/apps/desktop/src/i18n/locales/es.ts index fafbd2cf64..8d494127ee 100644 --- a/apps/desktop/src/i18n/locales/es.ts +++ b/apps/desktop/src/i18n/locales/es.ts @@ -792,7 +792,7 @@ export default withEnglishFallback({ mqttPayloadEncoding: "Codificación del Payload", mqttPayload: "Payload", mqttPayloadPlaceholderPlaintext: "Introduce el contenido del mensaje...", - mqttPayloadPlaceholderJson: 'Introduce JSON, por ejemplo {"key": "value"}', + mqttPayloadPlaceholderJson: "Introduce JSON, por ejemplo {example}", mqttPayloadPlaceholderBase64: "Introduce datos codificados en Base64", mqttPayloadPlaceholderHex: "Introduce datos hexadecimales, por ejemplo 48 65 6C 6C 6F", mqttPayloadPlaceholderStructured: "Introduce un valor JSON; se codificará como {encoding}", @@ -812,6 +812,11 @@ export default withEnglishFallback({ zookeeperAuthMethod: "Método de autenticación", zookeeperAuthDigest: "Digest (predeterminado)", zookeeperAuthSaslDigest: "SASL DIGEST-MD5", + mqttAuthCertificate: "Certificado de cliente", + mqttCaCertPath: "Ruta del certificado CA", + mqttClientCertPath: "Ruta del certificado de cliente", + mqttClientKeyPath: "Ruta de la clave privada del cliente", + mqttNoLocal: "Prohibir reenvío local", }, editor: { duckdbDraining: "La consulta anterior de DuckDB aún se está deteniendo. Vuelva a intentarlo en breve.", diff --git a/apps/desktop/src/i18n/locales/it.ts b/apps/desktop/src/i18n/locales/it.ts index a677e20593..269a6a3acb 100644 --- a/apps/desktop/src/i18n/locales/it.ts +++ b/apps/desktop/src/i18n/locales/it.ts @@ -790,7 +790,7 @@ export default withEnglishFallback({ mqttPayloadEncoding: "Codifica del Payload", mqttPayload: "Payload", mqttPayloadPlaceholderPlaintext: "Inserisci il contenuto del messaggio...", - mqttPayloadPlaceholderJson: 'Inserisci JSON, ad esempio {"key": "value"}', + mqttPayloadPlaceholderJson: "Inserisci JSON, ad esempio {example}", mqttPayloadPlaceholderBase64: "Inserisci dati codificati in Base64", mqttPayloadPlaceholderHex: "Inserisci dati esadecimali, ad esempio 48 65 6C 6C 6F", mqttPayloadPlaceholderStructured: "Inserisci un valore JSON; verrà codificato come {encoding}", @@ -810,6 +810,11 @@ export default withEnglishFallback({ zookeeperAuthMethod: "Metodo di autenticazione", zookeeperAuthDigest: "Digest (predefinito)", zookeeperAuthSaslDigest: "SASL DIGEST-MD5", + mqttAuthCertificate: "Certificato client", + mqttCaCertPath: "Percorso del certificato CA", + mqttClientCertPath: "Percorso del certificato client", + mqttClientKeyPath: "Percorso della chiave privata client", + mqttNoLocal: "Disabilita inoltro locale", }, editor: { duckdbDraining: "La query DuckDB precedente è ancora in fase di arresto. Riprova a breve.", diff --git a/apps/desktop/src/i18n/locales/ja.ts b/apps/desktop/src/i18n/locales/ja.ts index 958ac2ab77..1d29622fef 100644 --- a/apps/desktop/src/i18n/locales/ja.ts +++ b/apps/desktop/src/i18n/locales/ja.ts @@ -790,7 +790,7 @@ export default withEnglishFallback({ mqttPayloadEncoding: "ペイロードのエンコード", mqttPayload: "ペイロード", mqttPayloadPlaceholderPlaintext: "メッセージ内容を入力...", - mqttPayloadPlaceholderJson: 'JSON を入力（例: {"key": "value"}）', + mqttPayloadPlaceholderJson: "JSON を入力（例: {example}）", mqttPayloadPlaceholderBase64: "Base64 エンコード済みデータを入力", mqttPayloadPlaceholderHex: "16 進数データを入力（例: 48 65 6C 6C 6F）", mqttPayloadPlaceholderStructured: "JSON 値を入力してください。{encoding} としてエンコードされます", @@ -831,6 +831,11 @@ export default withEnglishFallback({ zookeeperAuthMethod: "認証方式", zookeeperAuthDigest: "Digest（デフォルト）", zookeeperAuthSaslDigest: "SASL DIGEST-MD5", + mqttAuthCertificate: "クライアント証明書", + mqttCaCertPath: "CA証明書パス", + mqttClientCertPath: "クライアント証明書パス", + mqttClientKeyPath: "クライアント秘密鍵パス", + mqttNoLocal: "ローカル転送禁止", }, editor: { duckdbDraining: "前回の DuckDB クエリはまだ停止処理中です。しばらくしてから再試行してください。", diff --git a/apps/desktop/src/i18n/locales/pt-BR.ts b/apps/desktop/src/i18n/locales/pt-BR.ts index a70906665b..844b295c98 100644 --- a/apps/desktop/src/i18n/locales/pt-BR.ts +++ b/apps/desktop/src/i18n/locales/pt-BR.ts @@ -791,7 +791,7 @@ export default withEnglishFallback({ mqttPayloadEncoding: "Codificação do Payload", mqttPayload: "Payload", mqttPayloadPlaceholderPlaintext: "Digite o conteúdo da mensagem...", - mqttPayloadPlaceholderJson: 'Digite JSON, por exemplo {"key": "value"}', + mqttPayloadPlaceholderJson: "Digite JSON, por exemplo {example}", mqttPayloadPlaceholderBase64: "Digite dados codificados em Base64", mqttPayloadPlaceholderHex: "Digite dados hexadecimais, por exemplo 48 65 6C 6C 6F", mqttPayloadPlaceholderStructured: "Digite um valor JSON; ele será codificado como {encoding}", @@ -811,6 +811,11 @@ export default withEnglishFallback({ zookeeperAuthMethod: "Método de autenticação", zookeeperAuthDigest: "Digest (padrão)", zookeeperAuthSaslDigest: "SASL DIGEST-MD5", + mqttAuthCertificate: "Certificado do cliente", + mqttCaCertPath: "Caminho do certificado CA", + mqttClientCertPath: "Caminho do certificado do cliente", + mqttClientKeyPath: "Caminho da chave privada do cliente", + mqttNoLocal: "Proibir encaminhamento local", }, editor: { duckdbDraining: "A consulta anterior do DuckDB ainda está sendo interrompida. Tente novamente em breve.", diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts index 4fa656ea56..05802da1e8 100644 --- a/apps/desktop/src/i18n/locales/zh-CN.ts +++ b/apps/desktop/src/i18n/locales/zh-CN.ts @@ -527,6 +527,10 @@ export default withEnglishFallback({ mqttUsername: "用户名", mqttUsernamePlaceholder: "选填", mqttPassword: "密码", + mqttAuthCertificate: "客户端证书", + mqttCaCertPath: "CA 证书路径", + mqttClientCertPath: "客户端证书路径", + mqttClientKeyPath: "客户端私钥路径", mqttPasswordPlaceholder: "选填", mqttTls: "TLS", mqttTlsSkipVerify: "跳过证书校验", @@ -543,6 +547,7 @@ export default withEnglishFallback({ mqttMessages: "消息", mqttPublish: "发布消息", mqttSubscribe: "订阅", + mqttNoLocal: "禁止本地转发", mqttUnsubscribe: "取消订阅", mqttNoMessages: "暂无消息", mqttNoTopics: "暂无已订阅的 Topic", @@ -572,7 +577,7 @@ export default withEnglishFallback({ mqttPayloadEncoding: "Payload 编码", mqttPayload: "消息内容 (Payload)", mqttPayloadPlaceholderPlaintext: "输入消息内容...", - mqttPayloadPlaceholderJson: '输入 JSON，例如 {"key": "value"}', + mqttPayloadPlaceholderJson: "输入 JSON，例如 {example}", mqttPayloadPlaceholderBase64: "输入 Base64 编码的数据", mqttPayloadPlaceholderHex: "输入十六进制数据，例如 48 65 6C 6C 6F", mqttPayloadPlaceholderStructured: "输入 JSON 值，发送时将自动编码为 {encoding}", diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts index 8aec5ccb2a..912a7199b0 100644 --- a/apps/desktop/src/i18n/locales/zh-TW.ts +++ b/apps/desktop/src/i18n/locales/zh-TW.ts @@ -790,7 +790,7 @@ export default withEnglishFallback({ mqttPayloadEncoding: "Payload 編碼", mqttPayload: "訊息內容 (Payload)", mqttPayloadPlaceholderPlaintext: "輸入訊息內容...", - mqttPayloadPlaceholderJson: '輸入 JSON，例如 {"key": "value"}', + mqttPayloadPlaceholderJson: "輸入 JSON，例如 {example}", mqttPayloadPlaceholderBase64: "輸入 Base64 編碼的資料", mqttPayloadPlaceholderHex: "輸入十六進位資料，例如 48 65 6C 6C 6F", mqttPayloadPlaceholderStructured: "輸入 JSON 值，傳送時將自動編碼為 {encoding}", @@ -810,6 +810,11 @@ export default withEnglishFallback({ zookeeperAuthMethod: "認證方式", zookeeperAuthDigest: "Digest（預設）", zookeeperAuthSaslDigest: "SASL DIGEST-MD5", + mqttAuthCertificate: "用戶端憑證", + mqttCaCertPath: "CA 憑證路徑", + mqttClientCertPath: "用戶端憑證路徑", + mqttClientKeyPath: "用戶端私鑰路徑", + mqttNoLocal: "禁止本機轉送", }, editor: { duckdbDraining: "上一筆 DuckDB 查詢仍在停止，請稍後重試。", diff --git a/apps/desktop/src/lib/backend/mqtt-http.ts b/apps/desktop/src/lib/backend/mqtt-http.ts index d678f75b41..f8c0e97b16 100644 --- a/apps/desktop/src/lib/backend/mqtt-http.ts +++ b/apps/desktop/src/lib/backend/mqtt-http.ts @@ -9,7 +9,7 @@ export async function mqttGetBrokerInfo(_connectionId: string) { mqttWebNotAvailable("mqttGetBrokerInfo"); } -export async function mqttSubscribe(_connectionId: string, _topic: string, _qos?: string | null) { +export async function mqttSubscribe(_connectionId: string, _topic: string, _qos?: string | null, _noLocal?: boolean) { mqttWebNotAvailable("mqttSubscribe"); } diff --git a/apps/desktop/src/lib/backend/mqtt-tauri.ts b/apps/desktop/src/lib/backend/mqtt-tauri.ts index 5dba9ab037..93d6ff8f50 100644 --- a/apps/desktop/src/lib/backend/mqtt-tauri.ts +++ b/apps/desktop/src/lib/backend/mqtt-tauri.ts @@ -4,8 +4,8 @@ export async function mqttGetBrokerInfo(connectionId: string) { return invoke("mqtt_get_broker_info", { connectionId }); } -export async function mqttSubscribe(connectionId: string, topic: string, qos?: string | null) { - return invoke("mqtt_subscribe", { connectionId, topic, qos: qos ?? null }); +export async function mqttSubscribe(connectionId: string, topic: string, qos?: string | null, noLocal?: boolean) { + return invoke("mqtt_subscribe", { connectionId, topic, qos: qos ?? null, noLocal: noLocal ?? false }); } export async function mqttUnsubscribe(connectionId: string, topic: string) { diff --git a/apps/desktop/src/types/mqtt.ts b/apps/desktop/src/types/mqtt.ts index 56d0283bfd..3aad08fbd1 100644 --- a/apps/desktop/src/types/mqtt.ts +++ b/apps/desktop/src/types/mqtt.ts @@ -26,6 +26,13 @@ export interface MqttConnectionConfig { connectTimeoutSecs: number; maxPacketSizeBytes: number; wsPath?: string; + savedTopics?: MqttSavedTopic[]; +} + +export interface MqttSavedTopic { + topic: string; + qos: MqttQoS; + noLocal?: boolean; } export interface MqttBrokerInfo { diff --git a/crates/dbx-core/src/mqtt/client.rs b/crates/dbx-core/src/mqtt/client.rs index af7e9e79ae..e1af2cc166 100644 --- a/crates/dbx-core/src/mqtt/client.rs +++ b/crates/dbx-core/src/mqtt/client.rs @@ -1,6 +1,8 @@ //! MQTT 客户端封装（基于 rumqttc），管理 MQTT broker 的长连接和消息收发。 use std::collections::{HashMap, HashSet, VecDeque}; +use std::fs::File; +use std::io::BufReader; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Weak}; use std::time::{Duration, SystemTime, UNIX_EPOCH}; @@ -12,8 +14,8 @@ use rumqttc::tokio_rustls::rustls::{ pki_types::{CertificateDer, ServerName, UnixTime}, }; use rumqttc::v5::{ - mqttbytes::QoS as QoSV5, AsyncClient as AsyncClientV5, Event as EventV5, Incoming as IncomingV5, - MqttOptions as MqttOptionsV5, + mqttbytes::{v5::Filter as MqttV5Filter, QoS as QoSV5}, + AsyncClient as AsyncClientV5, Event as EventV5, Incoming as IncomingV5, MqttOptions as MqttOptionsV5, }; use rumqttc::{ mqttbytes::Protocol, valid_filter, valid_topic, AsyncClient as AsyncClientV4, Event as EventV4, @@ -25,7 +27,7 @@ use tokio::task::JoinHandle; use super::types::{ MqttAuth, MqttBrokerInfo, MqttConnectionConfig, MqttMessage, MqttMessageDirection, MqttPublishRequest, MqttQoS, - MqttTlsVerificationMode, MqttTopicNode, MqttTransport, + MqttSavedTopic, MqttTlsVerificationMode, MqttTopicNode, MqttTransport, }; #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -56,6 +58,7 @@ struct PendingSubscribe { sequence: u64, topic: String, qos: MqttQoS, + no_local: bool, add_to_desired: bool, completion: Option, } @@ -308,6 +311,7 @@ pub struct MqttClient { granted_subscriptions: RwLock>, /// 需要在无会话重连后恢复的 topic 集合 desired_subscriptions: RwLock>, + no_local_topics: RwLock>, /// 当前连接是否已收到有效 CONNACK connected: AtomicBool, /// 订阅请求与 packet id 的关联状态 @@ -337,7 +341,7 @@ impl MqttClient { if !(1024..=268_435_455).contains(&config.max_packet_size_bytes) { return Err("MQTT 最大报文大小必须在 1024 到 268435455 字节之间".to_string()); } - let plan = build_connect_plan(&config); + let plan = build_connect_plan(&config)?; match plan.backend { MqttBackendKind::V4 => Self::connect_v4(config, plan).await, MqttBackendKind::V5 => Self::connect_v5(config, plan).await, @@ -388,12 +392,16 @@ impl MqttClient { backend: MqttBackendClient, shutdown_notify: Arc, ) -> Arc { + let desired_subscriptions = config.saved_topics.iter().map(|saved| (saved.topic.clone(), saved.qos)).collect(); + let no_local_topics = + config.saved_topics.iter().filter(|saved| saved.no_local).map(|saved| saved.topic.clone()).collect(); Arc::new(Self { backend, config, subscriptions: RwLock::new(Vec::new()), granted_subscriptions: RwLock::new(Vec::new()), - desired_subscriptions: RwLock::new(Vec::new()), + desired_subscriptions: RwLock::new(desired_subscriptions), + no_local_topics: RwLock::new(no_local_topics), connected: AtomicBool::new(false), subscription_requests: Mutex::new(SubscriptionRequestTracker::default()), subscription_send_lock: Mutex::new(()), @@ -775,7 +783,8 @@ impl MqttClient { if has_pending { continue; } - if let Err(error) = self.queue_subscribe_request(topic.clone(), qos, false, None).await { + let no_local = self.no_local_topics.read().await.contains(&topic); + if let Err(error) = self.queue_subscribe_request(topic.clone(), qos, no_local, false, None).await { log::warn!("MQTT 重连后恢复订阅 {topic} 失败: {error}"); } } @@ -833,6 +842,11 @@ impl MqttClient { upsert_subscription(&self.granted_subscriptions, &pending.topic, granted_qos).await; self.seen_retained.write().await.clear_filter(&pending.topic); if pending.add_to_desired { + if pending.no_local { + self.no_local_topics.write().await.insert(pending.topic.clone()); + } else { + self.no_local_topics.write().await.remove(&pending.topic); + } upsert_subscription(&self.desired_subscriptions, &pending.topic, pending.qos).await; } Ok(()) @@ -859,6 +873,7 @@ impl MqttClient { remove_subscription(&self.subscriptions, &pending.topic).await; remove_subscription(&self.granted_subscriptions, &pending.topic).await; remove_subscription(&self.desired_subscriptions, &pending.topic).await; + self.no_local_topics.write().await.remove(&pending.topic); self.seen_retained.write().await.clear_filter(&pending.topic); } else if let Err(error) = &result { log::warn!("{error}"); @@ -873,6 +888,7 @@ impl MqttClient { &self, topic: String, qos: MqttQoS, + no_local: bool, add_to_desired: bool, completion: Option, ) -> Result<(), String> { @@ -887,13 +903,14 @@ impl MqttClient { sequence, topic: topic.clone(), qos, + no_local, add_to_desired, completion, }); sequence }; - if let Err(error) = self.backend.subscribe(&topic, mqtt_qos(qos)).await { + if let Err(error) = self.backend.subscribe(&topic, mqtt_qos(qos), no_local).await { self.subscription_requests.lock().await.queued_subscribes.retain(|pending| pending.sequence != sequence); return Err(error); } @@ -941,15 +958,16 @@ impl MqttClient { } /// 订阅 topic - pub async fn subscribe(&self, topic: &str, qos: MqttQoS) -> Result<(), String> { + pub async fn subscribe(&self, topic: &str, qos: MqttQoS, no_local: bool) -> Result<(), String> { validate_topic_filter(topic)?; - if self + let has_desired = self .desired_subscriptions .read() .await .iter() - .any(|(current, current_qos)| current == topic && *current_qos == qos) - { + .any(|(current, current_qos)| current == topic && *current_qos == qos); + let no_local_matches = self.no_local_topics.read().await.contains(topic); + if has_desired && no_local_matches == no_local { return Ok(()); } if self.subscription_requests.lock().await.has_pending_topic(topic) { @@ -957,7 +975,7 @@ impl MqttClient { } let (completion, result) = oneshot::channel(); - self.queue_subscribe_request(topic.to_string(), qos, true, Some(completion)).await?; + self.queue_subscribe_request(topic.to_string(), qos, no_local, true, Some(completion)).await?; result.await.map_err(|_| "订阅确认通道已关闭，请检查 MQTT 连接状态".to_string())? } @@ -1031,6 +1049,19 @@ impl MqttClient { self.subscriptions.read().await.clone() } + /// 获取持久化意图中的订阅 Topic，用于保存连接配置。 + pub async fn desired_topics(&self) -> Vec<(String, MqttQoS)> { + self.desired_subscriptions.read().await.clone() + } + + pub async fn desired_topic_configs(&self) -> Vec { + let desired = self.desired_subscriptions.read().await.clone(); + let no_local = self.no_local_topics.read().await; + desired + .into_iter() + .map(|(topic, qos)| MqttSavedTopic { no_local: no_local.contains(&topic), topic, qos }) + .collect() + } /// 获取消息缓冲区中的消息 pub async fn get_messages(&self, topic_filter: Option<&str>, limit: usize) -> Vec { let buffer = self.message_buffer.read().await; @@ -1117,13 +1148,18 @@ impl Drop for MqttClient { } impl MqttBackendClient { - async fn subscribe(&self, topic: &str, qos: QoSV4) -> Result<(), String> { + async fn subscribe(&self, topic: &str, qos: QoSV4, no_local: bool) -> Result<(), String> { match self { MqttBackendClient::V4(client) => { + if no_local { + return Err("MQTT 3.x 不支持 No Local 订阅选项".to_string()); + } client.subscribe(topic, qos).await.map_err(|e| format!("订阅 topic 失败: {e}")) } MqttBackendClient::V5(client) => { - client.subscribe(topic, qos_v4_to_v5(qos)).await.map_err(|e| format!("订阅 topic 失败: {e}")) + let mut filter = MqttV5Filter::new(topic.to_string(), qos_v4_to_v5(qos)); + filter.nolocal = no_local; + client.subscribe_many([filter]).await.map_err(|e| format!("订阅 topic 失败: {e}")) } } } @@ -1159,7 +1195,7 @@ impl MqttBackendClient { } } -fn build_connect_plan(config: &MqttConnectionConfig) -> MqttConnectPlan { +fn build_connect_plan(config: &MqttConnectionConfig) -> Result { let backend = match config.protocol_version { super::types::MqttProtocolVersion::V3 | super::types::MqttProtocolVersion::V4 => MqttBackendKind::V4, super::types::MqttProtocolVersion::V5 => MqttBackendKind::V5, @@ -1170,13 +1206,27 @@ fn build_connect_plan(config: &MqttConnectionConfig) -> MqttConnectPlan { super::types::MqttProtocolVersion::V5 => None, }; let tls_verification_mode = config.tls_verification_mode(); - let transport = build_transport(config.transport, tls_verification_mode); + let transport = build_transport(config.transport, tls_verification_mode, &config.auth)?; - MqttConnectPlan { backend, broker_addr: config.broker_addr_for_transport(), transport, protocol } + Ok(MqttConnectPlan { backend, broker_addr: config.broker_addr_for_transport(), transport, protocol }) } -fn build_transport(transport: MqttTransport, tls_verification_mode: Option) -> Transport { - match (transport, tls_verification_mode) { +fn build_transport( + transport: MqttTransport, + tls_verification_mode: Option, + auth: &MqttAuth, +) -> Result { + if let Some(mode) = tls_verification_mode { + if matches!(auth, MqttAuth::Certificate { .. }) { + let tls = certificate_tls_configuration(auth, mode)?; + return Ok(match transport { + MqttTransport::Tcp => Transport::tls_with_config(tls), + MqttTransport::WebSocket => Transport::wss_with_config(tls), + }); + } + } + + Ok(match (transport, tls_verification_mode) { (MqttTransport::Tcp, None) => Transport::Tcp, (MqttTransport::Tcp, Some(MqttTlsVerificationMode::VerifyServerCert)) => Transport::tls_with_default_config(), (MqttTransport::Tcp, Some(MqttTlsVerificationMode::SkipServerCertVerification)) => { @@ -1189,9 +1239,57 @@ fn build_transport(transport: MqttTransport, tls_verification_mode: Option { Transport::wss_with_config(insecure_tls_configuration()) } - } + }) } +fn certificate_tls_configuration(auth: &MqttAuth, mode: MqttTlsVerificationMode) -> Result { + let MqttAuth::Certificate { ca_cert_path, client_cert_path, client_key_path } = auth else { + return Err("MQTT 证书认证配置无效".to_string()); + }; + let client_cert_path = client_cert_path.as_deref().ok_or("MQTT 证书认证缺少客户端证书路径")?; + let client_key_path = client_key_path.as_deref().ok_or("MQTT 证书认证缺少客户端私钥路径")?; + + let mut root_cert_store = rustls::RootCertStore::empty(); + if let Some(ca_cert_path) = ca_cert_path.as_deref().filter(|path| !path.trim().is_empty()) { + let file = File::open(ca_cert_path).map_err(|e| format!("读取 MQTT CA 证书失败: {e}"))?; + let certs = rustls_pemfile::certs(&mut BufReader::new(file)) + .collect::, _>>() + .map_err(|e| format!("解析 MQTT CA 证书失败: {e}"))?; + let (added, _) = root_cert_store.add_parsable_certificates(certs); + if added == 0 { + return Err("MQTT CA 证书文件不包含有效证书".to_string()); + } + } else { + root_cert_store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned()); + } + + let builder = match mode { + MqttTlsVerificationMode::VerifyServerCert => { + rustls::ClientConfig::builder().with_root_certificates(root_cert_store) + } + MqttTlsVerificationMode::SkipServerCertVerification => { + let provider = Arc::new(rustls::crypto::ring::default_provider()); + rustls::ClientConfig::builder() + .dangerous() + .with_custom_certificate_verifier(Arc::new(NoCertificateVerification { provider })) + } + }; + let cert_file = File::open(client_cert_path).map_err(|e| format!("读取 MQTT 客户端证书失败: {e}"))?; + let certs = rustls_pemfile::certs(&mut BufReader::new(cert_file)) + .collect::, _>>() + .map_err(|e| format!("解析 MQTT 客户端证书失败: {e}"))?; + if certs.is_empty() { + return Err("MQTT 客户端证书文件不包含有效证书".to_string()); + } + let key_file = File::open(client_key_path).map_err(|e| format!("读取 MQTT 客户端私钥失败: {e}"))?; + let mut key_reader = BufReader::new(key_file); + let key = rustls_pemfile::private_key(&mut key_reader) + .map_err(|e| format!("解析 MQTT 客户端私钥失败: {e}"))? + .ok_or("MQTT 客户端私钥文件不包含有效私钥")?; + let config = + builder.with_client_auth_cert(certs, key).map_err(|e| format!("构建 MQTT 客户端 TLS 配置失败: {e}"))?; + Ok(TlsConfiguration::from(config)) +} fn insecure_tls_configuration() -> TlsConfiguration { let provider = Arc::new(rustls::crypto::ring::default_provider()); let config = rustls::ClientConfig::builder() @@ -1503,7 +1601,7 @@ mod tests { } #[test] - fn connect_plan_uses_ws_path_and_insecure_tls_transport() { + fn connect_plan_uses_ws_path_and_insecure_tls_transport() -> Result<(), Box> { let config = MqttConnectionConfig { host: "broker.example.com".to_string(), port: 8084, @@ -1514,13 +1612,15 @@ mod tests { ..Default::default() }; - let plan = build_connect_plan(&config); + let plan = build_connect_plan(&config)?; assert_eq!(plan.broker_addr, "wss://broker.example.com:8084/custom/mqtt"); assert_eq!(plan.backend, MqttBackendKind::V5); match plan.transport { Transport::Wss(TlsConfiguration::Rustls(_)) => {} _ => panic!("tlsSkipVerify 应使用显式 Rustls 自定义校验器"), } + + Ok(()) } #[tokio::test] @@ -1572,7 +1672,7 @@ mod tests { let config = mqtt_config(port, crate::mqtt::types::MqttProtocolVersion::V4); let client = MqttClient::connect(config).await.unwrap(); - client.subscribe("device/reconnect/#", MqttQoS::AtMostOnce).await.unwrap(); + client.subscribe("device/reconnect/#", MqttQoS::AtMostOnce, false).await.unwrap(); assert!(client.broker_info().await.connected); assert_eq!(client.list_topics().await.len(), 1); @@ -1597,7 +1697,9 @@ mod tests { let subscribing_client = Arc::clone(&client); let subscribe = - tokio::spawn(async move { subscribing_client.subscribe("device/ack/#", MqttQoS::AtLeastOnce).await }); + tokio::spawn( + async move { subscribing_client.subscribe("device/ack/#", MqttQoS::AtLeastOnce, false).await }, + ); subscribe_seen.await.unwrap(); assert!(client.list_topics().await.is_empty()); allow_suback.send(()).unwrap(); @@ -1687,7 +1789,7 @@ mod tests { let config = mqtt_config(port, crate::mqtt::types::MqttProtocolVersion::V4); let client = MqttClient::connect(config).await.unwrap(); - client.subscribe("device/downgraded/#", MqttQoS::ExactlyOnce).await.unwrap(); + client.subscribe("device/downgraded/#", MqttQoS::ExactlyOnce, false).await.unwrap(); assert_eq!(client.list_topics().await, vec![("device/downgraded/#".to_string(), MqttQoS::AtMostOnce)]); client.disconnect().await; @@ -1700,14 +1802,14 @@ mod tests { let config = mqtt_config(port, crate::mqtt::types::MqttProtocolVersion::V4); let client = MqttClient::connect(config).await.unwrap(); - client.subscribe("retained/#", MqttQoS::AtMostOnce).await.unwrap(); + client.subscribe("retained/#", MqttQoS::AtMostOnce, false).await.unwrap(); wait_for_message_count(&client, 2).await; let initial = client.get_messages(None, 10).await; assert_eq!(initial.iter().filter(|message| message.payload_text.as_deref() == Some("v1")).count(), 1); assert!(initial.iter().any(|message| message.payload_text.as_deref() == Some("v2"))); client.unsubscribe("retained/#").await.unwrap(); - client.subscribe("retained/#", MqttQoS::AtMostOnce).await.unwrap(); + client.subscribe("retained/#", MqttQoS::AtMostOnce, false).await.unwrap(); wait_for_message_count(&client, 3).await; assert_eq!( client @@ -1777,7 +1879,7 @@ mod tests { let config = mqtt_config(port, crate::mqtt::types::MqttProtocolVersion::V4); let client = MqttClient::connect(config).await.unwrap(); - let error = client.subscribe("device/rejected/#", MqttQoS::AtMostOnce).await.unwrap_err(); + let error = client.subscribe("device/rejected/#", MqttQoS::AtMostOnce, false).await.unwrap_err(); assert!(error.contains("拒绝"), "unexpected error: {error}"); assert!(client.list_topics().await.is_empty()); diff --git a/crates/dbx-core/src/mqtt/service.rs b/crates/dbx-core/src/mqtt/service.rs index 415bbc8bbe..7a26e33f3c 100644 --- a/crates/dbx-core/src/mqtt/service.rs +++ b/crates/dbx-core/src/mqtt/service.rs @@ -31,8 +31,8 @@ pub async fn get_broker_info(client: &Arc) -> Result, topic: &str, qos: MqttQoS) -> Result<(), String> { - client.subscribe(topic, qos).await +pub async fn subscribe(client: &Arc, topic: &str, qos: MqttQoS, no_local: bool) -> Result<(), String> { + client.subscribe(topic, qos, no_local).await } /// 取消订阅 topic diff --git a/crates/dbx-core/src/mqtt/types.rs b/crates/dbx-core/src/mqtt/types.rs index cc4f690ad8..bccb1b5b9d 100644 --- a/crates/dbx-core/src/mqtt/types.rs +++ b/crates/dbx-core/src/mqtt/types.rs @@ -32,7 +32,14 @@ pub enum MqttAuth { #[serde(rename = "password")] Password { username: String, password: String }, #[serde(rename = "certificate")] - Certificate { ca_cert_path: Option, client_cert_path: Option, client_key_path: Option }, + Certificate { + #[serde(rename = "caCertPath")] + ca_cert_path: Option, + #[serde(rename = "clientCertPath")] + client_cert_path: Option, + #[serde(rename = "clientKeyPath")] + client_key_path: Option, + }, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -88,6 +95,9 @@ pub struct MqttConnectionConfig { /// WebSocket 路径（仅 WebSocket 传输时使用） #[serde(default, skip_serializing_if = "Option::is_none")] pub ws_path: Option, + /// 已保存的订阅 Topic，跨重连和应用重启恢复。 + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub saved_topics: Vec, } fn default_keep_alive() -> u64 { @@ -117,6 +127,7 @@ impl Default for MqttConnectionConfig { connect_timeout_secs: default_connect_timeout(), max_packet_size_bytes: default_max_packet_size(), ws_path: None, + saved_topics: Vec::new(), } } } @@ -133,6 +144,9 @@ impl MqttConnectionConfig { if parsed.client_id.trim().is_empty() { return Err("MQTT Client ID 不能为空".to_string()); } + if matches!(parsed.auth, MqttAuth::Certificate { .. }) && !parsed.tls { + return Err("MQTT 证书认证必须启用 TLS".to_string()); + } if !(1024..=268_435_455).contains(&parsed.max_packet_size_bytes) { return Err("MQTT 最大报文大小必须在 1024 到 268435455 字节之间".to_string()); } @@ -170,6 +184,17 @@ impl MqttConnectionConfig { } } +/// 已保存的 MQTT 订阅配置。 +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct MqttSavedTopic { + pub topic: String, + #[serde(default)] + pub qos: MqttQoS, + #[serde(default)] + pub no_local: bool, +} + /// MQTT 消息服务质量 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] #[serde(rename_all = "lowercase")] diff --git a/docs/mqtt-issues-fix-plan.md b/docs/mqtt-issues-fix-plan.md new file mode 100644 index 0000000000..669ea67cfa --- /dev/null +++ b/docs/mqtt-issues-fix-plan.md @@ -0,0 +1,58 @@ +# MQTT Issues 修复计划 + +关联 issue： + +- [#5471](https://github.com/t8y2/dbx/issues/5471)：JSON 发布面板消失、重新打开后 Topic 丢失（P0） +- [#5490](https://github.com/t8y2/dbx/issues/5490)：断线后 Topic 丢失、缺少 SSL 证书登录（P0） +- [#5450](https://github.com/t8y2/dbx/issues/5450)：Payload 切换为 JSON 后发送框消失（P0） +- [#5448](https://github.com/t8y2/dbx/issues/5448)：禁止 MQTT 本地转发、消息分区展示（P1） +- [#5456](https://github.com/t8y2/dbx/issues/5456)：保存 Topic 配置（P1） + +## 现状与根因 + +1. JSON placeholder 中包含原始 `{...}`，切换 JSON 时会被 Vue I18n 当作插值表达式编译，导致发布组件渲染异常。 +2. MQTT Topic 目前只保存在 `MqttClient` 的进程内存中；同一客户端的网络重连可以恢复，但重建客户端或重启 DBX 后会丢失。 +3. `MqttAuth` 已声明 `certificate` 类型，但连接 UI 只提供无认证和账号密码，Rust TLS 构建也只应用账号密码。 +4. 消息已经携带 `sent/received` 方向，但订阅请求没有 MQTT 5 `No Local` 选项，因此发布到自己订阅的 Topic 时会看到 broker 回送副本。 + +## 实施顺序 + +### 1. P0：JSON 发布面板 + +- 将 JSON 示例改为安全参数插值，避免语言包直接出现未转义花括号。 +- 保持 Payload textarea 和发布按钮在编码切换后继续挂载。 +- 增加切换所有 Payload 编码的组件回归测试，并覆盖非法 JSON。 + +### 2. P0/P1：Topic 持久化 + +- 在 MQTT external config 中新增带默认值的订阅配置数组，保持旧连接配置兼容。 +- 订阅成功后保存 Topic、QoS 和订阅选项；取消订阅成功后删除配置。 +- 客户端创建和首次 CONNACK 后恢复保存的订阅；保留现有网络重连恢复逻辑。 +- 使用按连接 ID 的合并更新，避免覆盖用户同时修改的其他连接字段。 + +### 3. P0：TLS 与证书登录 + +- 增加 CA、客户端证书、客户端私钥的选择和校验。 +- Rust 端构建系统 CA、自定义 CA、mTLS 和跳过服务端验证的 Rustls 配置。 +- 只保存证书路径，不把私钥内容写入连接配置。 + +### 4. P1：No Local 与消息分流 + +- MQTT 5 使用 `Filter.nolocal`；MQTT 3.x 明确提示协议不支持。 +- 订阅配置保存 `noLocal` 等 MQTT 5 选项。 +- 消息列表按方向左右展示，并提供全部/接收/发送过滤。 + +## 验收标准 + +- 选择 JSON 后发布面板不消失，合法 JSON 可发布，非法 JSON 有错误提示。 +- 断线重连、关闭并重新打开 DBX 后，已保存 Topic、QoS 和选项自动恢复。 +- TLS、CA、自签 CA、客户端证书认证均可连接，错误证书给出明确错误。 +- MQTT 5 开启 No Local 后不出现自己的 broker 回送副本；关闭后发送和接收消息仍能按方向区分。 + +## 验证命令 + +```text +pnpm typecheck +pnpm test +cargo test -p dbx-core --features mq-admin mqtt +``` diff --git a/packages/app-tests/mqttPayloadCodec.test.ts b/packages/app-tests/mqttPayloadCodec.test.ts index 0195c04a8d..dc46a79ff0 100644 --- a/packages/app-tests/mqttPayloadCodec.test.ts +++ b/packages/app-tests/mqttPayloadCodec.test.ts @@ -54,3 +54,8 @@ test("MQTT JSON publishing sends canonical bytes instead of the original text", assert.match(source, /payloadText:\s*encoding\.value === "plaintext" \? payloadText\.value : null/); assert.doesNotMatch(source, /encoding\.value === "plaintext" \|\| encoding\.value === "json"/); }); +test("MQTT JSON placeholder uses a named interpolation for literal braces", () => { + const source = readFileSync("apps/desktop/src/components/mqtt/MqttPublishDialog.vue", "utf8"); + assert.match(source, /mqttPayloadPlaceholderJson", \{ example:/); + assert.doesNotMatch(source, /mqttPayloadPlaceholderJson"\);/); +}); diff --git a/src-tauri/src/commands/mqtt_cmd.rs b/src-tauri/src/commands/mqtt_cmd.rs index 7653a61c93..96eea74cdc 100644 --- a/src-tauri/src/commands/mqtt_cmd.rs +++ b/src-tauri/src/commands/mqtt_cmd.rs @@ -1,3 +1,4 @@ +use serde_json::json; use std::sync::Arc; use tauri::State; @@ -19,6 +20,25 @@ async fn get_mqtt_client( } } +async fn persist_mqtt_topics( + state: &AppState, + connection_id: &str, + client: &dbx_core::mqtt::client::MqttClient, +) -> Result<(), String> { + let saved_topics = client.desired_topic_configs().await; + let Some(mut config) = state.configs.read().await.get(connection_id).cloned() else { + return Ok(()); + }; + let mut external_config = config.external_config.take().unwrap_or_else(|| json!({})); + let Some(external_object) = external_config.as_object_mut() else { + return Err("MQTT external_config 必须是 JSON 对象".to_string()); + }; + external_object.insert("savedTopics".to_string(), serde_json::to_value(saved_topics).map_err(|e| e.to_string())?); + config.external_config = Some(external_config); + state.storage.save_connections(std::slice::from_ref(&config)).await?; + state.configs.write().await.insert(connection_id.to_string(), config); + Ok(()) +} /// 获取 broker 基本信息 #[tauri::command] pub async fn mqtt_get_broker_info( @@ -36,9 +56,11 @@ pub async fn mqtt_subscribe( connection_id: String, topic: String, qos: Option, + no_local: Option, ) -> Result<(), String> { let client = get_mqtt_client(&state, &connection_id).await?; - service::subscribe(&client, &topic, qos.unwrap_or_default()).await + service::subscribe(&client, &topic, qos.unwrap_or_default(), no_local.unwrap_or(false)).await?; + persist_mqtt_topics(state.inner(), &connection_id, &client).await } /// 取消订阅 topic @@ -49,7 +71,8 @@ pub async fn mqtt_unsubscribe( topic: String, ) -> Result<(), String> { let client = get_mqtt_client(&state, &connection_id).await?; - service::unsubscribe(&client, &topic).await + service::unsubscribe(&client, &topic).await?; + persist_mqtt_topics(state.inner(), &connection_id, &client).await } /// 发布消息