From bad7f6b472d47076c0f39957e842df544996f907 Mon Sep 17 00:00:00 2001 From: jinmo <9446650@qq.com> Date: Sat, 8 Aug 2026 11:56:32 +0800 Subject: [PATCH] feat(mqtt): optimize console and subscription sync --- .../src/components/mqtt/MqttAdminConsole.vue | 357 ++++-- .../src/components/mqtt/TopicTreeNode.vue | 96 +- apps/desktop/src/i18n/locales/en.ts | 30 +- apps/desktop/src/i18n/locales/es.ts | 30 +- apps/desktop/src/i18n/locales/it.ts | 30 +- apps/desktop/src/i18n/locales/ja.ts | 30 +- apps/desktop/src/i18n/locales/pt-BR.ts | 30 +- apps/desktop/src/i18n/locales/zh-CN.ts | 30 +- apps/desktop/src/i18n/locales/zh-TW.ts | 30 +- apps/desktop/src/lib/backend/api.ts | 3 + apps/desktop/src/lib/backend/mqtt-http.ts | 12 + apps/desktop/src/lib/backend/mqtt-tauri.ts | 12 + apps/desktop/src/lib/sidebar/treeNodeClick.ts | 2 +- apps/desktop/src/stores/connectionStore.ts | 23 +- apps/desktop/src/types/mqtt.ts | 2 + crates/dbx-core/src/cloud_sync.rs | 172 +++ crates/dbx-core/src/db/agent_driver.rs | 6 +- crates/dbx-core/src/mqtt/client.rs | 69 +- crates/dbx-core/src/mqtt/types.rs | 8 + .../mqtt-console-optimization-requirements.md | 1031 +++++++++++++++++ src-tauri/src/commands/mqtt_cmd.rs | 34 + src-tauri/src/lib.rs | 6 + 22 files changed, 1814 insertions(+), 229 deletions(-) create mode 100644 docs/mqtt-console-optimization-requirements.md diff --git a/apps/desktop/src/components/mqtt/MqttAdminConsole.vue b/apps/desktop/src/components/mqtt/MqttAdminConsole.vue index 6845e777db..472aeb7e62 100644 --- a/apps/desktop/src/components/mqtt/MqttAdminConsole.vue +++ b/apps/desktop/src/components/mqtt/MqttAdminConsole.vue @@ -1,9 +1,10 @@ diff --git a/apps/desktop/src/components/mqtt/TopicTreeNode.vue b/apps/desktop/src/components/mqtt/TopicTreeNode.vue index 1ca6be6c54..cccc4dfffa 100644 --- a/apps/desktop/src/components/mqtt/TopicTreeNode.vue +++ b/apps/desktop/src/components/mqtt/TopicTreeNode.vue @@ -1,6 +1,6 @@ diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts index f78d43498a..dab08b121d 100644 --- a/apps/desktop/src/i18n/locales/en.ts +++ b/apps/desktop/src/i18n/locales/en.ts @@ -567,12 +567,6 @@ export default { mqttSent: "Sent", mqttReceived: "Received", mqttRetained: "Retained", - mqttUnsubscribeMissing: "No active subscriptions found. Refresh topics and try again.", - mqttUnsubscribeConfirm: 'Unsubscribe from "{topic}"?', - mqttUnsubscribeGroupConfirm: "Unsubscribe from {count} topics in this group?", - mqttUnsubscribeFailed: "Failed to unsubscribe from {count} topic(s): {errors}", - mqttUnsubscribeTopic: "Unsubscribe from {topic}", - mqttUnsubscribeGroup: "Unsubscribe from {count} topics in this group", mqttTopic: "Topic", mqttTopicExample: "e.g. sensors/temperature", mqttRetainMessage: "Retain message", @@ -591,6 +585,30 @@ export default { mqttQosAtMostOnce: "At most once", mqttQosAtLeastOnce: "At least once", mqttQosExactlyOnce: "Exactly once", + mqttSubscriptionConfig: "Subscription configuration", + mqttNewSubscription: "New subscription", + mqttEditSubscription: "Edit subscription", + mqttSearchSubscriptionsPlaceholder: "Search subscription filters...", + mqttNoSavedSubscriptions: 'No saved subscriptions. Click "New subscription" to add one.', + mqttTopicFilter: "Topic Filter", + mqttTopicFilterPlaceholder: "e.g. device/+/status", + mqttTopicFilterHint: "Single-level + and multi-level # wildcards are supported.", + mqttNoLocalMqtt5Only: "MQTT 5.0 only", + mqttEnableSubscription: "Enable subscription", + mqttEnableSubscriptionHint: "Subscribe immediately after saving and restore on reconnect.", + mqttSaving: "Saving...", + mqttSubscriptionTopicFilterInvalid: "Invalid Topic Filter: + must occupy one level, and # must be the final level.", + mqttSubscriptionTopicFilterDuplicate: "This Topic Filter already exists.", + mqttSubscriptionNoLocalProtocolError: "No Local is not supported by MQTT 3.x. Switch to MQTT 5.0.", + mqttDeleteSubscriptionConfirm: 'Delete subscription configuration "{topic}"?', + mqttSubscriptionStatusActive: "Subscribed", + mqttSubscriptionStatusPending: "Pending", + mqttSubscriptionStatusDisabled: "Disabled", + mqttSubscriptionDisable: "Disable subscription", + mqttSubscriptionEnable: "Enable subscription", + mqttSubscriptionEdit: "Edit configuration", + mqttSubscriptionDelete: "Delete configuration", + mqttSubscriptionOperationFailed: "Subscription operation failed: {error}. The configuration was kept.", searchDatabasePlaceholder: "Search database types", searchResults: "Search results", databaseCategories: "Categories", diff --git a/apps/desktop/src/i18n/locales/es.ts b/apps/desktop/src/i18n/locales/es.ts index 1a3309ec16..6e5d83ced9 100644 --- a/apps/desktop/src/i18n/locales/es.ts +++ b/apps/desktop/src/i18n/locales/es.ts @@ -794,12 +794,6 @@ export default withEnglishFallback({ mqttSent: "Enviado", mqttReceived: "Recibido", mqttRetained: "Retenido", - mqttUnsubscribeMissing: "No se encontraron suscripciones activas. Actualiza los Topics e inténtalo de nuevo.", - mqttUnsubscribeConfirm: '¿Cancelar la suscripción a "{topic}"?', - mqttUnsubscribeGroupConfirm: "¿Cancelar la suscripción a {count} Topics de este grupo?", - mqttUnsubscribeFailed: "No se pudo cancelar la suscripción a {count} Topic(s): {errors}", - mqttUnsubscribeTopic: "Cancelar la suscripción a {topic}", - mqttUnsubscribeGroup: "Cancelar la suscripción a {count} Topics de este grupo", mqttTopic: "Topic", mqttTopicExample: "p. ej. sensors/temperature", mqttRetainMessage: "Retener mensaje", @@ -818,6 +812,30 @@ export default withEnglishFallback({ mqttQosAtMostOnce: "Como máximo una vez", mqttQosAtLeastOnce: "Al menos una vez", mqttQosExactlyOnce: "Exactamente una vez", + mqttSubscriptionConfig: "Configuración de suscripción", + mqttNewSubscription: "Nueva suscripción", + mqttEditSubscription: "Editar suscripción", + mqttSearchSubscriptionsPlaceholder: "Buscar filtros de suscripción...", + mqttNoSavedSubscriptions: "No hay suscripciones guardadas. Haz clic en «Nueva suscripción» para añadir una.", + mqttTopicFilter: "Filtro de Topic", + mqttTopicFilterPlaceholder: "p. ej. device/+/status", + mqttTopicFilterHint: "Se admiten los comodines de un nivel + y de varios niveles #.", + mqttNoLocalMqtt5Only: "Solo MQTT 5.0", + mqttEnableSubscription: "Activar suscripción", + mqttEnableSubscriptionHint: "Suscríbete inmediatamente tras guardar y restaura al reconectar.", + mqttSaving: "Guardando...", + mqttSubscriptionTopicFilterInvalid: "Filtro de Topic no válido: + debe ocupar un nivel y # debe estar al final.", + mqttSubscriptionTopicFilterDuplicate: "Este filtro de Topic ya existe.", + mqttSubscriptionNoLocalProtocolError: "No Local no es compatible con MQTT 3.x. Cambia a MQTT 5.0.", + mqttDeleteSubscriptionConfirm: "¿Eliminar la configuración de suscripción «{topic}»?", + mqttSubscriptionStatusActive: "Suscrito", + mqttSubscriptionStatusPending: "Suscribiendo", + mqttSubscriptionStatusDisabled: "Desactivado", + mqttSubscriptionDisable: "Desactivar suscripción", + mqttSubscriptionEnable: "Activar suscripción", + mqttSubscriptionEdit: "Editar configuración", + mqttSubscriptionDelete: "Eliminar configuración", + mqttSubscriptionOperationFailed: "Error en la operación de suscripción: {error}. La configuración se conservó.", damengJvmOptions: "Propiedades del sistema JVM", damengJvmOptionsPlaceholder: "-Djava.net.preferIPv4Stack=true", damengJvmOptionsHint: "Una propiedad -Dkey o -Dkey=value por línea. Los valores con espacios no necesitan comillas del shell.", diff --git a/apps/desktop/src/i18n/locales/it.ts b/apps/desktop/src/i18n/locales/it.ts index 248904d857..e856e621e9 100644 --- a/apps/desktop/src/i18n/locales/it.ts +++ b/apps/desktop/src/i18n/locales/it.ts @@ -792,12 +792,6 @@ export default withEnglishFallback({ mqttSent: "Inviato", mqttReceived: "Ricevuto", mqttRetained: "Mantenuto", - mqttUnsubscribeMissing: "Nessuna sottoscrizione attiva trovata. Aggiorna i Topic e riprova.", - mqttUnsubscribeConfirm: 'Annullare la sottoscrizione a "{topic}"?', - mqttUnsubscribeGroupConfirm: "Annullare la sottoscrizione a {count} Topic in questo gruppo?", - mqttUnsubscribeFailed: "Impossibile annullare la sottoscrizione a {count} Topic: {errors}", - mqttUnsubscribeTopic: "Annulla la sottoscrizione a {topic}", - mqttUnsubscribeGroup: "Annulla la sottoscrizione a {count} Topic in questo gruppo", mqttTopic: "Topic", mqttTopicExample: "ad es. sensors/temperature", mqttRetainMessage: "Mantieni messaggio", @@ -816,6 +810,30 @@ export default withEnglishFallback({ mqttQosAtMostOnce: "Al massimo una volta", mqttQosAtLeastOnce: "Almeno una volta", mqttQosExactlyOnce: "Esattamente una volta", + mqttSubscriptionConfig: "Configurazione sottoscrizione", + mqttNewSubscription: "Nuova sottoscrizione", + mqttEditSubscription: "Modifica sottoscrizione", + mqttSearchSubscriptionsPlaceholder: "Cerca filtri di sottoscrizione...", + mqttNoSavedSubscriptions: "Nessuna sottoscrizione salvata. Fai clic su «Nuova sottoscrizione» per aggiungerne una.", + mqttTopicFilter: "Filtro Topic", + mqttTopicFilterPlaceholder: "es. device/+/status", + mqttTopicFilterHint: "Sono supportati i caratteri jolly a livello singolo + e multilivello #.", + mqttNoLocalMqtt5Only: "Solo MQTT 5.0", + mqttEnableSubscription: "Abilita sottoscrizione", + mqttEnableSubscriptionHint: "Sottoscrivi subito dopo il salvataggio e ripristina alla riconnessione.", + mqttSaving: "Salvataggio...", + mqttSubscriptionTopicFilterInvalid: "Filtro Topic non valido: + deve occupare un livello e # deve essere l'ultimo livello.", + mqttSubscriptionTopicFilterDuplicate: "Questo filtro Topic esiste già.", + mqttSubscriptionNoLocalProtocolError: "No Local non è supportato da MQTT 3.x. Passa a MQTT 5.0.", + mqttDeleteSubscriptionConfirm: "Eliminare la configurazione della sottoscrizione «{topic}»?", + mqttSubscriptionStatusActive: "Sottoscritto", + mqttSubscriptionStatusPending: "In sottoscrizione", + mqttSubscriptionStatusDisabled: "Disabilitato", + mqttSubscriptionDisable: "Disabilita sottoscrizione", + mqttSubscriptionEnable: "Abilita sottoscrizione", + mqttSubscriptionEdit: "Modifica configurazione", + mqttSubscriptionDelete: "Elimina configurazione", + mqttSubscriptionOperationFailed: "Operazione di sottoscrizione non riuscita: {error}. La configurazione è stata conservata.", damengJvmOptions: "Proprietà di sistema JVM", damengJvmOptionsPlaceholder: "-Djava.net.preferIPv4Stack=true", damengJvmOptionsHint: "Una proprietà -Dkey o -Dkey=value per riga. I valori con spazi non richiedono virgolette della shell.", diff --git a/apps/desktop/src/i18n/locales/ja.ts b/apps/desktop/src/i18n/locales/ja.ts index b202c867df..f048cede25 100644 --- a/apps/desktop/src/i18n/locales/ja.ts +++ b/apps/desktop/src/i18n/locales/ja.ts @@ -792,12 +792,6 @@ export default withEnglishFallback({ mqttSent: "送信", mqttReceived: "受信", mqttRetained: "保持", - mqttUnsubscribeMissing: "解除できる購読が見つかりません。トピックを更新して、もう一度お試しください。", - mqttUnsubscribeConfirm: "「{topic}」の購読を解除しますか?", - mqttUnsubscribeGroupConfirm: "このグループの {count} 件のトピック購読を解除しますか?", - mqttUnsubscribeFailed: "{count} 件のトピック購読を解除できませんでした: {errors}", - mqttUnsubscribeTopic: "{topic} の購読を解除", - mqttUnsubscribeGroup: "このグループの {count} 件のトピック購読を解除", mqttTopic: "トピック", mqttTopicExample: "例: sensors/temperature", mqttRetainMessage: "メッセージを保持", @@ -816,6 +810,30 @@ export default withEnglishFallback({ mqttQosAtMostOnce: "最大 1 回", mqttQosAtLeastOnce: "最低 1 回", mqttQosExactlyOnce: "正確に 1 回", + mqttSubscriptionConfig: "購読設定", + mqttNewSubscription: "購読を新規作成", + mqttEditSubscription: "購読を編集", + mqttSearchSubscriptionsPlaceholder: "購読フィルターを検索...", + mqttNoSavedSubscriptions: "保存された購読はありません。「購読を新規作成」をクリックしてください。", + mqttTopicFilter: "Topic Filter", + mqttTopicFilterPlaceholder: "例: device/+/status", + mqttTopicFilterHint: "単一レベルの + と複数レベルの # ワイルドカードに対応しています。", + mqttNoLocalMqtt5Only: "MQTT 5.0 のみ対応", + mqttEnableSubscription: "購読を有効化", + mqttEnableSubscriptionHint: "保存後すぐに購読し、再接続時に自動復元します。", + mqttSaving: "保存中...", + mqttSubscriptionTopicFilterInvalid: "Topic Filter が無効です。+ は 1 レベルのみ、# は最後のレベルに指定してください。", + mqttSubscriptionTopicFilterDuplicate: "この Topic Filter は既に存在します。", + mqttSubscriptionNoLocalProtocolError: "MQTT 3.x は No Local に対応していません。MQTT 5.0 に切り替えてください。", + mqttDeleteSubscriptionConfirm: "購読設定「{topic}」を削除しますか?", + mqttSubscriptionStatusActive: "購読済み", + mqttSubscriptionStatusPending: "購読中", + mqttSubscriptionStatusDisabled: "無効", + mqttSubscriptionDisable: "購読を無効化", + mqttSubscriptionEnable: "購読を有効化", + mqttSubscriptionEdit: "設定を編集", + mqttSubscriptionDelete: "設定を削除", + mqttSubscriptionOperationFailed: "購読操作に失敗しました: {error}。設定は保持されています。", damengJvmOptions: "JVM システムプロパティ", damengJvmOptionsPlaceholder: "-Djava.net.preferIPv4Stack=true", damengJvmOptionsHint: "1 行に 1 つの -Dkey または -Dkey=value を入力します。空白を含む値にもシェル引用符は不要です。", diff --git a/apps/desktop/src/i18n/locales/pt-BR.ts b/apps/desktop/src/i18n/locales/pt-BR.ts index 827a16e9b7..c03a48e59f 100644 --- a/apps/desktop/src/i18n/locales/pt-BR.ts +++ b/apps/desktop/src/i18n/locales/pt-BR.ts @@ -793,12 +793,6 @@ export default withEnglishFallback({ mqttSent: "Enviado", mqttReceived: "Recebido", mqttRetained: "Retido", - mqttUnsubscribeMissing: "Nenhuma inscrição ativa encontrada. Atualize os Topics e tente novamente.", - mqttUnsubscribeConfirm: 'Cancelar a inscrição em "{topic}"?', - mqttUnsubscribeGroupConfirm: "Cancelar a inscrição em {count} Topics deste grupo?", - mqttUnsubscribeFailed: "Falha ao cancelar a inscrição em {count} Topic(s): {errors}", - mqttUnsubscribeTopic: "Cancelar a inscrição em {topic}", - mqttUnsubscribeGroup: "Cancelar a inscrição em {count} Topics deste grupo", mqttTopic: "Topic", mqttTopicExample: "ex.: sensors/temperature", mqttRetainMessage: "Reter mensagem", @@ -817,6 +811,30 @@ export default withEnglishFallback({ mqttQosAtMostOnce: "No máximo uma vez", mqttQosAtLeastOnce: "Pelo menos uma vez", mqttQosExactlyOnce: "Exatamente uma vez", + mqttSubscriptionConfig: "Configuração de inscrição", + mqttNewSubscription: "Nova inscrição", + mqttEditSubscription: "Editar inscrição", + mqttSearchSubscriptionsPlaceholder: "Pesquisar filtros de inscrição...", + mqttNoSavedSubscriptions: "Nenhuma inscrição salva. Clique em «Nova inscrição» para adicionar uma.", + mqttTopicFilter: "Filtro de Topic", + mqttTopicFilterPlaceholder: "ex.: device/+/status", + mqttTopicFilterHint: "Os curingas de nível único + e de vários níveis # são compatíveis.", + mqttNoLocalMqtt5Only: "Somente MQTT 5.0", + mqttEnableSubscription: "Ativar inscrição", + mqttEnableSubscriptionHint: "Inscreva imediatamente após salvar e restaure ao reconectar.", + mqttSaving: "Salvando...", + mqttSubscriptionTopicFilterInvalid: "Filtro de Topic inválido: + deve ocupar um nível e # deve ser o último nível.", + mqttSubscriptionTopicFilterDuplicate: "Este filtro de Topic já existe.", + mqttSubscriptionNoLocalProtocolError: "No Local não é compatível com MQTT 3.x. Mude para MQTT 5.0.", + mqttDeleteSubscriptionConfirm: "Excluir a configuração de inscrição «{topic}»?", + mqttSubscriptionStatusActive: "Inscrito", + mqttSubscriptionStatusPending: "Inscrevendo", + mqttSubscriptionStatusDisabled: "Desativado", + mqttSubscriptionDisable: "Desativar inscrição", + mqttSubscriptionEnable: "Ativar inscrição", + mqttSubscriptionEdit: "Editar configuração", + mqttSubscriptionDelete: "Excluir configuração", + mqttSubscriptionOperationFailed: "Falha na operação de inscrição: {error}. A configuração foi mantida.", damengJvmOptions: "Propriedades do sistema da JVM", damengJvmOptionsPlaceholder: "-Djava.net.preferIPv4Stack=true", damengJvmOptionsHint: "Uma propriedade -Dkey ou -Dkey=value por linha. Valores com espaços não precisam de aspas do shell.", diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts index 2eb5c60833..0ec0b60347 100644 --- a/apps/desktop/src/i18n/locales/zh-CN.ts +++ b/apps/desktop/src/i18n/locales/zh-CN.ts @@ -567,12 +567,6 @@ export default withEnglishFallback({ mqttSent: "发送", mqttReceived: "接收", mqttRetained: "保留", - mqttUnsubscribeMissing: "未找到可取消的订阅,Topic 列表可能已经刷新,请重试。", - mqttUnsubscribeConfirm: "确定要取消订阅“{topic}”吗?", - mqttUnsubscribeGroupConfirm: "确定要取消此分组下的 {count} 个订阅吗?", - mqttUnsubscribeFailed: "有 {count} 个订阅未能取消:{errors}", - mqttUnsubscribeTopic: "取消订阅 {topic}", - mqttUnsubscribeGroup: "取消此分组下的 {count} 个订阅", mqttTopic: "Topic", mqttTopicExample: "例如 sensors/temperature", mqttRetainMessage: "保留消息", @@ -591,6 +585,30 @@ export default withEnglishFallback({ mqttQosAtMostOnce: "最多一次", mqttQosAtLeastOnce: "至少一次", mqttQosExactlyOnce: "精确一次", + mqttSubscriptionConfig: "订阅配置", + mqttNewSubscription: "新建订阅", + mqttEditSubscription: "编辑订阅", + mqttSearchSubscriptionsPlaceholder: "搜索订阅过滤器...", + mqttNoSavedSubscriptions: "暂无订阅配置,请点击“新建订阅”添加。", + mqttTopicFilter: "Topic Filter", + mqttTopicFilterPlaceholder: "例如 device/+/status", + mqttTopicFilterHint: "支持单层通配符 + 和多层通配符 #。", + mqttNoLocalMqtt5Only: "仅 MQTT 5.0 支持", + mqttEnableSubscription: "启用订阅", + mqttEnableSubscriptionHint: "保存后立即订阅,重新连接时自动恢复。", + mqttSaving: "保存中...", + mqttSubscriptionTopicFilterInvalid: "Topic Filter 格式无效:+ 必须独占一个层级,# 必须位于最后一个层级。", + mqttSubscriptionTopicFilterDuplicate: "该 Topic Filter 已经存在。", + mqttSubscriptionNoLocalProtocolError: "MQTT 3.x 不支持禁止本地转发,请切换到 MQTT 5.0。", + mqttDeleteSubscriptionConfirm: "确定要删除订阅配置“{topic}”吗?", + mqttSubscriptionStatusActive: "已订阅", + mqttSubscriptionStatusPending: "订阅中", + mqttSubscriptionStatusDisabled: "未启用", + mqttSubscriptionDisable: "停用订阅", + mqttSubscriptionEnable: "启用订阅", + mqttSubscriptionEdit: "编辑配置", + mqttSubscriptionDelete: "删除配置", + mqttSubscriptionOperationFailed: "订阅操作失败:{error}。配置已保留。", searchDatabasePlaceholder: "搜索数据库类型", searchResults: "搜索结果", databaseCategories: "分类", diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts index 3213a9a96f..9b7122a177 100644 --- a/apps/desktop/src/i18n/locales/zh-TW.ts +++ b/apps/desktop/src/i18n/locales/zh-TW.ts @@ -792,12 +792,6 @@ export default withEnglishFallback({ mqttSent: "傳送", mqttReceived: "接收", mqttRetained: "保留", - mqttUnsubscribeMissing: "找不到可取消的訂閱。Topic 清單可能已重新整理,請再試一次。", - mqttUnsubscribeConfirm: "確定要取消訂閱「{topic}」嗎?", - mqttUnsubscribeGroupConfirm: "確定要取消此群組下的 {count} 個訂閱嗎?", - mqttUnsubscribeFailed: "有 {count} 個訂閱無法取消:{errors}", - mqttUnsubscribeTopic: "取消訂閱 {topic}", - mqttUnsubscribeGroup: "取消此群組下的 {count} 個訂閱", mqttTopic: "Topic", mqttTopicExample: "例如 sensors/temperature", mqttRetainMessage: "保留訊息", @@ -816,6 +810,30 @@ export default withEnglishFallback({ mqttQosAtMostOnce: "最多一次", mqttQosAtLeastOnce: "至少一次", mqttQosExactlyOnce: "恰好一次", + mqttSubscriptionConfig: "訂閱設定", + mqttNewSubscription: "新增訂閱", + mqttEditSubscription: "編輯訂閱", + mqttSearchSubscriptionsPlaceholder: "搜尋訂閱篩選器...", + mqttNoSavedSubscriptions: "尚無訂閱設定,請點擊「新增訂閱」加入。", + mqttTopicFilter: "Topic Filter", + mqttTopicFilterPlaceholder: "例如 device/+/status", + mqttTopicFilterHint: "支援單層萬用字元 + 和多層萬用字元 #。", + mqttNoLocalMqtt5Only: "僅支援 MQTT 5.0", + mqttEnableSubscription: "啟用訂閱", + mqttEnableSubscriptionHint: "儲存後立即訂閱,重新連線時自動恢復。", + mqttSaving: "儲存中...", + mqttSubscriptionTopicFilterInvalid: "Topic Filter 格式無效:+ 必須獨占一個層級,# 必須位於最後一個層級。", + mqttSubscriptionTopicFilterDuplicate: "此 Topic Filter 已存在。", + mqttSubscriptionNoLocalProtocolError: "MQTT 3.x 不支援禁止本地轉發,請切換到 MQTT 5.0。", + mqttDeleteSubscriptionConfirm: "確定要刪除訂閱設定「{topic}」嗎?", + mqttSubscriptionStatusActive: "已訂閱", + mqttSubscriptionStatusPending: "訂閱中", + mqttSubscriptionStatusDisabled: "未啟用", + mqttSubscriptionDisable: "停用訂閱", + mqttSubscriptionEnable: "啟用訂閱", + mqttSubscriptionEdit: "編輯設定", + mqttSubscriptionDelete: "刪除設定", + mqttSubscriptionOperationFailed: "訂閱操作失敗:{error}。設定已保留。", damengJvmOptions: "JVM 系統屬性", damengJvmOptionsPlaceholder: "-Djava.net.preferIPv4Stack=true", damengJvmOptionsHint: "每行一個 -Dkey 或 -Dkey=value;值包含空格時也不要使用 shell 引號。", diff --git a/apps/desktop/src/lib/backend/api.ts b/apps/desktop/src/lib/backend/api.ts index bb2f13a730..ad9c4ed008 100644 --- a/apps/desktop/src/lib/backend/api.ts +++ b/apps/desktop/src/lib/backend/api.ts @@ -771,9 +771,12 @@ export type { // MQTT export const mqttGetBrokerInfo = forward("mqttGetBrokerInfo"); export const mqttSubscribe = forward("mqttSubscribe"); +export const mqttSaveTopicConfig = forward("mqttSaveTopicConfig"); +export const mqttDeleteTopicConfig = forward("mqttDeleteTopicConfig"); export const mqttUnsubscribe = forward("mqttUnsubscribe"); export const mqttPublish = forward("mqttPublish"); export const mqttListTopics = forward("mqttListTopics"); +export const mqttListSavedTopicConfigs = forward("mqttListSavedTopicConfigs"); export const mqttGetTopicTree = forward("mqttGetTopicTree"); export const mqttGetMessages = forward("mqttGetMessages"); export const mqttClearMessages = forward("mqttClearMessages"); diff --git a/apps/desktop/src/lib/backend/mqtt-http.ts b/apps/desktop/src/lib/backend/mqtt-http.ts index f8c0e97b16..deff8068fb 100644 --- a/apps/desktop/src/lib/backend/mqtt-http.ts +++ b/apps/desktop/src/lib/backend/mqtt-http.ts @@ -13,6 +13,14 @@ export async function mqttSubscribe(_connectionId: string, _topic: string, _qos? mqttWebNotAvailable("mqttSubscribe"); } +export async function mqttSaveTopicConfig(_connectionId: string, _topic: string, _qos: string, _noLocal: boolean, _enabled = false) { + mqttWebNotAvailable("mqttSaveTopicConfig"); +} + +export async function mqttDeleteTopicConfig(_connectionId: string, _topic: string) { + mqttWebNotAvailable("mqttDeleteTopicConfig"); +} + export async function mqttUnsubscribe(_connectionId: string, _topic: string) { mqttWebNotAvailable("mqttUnsubscribe"); } @@ -25,6 +33,10 @@ export async function mqttListTopics(_connectionId: string) { mqttWebNotAvailable("mqttListTopics"); } +export async function mqttListSavedTopicConfigs(_connectionId: string) { + mqttWebNotAvailable("mqttListSavedTopicConfigs"); +} + export async function mqttGetTopicTree(_connectionId: string) { mqttWebNotAvailable("mqttGetTopicTree"); } diff --git a/apps/desktop/src/lib/backend/mqtt-tauri.ts b/apps/desktop/src/lib/backend/mqtt-tauri.ts index 93d6ff8f50..0d1c0a66ee 100644 --- a/apps/desktop/src/lib/backend/mqtt-tauri.ts +++ b/apps/desktop/src/lib/backend/mqtt-tauri.ts @@ -8,6 +8,14 @@ export async function mqttSubscribe(connectionId: string, topic: string, qos?: s return invoke("mqtt_subscribe", { connectionId, topic, qos: qos ?? null, noLocal: noLocal ?? false }); } +export async function mqttSaveTopicConfig(connectionId: string, topic: string, qos: string, noLocal: boolean, enabled = false) { + return invoke("mqtt_save_topic_config", { connectionId, config: { topic, qos, noLocal, enabled } }); +} + +export async function mqttDeleteTopicConfig(connectionId: string, topic: string) { + return invoke("mqtt_delete_topic_config", { connectionId, topic }); +} + export async function mqttUnsubscribe(connectionId: string, topic: string) { return invoke("mqtt_unsubscribe", { connectionId, topic }); } @@ -20,6 +28,10 @@ export async function mqttListTopics(connectionId: string) { return invoke("mqtt_list_topics", { connectionId }); } +export async function mqttListSavedTopicConfigs(connectionId: string) { + return invoke("mqtt_list_saved_topic_configs", { connectionId }); +} + export async function mqttGetTopicTree(connectionId: string) { return invoke("mqtt_get_topic_tree", { connectionId }); } diff --git a/apps/desktop/src/lib/sidebar/treeNodeClick.ts b/apps/desktop/src/lib/sidebar/treeNodeClick.ts index 9d77226667..3817699f9c 100644 --- a/apps/desktop/src/lib/sidebar/treeNodeClick.ts +++ b/apps/desktop/src/lib/sidebar/treeNodeClick.ts @@ -8,7 +8,7 @@ export type SidebarActivation = "single" | "double"; const dataNodeTypes = new Set(["table", "view", "materialized_view"]); const documentBrowserNodeTypes = new Set(["mongo-collection", "mongo-bucket"]); -const toggleLeafNodeTypes = new Set(["redis-db", "mq-tenant", "etcd-root", "etcd-dashboard", "etcd-access-control", "zookeeper-root", "mongo-gridfs", "mongo-collection", "mongo-bucket", "vector-collection", "elasticsearch-index", "user-admin"]); +const toggleLeafNodeTypes = new Set(["redis-db", "mq-tenant", "mqtt-topic", "etcd-root", "etcd-dashboard", "etcd-access-control", "zookeeper-root", "mongo-gridfs", "mongo-collection", "mongo-bucket", "vector-collection", "elasticsearch-index", "user-admin"]); const objectBrowserNodeTypes = new Set(["database", "schema", "object-browser"]); const sourceNodeTypes = new Set(["materialized_view", "procedure", "function", "trigger", "sequence", "synonym", "package", "package-body", "type", "type-body"]); const savedSqlNodeTypes = new Set(["saved-sql-file"]); diff --git a/apps/desktop/src/stores/connectionStore.ts b/apps/desktop/src/stores/connectionStore.ts index fcc5ec6929..980a791d29 100644 --- a/apps/desktop/src/stores/connectionStore.ts +++ b/apps/desktop/src/stores/connectionStore.ts @@ -3410,18 +3410,15 @@ export const useConnectionStore = defineStore("connection", () => { node.isLoading = true; try { await ensureConnected(connectionId); - const { mqttGetTopicTree } = await import("@/lib/backend/api"); - const topicTree = (await mqttGetTopicTree(connectionId)) as { name: string; fullPath: string; children?: unknown[]; isLeaf: boolean }; - const topicNodes = mqttTopicTreeToSidebarNodes(connectionId, topicTree); - // Always prepend a synthetic console entry so users can open the MQTT admin - // even when no topics are subscribed yet. + // MQTT subscription state belongs to the console. The global sidebar only + // exposes a single navigation entry and must not keep a second topic tree. const consoleNode: TreeNode = { id: `${connectionId}:mqtt-topic:__console__`, label: "MQTT 控制台", type: "mqtt-topic" as const, connectionId, - children: topicNodes.length > 0 ? topicNodes : [], - isExpanded: topicNodes.length > 0, + children: [], + isExpanded: false, }; setChildren(node, [consoleNode]); node.isExpanded = true; @@ -3433,18 +3430,6 @@ export const useConnectionStore = defineStore("connection", () => { } } - function mqttTopicTreeToSidebarNodes(connectionId: string, tree: { name: string; fullPath: string; children?: unknown[]; isLeaf: boolean }): TreeNode[] { - const children = tree.children ?? []; - return children.map((child: any) => ({ - id: `${connectionId}:mqtt-topic:${child.fullPath}`, - label: child.isLeaf ? child.name : `${child.name}/`, - type: "mqtt-topic", - connectionId, - children: child.children?.length > 0 ? mqttTopicTreeToSidebarNodes(connectionId, child) : [], - isExpanded: false, - })) as TreeNode[]; - } - async function loadMqTenants(connectionId: string, options?: LoadTreeOptions) { const node = findConnectionNode(connectionId); if (!node) return; diff --git a/apps/desktop/src/types/mqtt.ts b/apps/desktop/src/types/mqtt.ts index 3aad08fbd1..ddc476f6e4 100644 --- a/apps/desktop/src/types/mqtt.ts +++ b/apps/desktop/src/types/mqtt.ts @@ -33,6 +33,8 @@ export interface MqttSavedTopic { topic: string; qos: MqttQoS; noLocal?: boolean; + /** Whether the topic is restored and subscribed when the connection opens. */ + enabled?: boolean; } export interface MqttBrokerInfo { diff --git a/crates/dbx-core/src/cloud_sync.rs b/crates/dbx-core/src/cloud_sync.rs index f40eda393d..4bc6251b86 100644 --- a/crates/dbx-core/src/cloud_sync.rs +++ b/crates/dbx-core/src/cloud_sync.rs @@ -8,6 +8,7 @@ use chrono::Utc; use reqwest::{header, Client, Method, StatusCode, Url}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; +use std::collections::HashSet; use crate::ai::AiConfigItem; use crate::connection_secrets::{ @@ -108,6 +109,11 @@ pub struct SyncSnapshot { pub exported_at: String, pub app_version: String, pub connections: Vec, + /// Explicit MQTT subscription metadata. `None` means this is a legacy + /// snapshot that predates MQTT subscription sync and must not clear local + /// subscriptions when applied. + #[serde(default)] + pub mqtt_subscriptions: Option>, /// Shared tunnel profiles (secrets scrubbed). `None` means the snapshot /// predates tunnel profiles — applying it leaves local profiles alone. #[serde(default)] @@ -120,6 +126,33 @@ pub struct SyncSnapshot { pub encrypted_secrets: Option, } +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct MqttSubscriptionSyncEntry { + pub connection_id: String, + pub subscriptions: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct MqttSubscriptionSyncTopic { + pub topic: String, + #[serde(default = "default_sync_qos")] + pub qos: String, + #[serde(default)] + pub no_local: bool, + #[serde(default = "default_sync_enabled")] + pub enabled: bool, +} + +fn default_sync_qos() -> String { + "atmostonce".to_string() +} + +fn default_sync_enabled() -> bool { + true +} + #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct EncryptedSecretsBlob { @@ -221,6 +254,7 @@ pub async fn build_sync_snapshot( )?), None => None, }; + let mqtt_subscriptions = Some(extract_mqtt_subscriptions(&connections)?); for config in &mut connections { scrub_connection_secrets(config); } @@ -233,6 +267,7 @@ pub async fn build_sync_snapshot( exported_at: Utc::now().to_rfc3339(), app_version: app_version.into(), connections, + mqtt_subscriptions, tunnel_profiles: Some(tunnel_profiles), sidebar_layout: storage.load_sidebar_layout().await?, pinned_tree_node_ids: storage.load_pinned_tree_node_ids().await?, @@ -279,6 +314,15 @@ pub async fn apply_sync_snapshot( }; let mut connections = snapshot.connections.clone(); + if let Some(mqtt_subscriptions) = &snapshot.mqtt_subscriptions { + apply_mqtt_subscriptions(&mut connections, mqtt_subscriptions)?; + } else { + // Snapshots created before MQTT subscription sync may still contain a + // stale `savedTopics` value inside `externalConfig`. Preserve the + // current device value instead of allowing that legacy metadata to + // overwrite local subscriptions during restore. + preserve_local_mqtt_subscriptions_for_legacy_snapshot(storage, &mut connections).await?; + } for config in &mut connections { scrub_connection_secrets(config); } @@ -784,6 +828,119 @@ impl SnippetSyncClient { } } +fn extract_mqtt_subscriptions(connections: &[ConnectionConfig]) -> Result, String> { + connections + .iter() + .filter(|config| config.db_type == DatabaseType::Mqtt) + .map(|config| { + let subscriptions = config + .external_config + .as_ref() + .and_then(|external| external.get("savedTopics")) + .cloned() + .unwrap_or_else(|| serde_json::Value::Array(Vec::new())); + let subscriptions: Vec = serde_json::from_value(subscriptions) + .map_err(|error| format!("Invalid MQTT subscriptions for connection {}: {error}", config.id))?; + validate_mqtt_subscriptions(config, &subscriptions)?; + Ok(MqttSubscriptionSyncEntry { connection_id: config.id.clone(), subscriptions }) + }) + .collect() +} + +fn apply_mqtt_subscriptions( + connections: &mut [ConnectionConfig], + entries: &[MqttSubscriptionSyncEntry], +) -> Result<(), String> { + let mut seen_connections = HashSet::new(); + for entry in entries { + if !seen_connections.insert(entry.connection_id.clone()) { + return Err(format!("重复的 MQTT 同步连接 ID: {}", entry.connection_id)); + } + let config = connections + .iter_mut() + .find(|config| config.id == entry.connection_id) + .ok_or_else(|| format!("MQTT 同步配置引用了不存在的连接: {}", entry.connection_id))?; + if config.db_type != DatabaseType::Mqtt { + return Err(format!("同步连接 {} 不是 MQTT 连接", entry.connection_id)); + } + validate_mqtt_subscriptions(config, &entry.subscriptions)?; + let mut external = config.external_config.take().unwrap_or_else(|| serde_json::json!({})); + let object = external + .as_object_mut() + .ok_or_else(|| format!("MQTT 连接 {} 的 externalConfig 必须是 JSON 对象", entry.connection_id))?; + object + .insert("savedTopics".to_string(), serde_json::to_value(&entry.subscriptions).map_err(|e| e.to_string())?); + config.external_config = Some(external); + } + Ok(()) +} + +async fn preserve_local_mqtt_subscriptions_for_legacy_snapshot( + storage: &Storage, + connections: &mut [ConnectionConfig], +) -> Result<(), String> { + let local_connections = storage.load_connections().await?; + for config in connections.iter_mut().filter(|config| config.db_type == DatabaseType::Mqtt) { + let Some(local_config) = local_connections.iter().find(|local| local.id == config.id) else { + continue; + }; + let Some(local_saved_topics) = + local_config.external_config.as_ref().and_then(|external| external.get("savedTopics")).cloned() + else { + continue; + }; + let mut external = config.external_config.take().unwrap_or_else(|| serde_json::json!({})); + if let Some(object) = external.as_object_mut() { + object.insert("savedTopics".to_string(), local_saved_topics); + config.external_config = Some(external); + } else { + config.external_config = Some(serde_json::json!({ "savedTopics": local_saved_topics })); + } + } + Ok(()) +} + +fn validate_mqtt_subscriptions( + config: &ConnectionConfig, + subscriptions: &[MqttSubscriptionSyncTopic], +) -> Result<(), String> { + let protocol = config + .external_config + .as_ref() + .and_then(|external| external.get("protocolVersion")) + .and_then(serde_json::Value::as_str) + .unwrap_or("v5"); + let mut topics = HashSet::new(); + for subscription in subscriptions { + if !valid_mqtt_filter(&subscription.topic) { + return Err(format!("MQTT 连接 {} 包含无效 Topic Filter: {}", config.id, subscription.topic)); + } + if !topics.insert(subscription.topic.clone()) { + return Err(format!("MQTT 连接 {} 包含重复 Topic Filter: {}", config.id, subscription.topic)); + } + if !matches!(subscription.qos.as_str(), "atmostonce" | "atleastonce" | "exactlyonce") { + return Err(format!("MQTT 连接 {} 包含无效 QoS: {}", config.id, subscription.qos)); + } + if subscription.no_local && protocol != "v5" { + return Err(format!("MQTT 连接 {} 使用了 MQTT 3.x 不支持的 No Local", config.id)); + } + } + Ok(()) +} + +fn valid_mqtt_filter(topic: &str) -> bool { + if topic.is_empty() || topic.chars().any(char::is_whitespace) { + return false; + } + let segments: Vec<&str> = topic.split('/').collect(); + segments.iter().enumerate().all(|(index, segment)| { + if *segment == "#" { + return index == segments.len() - 1; + } + !segment.contains('#') && (!segment.contains('+') || *segment == "+") + }) +} + fn scrub_connection_secrets(config: &mut ConnectionConfig) { config.password.clear(); for layer in &mut config.transport_layers { @@ -803,10 +960,25 @@ fn scrub_connection_secrets(config: &mut ConnectionConfig) { config.redis_sentinel_password.clear(); config.connection_string = None; config.init_script = None; + scrub_mqtt_auth_secrets(config); scrub_mq_external_config_secrets(config); scrub_nacos_auth_secrets(config); } +fn scrub_mqtt_auth_secrets(config: &mut ConnectionConfig) { + if config.db_type != DatabaseType::Mqtt { + return; + } + let Some(auth) = config.external_config.as_mut().and_then(|external| external.get_mut("auth")) else { + return; + }; + if auth.get("kind").and_then(serde_json::Value::as_str) == Some("password") { + if let Some(object) = auth.as_object_mut() { + object.insert("password".to_string(), serde_json::Value::String(String::new())); + } + } +} + fn webdav_password_account(config: &WebDavConfig) -> String { let mut hasher = Sha256::new(); hasher.update(config.endpoint.trim().as_bytes()); diff --git a/crates/dbx-core/src/db/agent_driver.rs b/crates/dbx-core/src/db/agent_driver.rs index 7fc7a5aa26..5266778535 100644 --- a/crates/dbx-core/src/db/agent_driver.rs +++ b/crates/dbx-core/src/db/agent_driver.rs @@ -3100,7 +3100,7 @@ mod tests { agent_schema_params, agent_schema_table_params, agent_supports_capability, agent_transaction_params, append_legacy_error_context, decode_agent_response, format_agent_process_error, format_agent_startup_error, is_agent_rpc_response_error, is_unsupported_handshake_error, legacy_agent_call_error, mongo_collection_params, - mongo_database_params, mongo_document_id_params, parse_agent_java_opts, read_agent_line, spawn_agent_process, + mongo_database_params, mongo_document_id_params, parse_agent_java_opts, read_agent_line, start_stderr_collector, validate_dameng_java_system_properties, AgentCallError, AgentCapability, AgentDriverClient, AgentErrorCategory, AgentErrorContext, AgentErrorStage, AgentHandshake, AgentKvMethod, AgentLaunchSpec, AgentMethod, AgentOperationOutcome, AgentRuntimeClient, AgentSessionDisposition, @@ -3127,7 +3127,7 @@ mod tests { fs::write(&program, b"#!/bin/sh\nexit 0\n").unwrap(); fs::set_permissions(&program, fs::Permissions::from_mode(0o600)).unwrap(); - let mut child = spawn_agent_process(&AgentLaunchSpec::new(&program)).unwrap(); + let mut child = super::spawn_agent_process(&AgentLaunchSpec::new(&program)).unwrap(); assert!(child.wait().unwrap().success()); assert_ne!(fs::metadata(&program).unwrap().permissions().mode() & 0o100, 0); @@ -3146,7 +3146,7 @@ mod tests { fs::write(&program, b"#!/bin/sh\nexit 0\n").unwrap(); fs::set_permissions(&program, fs::Permissions::from_mode(0o700)).unwrap(); - let mut child = spawn_agent_process(&AgentLaunchSpec::new(&program)).unwrap(); + let mut child = super::spawn_agent_process(&AgentLaunchSpec::new(&program)).unwrap(); assert!(child.wait().unwrap().success()); assert_eq!(fs::metadata(&program).unwrap().permissions().mode() & 0o777, 0o700); diff --git a/crates/dbx-core/src/mqtt/client.rs b/crates/dbx-core/src/mqtt/client.rs index e1af2cc166..6e398d2d24 100644 --- a/crates/dbx-core/src/mqtt/client.rs +++ b/crates/dbx-core/src/mqtt/client.rs @@ -311,6 +311,8 @@ pub struct MqttClient { granted_subscriptions: RwLock>, /// 需要在无会话重连后恢复的 topic 集合 desired_subscriptions: RwLock>, + /// All persisted subscription configurations, including disabled entries. + saved_topic_configs: RwLock>, no_local_topics: RwLock>, /// 当前连接是否已收到有效 CONNACK connected: AtomicBool, @@ -392,15 +394,24 @@ 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(); + let saved_topic_configs = config.saved_topics.clone(); + let desired_subscriptions = saved_topic_configs + .iter() + .filter(|saved| saved.enabled) + .map(|saved| (saved.topic.clone(), saved.qos)) + .collect(); + let no_local_topics = saved_topic_configs + .iter() + .filter(|saved| saved.enabled && 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(desired_subscriptions), + saved_topic_configs: RwLock::new(saved_topic_configs), no_local_topics: RwLock::new(no_local_topics), connected: AtomicBool::new(false), subscription_requests: Mutex::new(SubscriptionRequestTracker::default()), @@ -848,6 +859,19 @@ impl MqttClient { self.no_local_topics.write().await.remove(&pending.topic); } upsert_subscription(&self.desired_subscriptions, &pending.topic, pending.qos).await; + let mut saved = self.saved_topic_configs.write().await; + if let Some(config) = saved.iter_mut().find(|config| config.topic == pending.topic) { + config.qos = pending.qos; + config.no_local = pending.no_local; + config.enabled = true; + } else { + saved.push(MqttSavedTopic { + topic: pending.topic.clone(), + qos: pending.qos, + no_local: pending.no_local, + enabled: true, + }); + } } Ok(()) } @@ -874,6 +898,11 @@ impl MqttClient { 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); + if let Some(config) = + self.saved_topic_configs.write().await.iter_mut().find(|config| config.topic == pending.topic) + { + config.enabled = false; + } self.seen_retained.write().await.clear_filter(&pending.topic); } else if let Err(error) = &result { log::warn!("{error}"); @@ -960,6 +989,9 @@ impl MqttClient { /// 订阅 topic pub async fn subscribe(&self, topic: &str, qos: MqttQoS, no_local: bool) -> Result<(), String> { validate_topic_filter(topic)?; + if no_local && !matches!(self.config.protocol_version, super::types::MqttProtocolVersion::V5) { + return Err("MQTT 3.x 不支持 No Local 订阅选项".to_string()); + } let has_desired = self .desired_subscriptions .read() @@ -995,6 +1027,30 @@ impl MqttClient { result.await.map_err(|_| "取消订阅确认通道已关闭,请检查 MQTT 连接状态".to_string())? } + /// Save or update a subscription configuration without sending a broker request. + pub async fn save_topic_config(&self, mut config: MqttSavedTopic) -> Result<(), String> { + config.topic = config.topic.trim().to_string(); + validate_topic_filter(&config.topic)?; + if config.no_local && !matches!(self.config.protocol_version, super::types::MqttProtocolVersion::V5) { + return Err("MQTT 3.x 不支持 No Local 订阅选项".to_string()); + } + let mut saved = self.saved_topic_configs.write().await; + if let Some(existing) = saved.iter_mut().find(|current| current.topic == config.topic) { + *existing = config; + } else { + saved.push(config); + } + Ok(()) + } + + /// Delete a saved subscription configuration. The caller must unsubscribe first + /// when the configuration is currently active. + pub async fn delete_topic_config(&self, topic: &str) -> Result<(), String> { + validate_topic_filter(topic)?; + self.saved_topic_configs.write().await.retain(|saved| saved.topic != topic); + Ok(()) + } + /// 发布消息;QoS 0 等待写入网络,QoS 1/2 分别等待 PUBACK/PUBCOMP。 pub async fn publish(&self, request: &MqttPublishRequest) -> Result<(), String> { validate_publish_topic(&request.topic)?; @@ -1055,12 +1111,7 @@ impl MqttClient { } 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() + self.saved_topic_configs.read().await.clone() } /// 获取消息缓冲区中的消息 pub async fn get_messages(&self, topic_filter: Option<&str>, limit: usize) -> Vec { diff --git a/crates/dbx-core/src/mqtt/types.rs b/crates/dbx-core/src/mqtt/types.rs index bccb1b5b9d..2af493e39d 100644 --- a/crates/dbx-core/src/mqtt/types.rs +++ b/crates/dbx-core/src/mqtt/types.rs @@ -193,6 +193,14 @@ pub struct MqttSavedTopic { pub qos: MqttQoS, #[serde(default)] pub no_local: bool, + /// Whether this saved subscription should be restored and subscribed on connect. + /// Missing values from older configs are treated as enabled. + #[serde(default = "default_saved_topic_enabled")] + pub enabled: bool, +} + +fn default_saved_topic_enabled() -> bool { + true } /// MQTT 消息服务质量 diff --git a/docs/mqtt-console-optimization-requirements.md b/docs/mqtt-console-optimization-requirements.md new file mode 100644 index 0000000000..45a11fae59 --- /dev/null +++ b/docs/mqtt-console-optimization-requirements.md @@ -0,0 +1,1031 @@ +# MQTT 控制台优化需求文档 v1.3 + +> 本版本在 v1.2 基础上新增 WebDAV、GitHub 同步 MQTT 订阅配置的需求。本文作为开发基线;当前实现已按本文档完成核心功能,未完成项见文末“实现状态”。 + +## 一、背景与目标 + +MQTT Broker 通常不提供客户端订阅配置的持久化管理。DBX 中保存的 Topic Filter、QoS、No Local、启用状态属于本地配置,如果没有纳入项目同步,在重装、换设备或恢复项目时容易丢失。 + +本次优化目标: + +- 左侧全局侧栏只保留 MQTT 控制台入口。 +- 右侧控制台统一管理全部订阅配置。 +- 支持只保存配置,不立即订阅。 +- 支持启用、停用、编辑和删除订阅配置。 +- 改善 Topic 列表与消息方向的视觉辨识度。 +- WebDAV、GitHub 同步必须包含 MQTT 订阅配置。 +- 同步订阅配置不依赖“同步密码/敏感信息”选项。 +- 下载旧快照时不得意外清空本地 MQTT 订阅配置。 + +## 二、现有问题与现状判断 + +### 2.1 控制台现有问题 + +1. “MQTT 控制台”节点同时承担页面入口和 Topic 父节点职责。 +2. 单击控制台节点时容易执行展开或折叠,而不是打开页面。 +3. MQTT Topic 没有右键功能。 +4. 左侧 Topic 树和右侧控制台分别加载、缓存订阅数据。 +5. 控制台订阅或取消订阅后,侧栏没有同步刷新。 +6. “保存订阅”和“当前订阅”没有明确区分。 +7. 保存配置后会直接订阅,无法仅保存备用配置。 +8. MQTT 通配符订阅与实际消息 Topic 容易混淆。 +9. 多个 Topic 行之间没有明显边界。 +10. Topic 字体较小、颜色较淡。 +11. 接收消息缺少背景色,与控制台底色不容易区分。 +12. 已有控制台标签页可能无法响应外部目标 Topic 的变化。 + +### 2.2 同步现状 + +当前同步快照已经包含连接配置。MQTT 的 `savedTopics` 当前保存在连接的 `externalConfig` 中,因此从数据结构上看,现有同步可能已经间接携带 MQTT 订阅。 + +但目前存在以下不足: + +- MQTT 订阅没有成为明确、独立的同步契约。 +- 缺少针对 WebDAV、GitHub 的 MQTT 订阅往返测试。 +- 无法明确区分“旧快照没有订阅字段”和“用户明确清空全部订阅”。 +- 新增的 `enabled` 状态尚未纳入同步兼容规则。 +- MQTT 认证信息位于 `externalConfig`,必须确认不会因为同步订阅而泄露密码、证书私钥等敏感信息。 +- 下载快照采用连接元数据替换方式,必须明确 MQTT 订阅覆盖规则。 + +因此,本需求不是简单地假设“现有同步已经支持”,而是将 MQTT 订阅同步正式定义为必须保障的功能。 + +## 三、整体信息架构 + +### 3.1 左侧全局侧栏 + +左侧连接树只保留: + +```text +MQTT 连接 +└─ MQTT 控制台 +``` + +要求: + +- 不在“MQTT 控制台”下展开 Topic。 +- 单击直接打开或切换到对应 MQTT 控制台。 +- 不在左侧提供新建、刷新、启用、停用、删除订阅等功能。 +- 不再维护 MQTT Topic 的侧栏数据快照。 +- 不需要 MQTT Topic 右键菜单。 + +左侧全局侧栏仅负责页面导航,不参与订阅状态管理。 + +### 3.2 右侧 MQTT 控制台 + +右侧控制台作为 MQTT 业务操作和状态展示的唯一入口: + +```text +┌──────────────────────────────────────────────────────────┐ +│ MQTT 控制台 刷新连接 清空消息 │ +├────────────────────┬─────────────────────────────────────┤ +│ 订阅配置 │ 当前过滤器:device/+/status │ +│ [+ 新建] [刷新] │ │ +│ [搜索订阅配置] │ 消息列表 │ +│ │ │ +│ ● device/# │ │ +│ ○ device/+/status ├─────────────────────────────────────┤ +│ ! system/events/# │ 发布消息 │ +└────────────────────┴─────────────────────────────────────┘ +``` + +右侧订阅区域负责: + +- 新建订阅配置 +- 编辑订阅配置 +- 启用订阅 +- 停用订阅 +- 删除订阅配置 +- 刷新订阅状态 +- 搜索订阅配置 +- 选择消息过滤器 +- 展示 QoS +- 展示 No Local +- 展示订阅状态和错误信息 + +## 四、MQTT 概念和界面命名 + +### 4.1 订阅配置 + +一条持久化配置包含: + +```text +Topic Filter +QoS +No Local +Enabled +``` + +配置可以保存但不启用。界面主标题建议使用“订阅配置”。 + +### 4.2 订阅过滤器 + +允许使用 MQTT 通配符: + +```text +device/+/status +device/# +system/events/# +``` + +不要将订阅过滤器描述成 Broker 上全部存在的 Topic。 + +### 4.3 发布 Topic + +发布必须使用具体 Topic,不能包含 `+` 或 `#`: + +```text +device/001/control +``` + +### 4.4 实际消息 Topic + +接收或发送消息时展示实际 Topic: + +```text +device/001/status +device/002/status +``` + +例如当前过滤器为 `device/+/status`,消息列表可以同时展示 `device/001/status` 和 `device/002/status`。这属于正常 MQTT 匹配行为,不是数据不一致。 + +## 五、订阅配置数据模型 + +建议数据结构: + +```ts +interface MqttSavedTopic { + topic: string; + qos: MqttQoS; + noLocal?: boolean; + enabled: boolean; +} +``` + +字段含义: + +| 字段 | 含义 | +| --- | --- | +| `topic` | MQTT Topic Filter | +| `qos` | 请求的订阅 QoS | +| `noLocal` | 是否禁止 Broker 将本客户端发布的消息转发回来 | +| `enabled` | 是否启用、立即订阅并在重连时自动恢复 | + +兼容规则: + +```text +enabled 缺失 → true +noLocal 缺失 → false +``` + +Topic Filter 大小写敏感,不得转换大小写。 + +### 5.1 自动恢复规则 + +建立或恢复 MQTT 连接时: + +- `enabled = true`:自动发送订阅请求。 +- `enabled = false`:只保留配置,不发送订阅请求。 + +恢复订阅时必须使用原有 Topic Filter、QoS 和 No Local。 + +## 六、新建订阅配置 + +### 6.1 交互形式 + +采用与 Redis 新增操作一致的模态弹出对话框。 + +不使用: + +- 新标签页 +- 内联展开卡片 +- 侧栏右键表单 + +入口位于右侧订阅配置栏:`+ 新建`。 + +### 6.2 弹窗布局 + +```text +┌────────────── 新建订阅配置 ──────────────┐ +│ │ +│ Topic Filter │ +│ [ device/+/status ] │ +│ 支持单层通配符 + 和多层通配符 # │ +│ │ +│ QoS │ +│ [ QoS 0 ▼ ] │ +│ │ +│ □ 禁止本地转发(No Local) │ +│ │ +│ ☑ 启用订阅 │ +│ 保存后立即订阅,重新连接时自动恢复 │ +│ │ +│ [取消] [保存] │ +└──────────────────────────────────────────┘ +``` + +主按钮统一使用“保存”,不再固定使用“创建并订阅”。 + +### 6.3 启用订阅 + +开启时: + +- 保存订阅配置。 +- 当前已经连接时立即发送订阅请求。 +- 以后重新连接时自动恢复订阅。 + +关闭时: + +- 只保存订阅配置。 +- 当前不发送订阅请求。 +- 以后重新连接时不自动订阅。 + +使用单一“启用订阅”选项,避免同时提供“立即订阅”和“自动订阅”产生复杂组合。 + +### 6.4 Topic Filter 校验 + +要求: + +- 必填。 +- 自动去除首尾空格。 +- 不允许包含空字符。 +- `+` 必须独占一个层级。 +- `#` 必须独占最后一个层级。 +- 不允许重复保存相同 Topic Filter。 +- 输入框按 Enter 可以提交。 + +有效示例: + +```text +device/+/status +device/# +system/events +``` + +无效示例: + +```text +device+ +device/#/status +device/abc# +``` + +### 6.5 QoS + +使用下拉框: + +```text +QoS 0 — 最多一次 +QoS 1 — 至少一次 +QoS 2 — 恰好一次 +``` + +默认值保持当前行为,建议使用 QoS 0。 + +### 6.6 提交状态 + +点击“保存”后: + +1. 禁用表单,避免重复提交。 +2. 按钮显示“保存中…”。 +3. 先持久化配置。 +4. 如果启用,则等待 Broker 订阅确认。 +5. 成功后关闭弹窗。 +6. 刷新右侧订阅配置列表。 +7. 自动选中新配置。 +8. 显示成功反馈。 + +如果只保存未启用配置: + +- 不发送 MQTT 订阅请求。 +- 状态显示为“未启用”。 + +如果订阅失败: + +- 配置仍可保存。 +- 状态显示为“订阅失败”。 +- 保留 Broker 返回的错误信息。 +- 用户可以稍后重试或停用。 + +## 七、禁止本地转发(No Local) + +“禁止本地转发”对应 MQTT 5.0 的 `No Local`。 + +要求: + +- 保存配置时持久化 `noLocal`。 +- 启用和恢复订阅时携带 `noLocal`。 +- 订阅配置列表中展示 `NL` 或 `No Local`。 +- MQTT 5.0 时允许设置。 +- MQTT 3.1 和 MQTT 3.1.1 时禁用。 +- 禁用时提示“仅 MQTT 5.0 支持”。 + +提示说明: + +> 开启后,Broker 不会将当前客户端自己发布的匹配消息再次转发给该客户端。 + +## 八、订阅状态和操作 + +### 8.1 状态定义 + +订阅配置至少包含以下界面状态: + +```text +已订阅 +未启用 +订阅中 +订阅失败 +连接已断开 +``` + +示例: + +```text +● device/+/status QoS 1 NL 已订阅 +○ device/+/alarm QoS 0 未启用 +◌ system/events/# QoS 1 订阅中 +! audit/# QoS 1 订阅失败 +``` + +### 8.2 启用 + +对未启用配置执行“启用”: + +1. 将 `enabled` 更新为 `true`。 +2. 当前连接可用时立即订阅。 +3. 后续重新连接自动恢复。 +4. 状态由“未启用”变为“订阅中”,最终变为“已订阅”或“订阅失败”。 + +### 8.3 停用 + +对启用配置执行“停用”: + +1. 当前已经订阅时发送取消订阅请求。 +2. 将 `enabled` 更新为 `false`。 +3. 保留 Topic Filter、QoS 和 No Local 配置。 +4. 后续重新连接不自动订阅。 + +```text +停用 = 取消当前订阅,但保留配置 +``` + +### 8.4 删除 + +删除订阅配置: + +1. 如果当前已经订阅,先取消订阅。 +2. 删除本地保存配置。 +3. 从订阅配置列表移除。 + +```text +删除 = 取消当前订阅,并永久删除配置 +``` + +删除前需要确认。 + +### 8.5 编辑 + +允许编辑: + +- Topic Filter +- QoS +- No Local +- 启用状态 + +如果修改正在启用的配置: + +1. 取消旧订阅。 +2. 保存新配置。 +3. 使用新配置重新订阅。 + +失败时应提供明确错误,避免旧配置和新配置同时处于不确定状态。 + +## 九、订阅列表视觉优化 + +### 9.1 行布局 + +采用“紧凑列表 + 行间分隔线”,不使用大量独立卡片。 + +建议: + +```text +行高:32~36px +水平内边距:8px +Topic 字号:13px 或 text-sm +分隔线:border-b border-border/50 +``` + +示例: + +```text +● device/+/status QoS 1 NL +──────────────────────────────── +○ device/+/alarm QoS 0 +──────────────────────────────── +! system/events/# QoS 1 NL +``` + +未选中时也必须能区分每一条订阅。 + +### 9.2 字体 + +Topic Filter 使用: + +```text +font-mono +font-medium +text-[13px] 或 text-sm +text-foreground/90 +``` + +辅助文字使用: + +```text +text-xs +text-muted-foreground +``` + +避免 Topic 主文字使用过小或过淡的样式。 + +### 9.3 分组和订阅节点 + +如果使用树形结构: + +```text +▾ device/ + +/ + status QoS 1 NL + alarm QoS 0 +``` + +分组节点: + +- 使用展开或文件夹图标。 +- 使用 `font-semibold`。 +- 不显示 QoS 和 No Local。 + +完整订阅节点: + +- 使用状态图标。 +- 使用等宽字体。 +- 展示 QoS、No Local 和启用状态。 +- 提供编辑、启用、停用、删除操作。 + +如果一个节点既是完整订阅又存在子节点,应同时展示展开按钮和订阅状态。 + +### 9.4 树形引导线 + +增加低对比度引导线: + +```text +device +│ +├─ production +│ ├─ status +│ └─ alarm +└─ test + └─ status +``` + +### 9.5 行状态 + +普通状态: + +```text +bg-transparent +text-foreground/90 +border-b border-border/50 +``` + +悬停状态: + +```text +hover:bg-muted/60 +``` + +选中状态: + +```text +bg-primary/10 +border-l-2 +border-l-primary +text-primary +``` + +未启用: + +```text +text-muted-foreground +``` + +订阅中: + +```text +bg-amber-500/5 +border-l-2 +border-l-amber-400 +``` + +订阅失败: + +```text +bg-destructive/5 +border-l-2 +border-l-destructive +``` + +状态不能只通过颜色表达,还需要状态文字或图标。 + +### 9.6 辅助标签 + +右侧显示: + +```text +QoS 0 +QoS 1 +QoS 2 +NL +已订阅 +未启用 +``` + +要求: + +- 标签不能挤压 Topic 主文字。 +- Topic 过长时省略。 +- 悬停显示完整 Topic。 +- `NL` 悬停显示完整含义。 + +## 十、消息列表视觉优化 + +### 10.1 接收消息 + +- 靠左显示。 +- 使用浅蓝色背景。 +- 使用蓝色左边线。 +- 显示“接收”标签。 + +建议样式: + +```text +bg-blue-50/70 +dark:bg-blue-950/30 +border-l-2 +border-blue-400 +``` + +### 10.2 发送消息 + +- 靠右显示。 +- 使用浅绿色背景。 +- 使用绿色右边线。 +- 显示“发送”标签。 + +建议样式: + +```text +bg-emerald-50/70 +dark:bg-emerald-950/30 +border-r-2 +border-emerald-400 +``` + +### 10.3 消息方向表达 + +方向同时通过以下方式表达: + +- 左右对齐 +- “发送/接收”标签 +- 不同方向边线 +- 不同背景色 + +不能只依赖颜色。 + +## 十一、其他交互 + +### 11.1 刷新订阅配置 + +刷新按钮位于右侧订阅配置区域。 + +刷新后: + +- 保留当前选择。 +- 当前选择不存在时切换到“全部消息”。 +- 不重复创建订阅。 +- 不清空消息缓冲区。 + +### 11.2 搜索订阅配置 + +支持按照完整 Topic Filter 搜索。 + +要求: + +- 搜索结果保留必要的树形上下文,或临时切换为扁平列表。 +- 清除搜索后恢复原展开状态。 + +### 11.3 选择未启用配置 + +点击未启用配置时: + +- 可以查看和编辑配置。 +- 消息区域可以显示该过滤器已有的历史消息。 +- 不应暗示当前正在接收新消息。 +- 明确显示“该订阅尚未启用”。 + +## 十二、状态管理 + +右侧 MQTT 控制台作为唯一状态所有者,维护: + +- Broker 连接状态 +- 已保存订阅配置 +- 当前有效订阅 +- 当前选择的过滤器 +- 启用和停用状态 +- 订阅请求状态 +- 订阅错误 +- 消息列表 +- 发布面板状态 + +左侧连接树不再维护 MQTT Topic Tree。 + +必须区分: + +```text +已保存配置 +已启用配置 +Broker 已确认订阅 +``` + +不能将这三个状态合并为一个列表状态。 + +## 十三、WebDAV 与 GitHub 同步 + +### 13.1 同步范围 + +WebDAV 和 GitHub 项目同步必须包含每个 MQTT 连接的全部订阅配置: + +```text +connectionId +Topic Filter +QoS +No Local +Enabled +``` + +需要同步: + +- 已启用配置 +- 未启用配置 +- QoS +- No Local +- 启用状态 +- 空订阅列表 + +不需要同步: + +- 当前选中的过滤器 +- Topic 树展开状态 +- 当前“订阅中”“订阅失败”等临时运行状态 +- MQTT 消息缓存 +- 已发布或已接收的消息历史 +- 当前连接在线状态 + +### 13.2 与连接的关联 + +订阅配置必须通过稳定的 `connectionId` 与 MQTT 连接关联。 + +下载同步快照后: + +- 连接存在:恢复该连接对应的订阅配置。 +- 连接不存在:由同步快照恢复连接及其订阅配置。 +- 非 MQTT 连接:忽略 MQTT 订阅数据。 +- 不得把一个 MQTT 连接的订阅错误应用到另一个连接。 + +### 13.3 建议的同步快照结构 + +建议在同步快照中将 MQTT 订阅定义为明确字段,而不是只依赖 `externalConfig.savedTopics`: + +```ts +interface MqttSubscriptionSyncEntry { + connectionId: string; + subscriptions: MqttSavedTopic[]; +} + +interface SyncSnapshot { + // 其他现有字段 + mqttSubscriptions?: MqttSubscriptionSyncEntry[]; +} +``` + +语义: + +```text +mqttSubscriptions 缺失 += 旧版本快照,不修改本地 MQTT 订阅配置 + +mqttSubscriptions 存在但某连接 subscriptions 为空 += 明确清空该连接的全部订阅配置 +``` + +该设计可以解决“旧快照没有字段”和“用户明确删除全部订阅”无法区分的问题。 + +`externalConfig.savedTopics` 仍可作为本地连接配置的存储位置,但同步快照应具有明确、可测试的 MQTT 订阅契约。 + +### 13.4 上传规则 + +上传到 WebDAV 或 GitHub 时: + +1. 从持久化存储读取最新 MQTT 订阅配置。 +2. 包含所有已启用和未启用配置。 +3. 不读取仅存在于页面内、尚未保存的表单内容。 +4. 删除的配置不能继续出现在新快照中。 +5. 上传成功提示中显示 MQTT 同步摘要。 + +示例: + +```text +同步完成:3 个 MQTT 连接,18 条订阅配置 +``` + +WebDAV 自动上传也必须包含最新订阅配置。 + +### 13.5 下载规则 + +下载快照前: + +- 保持现有全局下载确认。 +- 增加 MQTT 订阅变化摘要。 +- 明确提示匹配连接中将新增、更新或删除多少条订阅配置。 + +示例: + +```text +MQTT 订阅配置将发生变化: +新增 5 条,更新 2 条,删除 1 条 +``` + +下载应用规则: + +- 本轮沿用现有快照方向语义:下载时远端快照为准。 +- 仅当快照明确包含 MQTT 订阅字段时覆盖本地配置。 +- 旧快照不包含该字段时,保留本地 MQTT 订阅。 +- 整个下载应用过程必须是原子的。 +- 数据校验失败时不允许部分写入。 + +### 13.6 已连接状态下下载 + +下载同步快照时,如果 MQTT 连接当前已打开: + +- 不应静默取消或新增运行中的订阅。 +- 先更新持久化配置。 +- 明确提示“重新连接后应用新的订阅配置”。 +- 当前运行连接保持现状,直到用户重新连接。 +- 后续可选增强“立即应用”,但本轮默认不做自动运行时协调。 + +这样可以避免同步下载过程中突然停止或增加消息接收。 + +### 13.7 冲突策略 + +本轮遵循现有项目同步的上传/下载方向,不单独实现 Topic 级双向自动合并: + +- 上传:本地快照覆盖远端快照。 +- 下载:远端快照覆盖对应本地配置。 +- 下载前展示变化摘要并确认。 + +原因是简单求并集会导致已删除的订阅重新出现。如果未来需要自动双向合并,必须增加: + +```text +updatedAt +deletedAt 或删除墓碑 +设备/版本标识 +``` + +在没有版本和删除记录前,不采用自动合并。 + +### 13.8 安全边界 + +MQTT 订阅配置不是敏感凭据,因此: + +- 无论是否开启“同步敏感信息”,订阅配置都必须同步。 +- Topic Filter、QoS、No Local、Enabled 属于普通元数据。 + +但以下信息不得因订阅同步而以明文泄露: + +- MQTT 用户密码 +- 客户端私钥 +- 私钥密码 +- Token +- 其他认证密钥 + +证书、私钥文件路径属于设备相关配置,不应被当作订阅数据同步覆盖到其他设备。 + +要求: + +- WebDAV 未启用敏感信息同步时,订阅仍同步,但认证秘密必须清除。 +- GitHub 同步继续遵循现有快照加密要求。 +- MQTT `externalConfig` 的秘密清理必须专门验证,不能只验证其他 MQ 类型。 + +### 13.9 数据校验 + +下载时对每条订阅配置校验: + +- `connectionId` 存在且指向 MQTT 连接。 +- Topic Filter 格式有效。 +- Topic Filter 不重复。 +- QoS 值合法。 +- No Local 类型合法。 +- Enabled 类型合法。 +- MQTT 3.1 和 MQTT 3.1.1 下收到 `noLocal = true` 时不得静默错误启用。 + +若快照中存在非法 MQTT 订阅数据: + +- 下载整体失败。 +- 本地订阅配置保持不变。 +- 显示具体连接和 Topic Filter 错误。 +- 不允许只恢复部分订阅。 + +### 13.10 版本兼容 + +兼容规则: + +```text +旧订阅缺少 enabled → true +旧订阅缺少 noLocal → false +旧快照缺少 mqttSubscriptions → 保留本地订阅 +新快照明确 subscriptions: [] → 清空对应连接订阅 +``` + +新版快照被不支持 MQTT 同步的旧版应用读取时,应遵循现有快照版本兼容策略,不得静默损坏远端快照。 + +### 13.11 同步后的界面刷新 + +下载成功后: + +- 重新加载连接配置。 +- 刷新右侧 MQTT 订阅配置列表。 +- 当前选择仍存在时保留选择。 +- 当前选择已被远端删除时切换到“全部消息”。 +- 已连接实例提示重新连接。 +- 未连接实例在下次连接时恢复 `enabled = true` 的订阅。 + +## 十四、预计代码改动范围 + +本章仅作为后续实施范围,本次不进行开发。 + +### 14.1 左侧连接树 + +- `apps/desktop/src/stores/connectionStore.ts` + - MQTT 连接下只创建“MQTT 控制台”节点。 + - 移除侧栏 Topic Tree 加载与转换逻辑。 + - 不再默认展开 MQTT Topic。 + +- `apps/desktop/src/components/sidebar/SidebarTreeRuntimeHost.vue` + - 单击 MQTT 控制台节点直接打开控制台。 + - 不再执行 MQTT Topic 节点展开逻辑。 + - 移除 MQTT Topic 导航分支。 + +### 14.2 MQTT 控制台 + +- `apps/desktop/src/components/mqtt/MqttAdminConsole.vue` + - 作为唯一订阅管理入口。 + - 增加新建订阅配置弹窗。 + - 增加编辑、启用、停用、删除操作。 + - 展示 QoS、No Local、Enabled 和运行状态。 + - 增加搜索和刷新。 + - 优化发送及接收消息样式。 + - 正确处理已有标签页的目标过滤器变化。 + +- `apps/desktop/src/components/mqtt/TopicTreeNode.vue` + - 建议重构为 `MqttSubscriptionTreeNode.vue`。 + - 增加行间分隔线和树形引导线。 + - 提高 Topic 字体可读性。 + - 区分分组和订阅节点。 + - 展示 QoS、No Local、启用状态和运行状态。 + +- 建议新增 `MqttSubscriptionDialog.vue` + - 用于新建和编辑订阅配置。 + - 负责 Topic Filter 校验、QoS、No Local 和启用状态设置。 + +### 14.3 数据类型和后端 + +- `apps/desktop/src/types/mqtt.ts` + - 为保存配置增加 `enabled`。 + - 补充订阅运行状态和同步类型。 + +- `crates/dbx-core/src/mqtt/client.rs` + - 区分保存配置、启用、停用、删除和当前已确认订阅。 + - 连接恢复时仅恢复 `enabled = true` 的配置。 + +- `crates/dbx-core/src/storage.rs` + - 提供完整保存、读取和替换 MQTT 订阅配置的能力。 + - 保证同步应用的原子性。 + +### 14.4 云同步 + +- `crates/dbx-core/src/cloud_sync.rs` + - 明确定义 MQTT 订阅同步字段。 + - 实现导入导出、兼容、校验和摘要规则。 + - 验证 MQTT 敏感信息清理。 + +- `apps/desktop/src/components/editor/EditorSettingsDialog.vue` + - WebDAV/GitHub 下载前展示 MQTT 变化摘要。 + - 下载后提示重新连接。 + +### 14.5 标签页状态 + +- `apps/desktop/src/stores/queryStore.ts` + - 简化 MQTT 控制台打开逻辑。 + - 确保已有标签页响应目标过滤器变化。 + - 移除不再需要的侧栏 Topic 跳转逻辑。 + +## 十五、验收标准 + +### 15.1 页面结构 + +1. MQTT 连接下只显示“MQTT 控制台”。 +2. 控制台节点不再展开 Topic。 +3. 单击控制台节点直接打开或切换页面。 +4. 全局侧栏不保存 MQTT Topic 快照。 + +### 15.2 新建订阅配置 + +5. 新建操作使用与 Redis 新增一致的模态弹窗。 +6. 不打开新标签页,不使用内联卡片。 +7. 主按钮显示“保存”。 +8. 弹窗包含 Topic Filter、QoS、No Local 和“启用订阅”。 +9. 关闭“启用订阅”后只保存配置,不发送订阅请求。 +10. 开启“启用订阅”后保存并立即订阅。 +11. 已启用配置重新连接时自动恢复。 +12. 未启用配置重新连接时不自动订阅。 +13. 历史数据缺少 `enabled` 时按启用处理。 + +### 15.3 订阅操作 + +14. 可以启用未启用配置。 +15. 可以停用当前订阅并保留配置。 +16. 可以删除配置并取消当前订阅。 +17. 可以编辑 Topic Filter、QoS、No Local 和启用状态。 +18. Topic Filter 重复或格式错误时不能保存。 +19. 订阅失败时配置和错误信息可被保留。 + +### 15.4 No Local + +20. No Local 可以保存、恢复和同步。 +21. MQTT 5.0 可以设置 No Local。 +22. MQTT 3.1 和 MQTT 3.1.1 下禁用 No Local。 +23. 列表能够显示 `NL` 状态。 + +### 15.5 订阅列表视觉 + +24. 未选中的订阅行之间也有明确分隔。 +25. Topic 主文字不小于 13px,并具有足够对比度。 +26. 分组和完整订阅节点有明显区别。 +27. 树形结构具有清晰的缩进和引导线。 +28. 每条配置能够展示 QoS、No Local 和启用状态。 +29. 选中状态具有背景色和左侧强调线。 +30. 未启用、订阅中、失败状态均可明确辨认。 +31. 状态不能只通过颜色表达。 +32. Topic 过长时可以查看完整内容。 + +### 15.6 消息列表视觉 + +33. 接收消息为浅蓝色、靠左并带蓝色左边线。 +34. 发送消息为浅绿色、靠右并带绿色右边线。 +35. 消息方向通过位置、标签、边线和颜色共同表达。 +36. 浅色和深色主题下均能清楚区分消息方向。 + +### 15.7 WebDAV 与 GitHub 同步 + +37. WebDAV 上传包含所有 MQTT 订阅配置。 +38. GitHub 上传包含所有 MQTT 订阅配置。 +39. WebDAV 自动上传包含最新订阅配置。 +40. 已启用和未启用配置均可同步。 +41. QoS、No Local、Enabled 往返同步后完全一致。 +42. 不开启敏感信息同步时,MQTT 订阅仍能同步。 +43. 不开启敏感信息同步时,MQTT 密码和私钥不得进入明文快照。 +44. 下载新快照可以恢复全部 MQTT 订阅配置。 +45. 旧快照缺少 MQTT 订阅字段时,不清空本地订阅。 +46. 新快照明确为空列表时,可以清空对应连接的订阅配置。 +47. 非法订阅数据导致整个下载失败,本地数据不变。 +48. 已连接 MQTT 实例下载后不立即改变运行订阅。 +49. 下载后明确提示重新连接。 +50. 未连接实例下次连接时只恢复 `enabled = true` 的配置。 +51. 下载前显示 MQTT 新增、更新和删除数量。 +52. 同步成功结果显示 MQTT 连接数和订阅配置数量。 +53. WebDAV 和 GitHub 都具备独立的 MQTT 同步回归测试。 +54. 连续执行“上传—下载—上传”后,MQTT 订阅数据保持稳定,不发生字段丢失。 + +## 十六、实现状态 + +已完成: + +- 左侧仅保留 MQTT 控制台入口,订阅配置统一在右侧控制台管理。 +- 使用模态弹窗新建和编辑订阅,支持仅保存、启用、停用、删除、QoS、No Local 和 Enabled。 +- 订阅配置与当前运行订阅分离,历史配置缺少 Enabled 时兼容为启用。 +- Topic 树、状态、QoS、No Local、选中态和消息收发方向完成视觉区分。 +- WebDAV/GitHub 快照增加 MQTT 订阅专属字段,支持启用和未启用配置同步,并兼容旧快照。 +- MQTT 密码不进入普通同步快照;同步下载校验非法订阅数据并保留本地原子性。 + +后续可继续完善: + +- 同步设置页面增加 MQTT 新增、更新、删除数量摘要及“请重新连接”提示。 +- 增加 WebDAV/GitHub 上传—下载—再次上传的自动化回归测试。 diff --git a/src-tauri/src/commands/mqtt_cmd.rs b/src-tauri/src/commands/mqtt_cmd.rs index 2a2c557b5c..8e54f0b975 100644 --- a/src-tauri/src/commands/mqtt_cmd.rs +++ b/src-tauri/src/commands/mqtt_cmd.rs @@ -65,6 +65,18 @@ pub async fn mqtt_subscribe( persist_mqtt_topics(state.inner(), &connection_id, &client).await } +/// 保存订阅配置,不向 broker 发送 SUBSCRIBE。 +#[tauri::command] +pub async fn mqtt_save_topic_config( + state: State<'_, Arc>, + connection_id: String, + config: MqttSavedTopic, +) -> Result<(), String> { + let client = get_mqtt_client(&state, &connection_id).await?; + client.save_topic_config(config).await?; + persist_mqtt_topics(state.inner(), &connection_id, &client).await +} + /// 取消订阅 topic #[tauri::command] pub async fn mqtt_unsubscribe( @@ -77,6 +89,18 @@ pub async fn mqtt_unsubscribe( persist_mqtt_topics(state.inner(), &connection_id, &client).await } +/// 删除已保存的订阅配置;当前订阅由调用方先取消。 +#[tauri::command] +pub async fn mqtt_delete_topic_config( + state: State<'_, Arc>, + connection_id: String, + topic: String, +) -> Result<(), String> { + let client = get_mqtt_client(&state, &connection_id).await?; + client.delete_topic_config(&topic).await?; + persist_mqtt_topics(state.inner(), &connection_id, &client).await +} + /// 发布消息 #[tauri::command] pub async fn mqtt_publish( @@ -98,6 +122,16 @@ pub async fn mqtt_list_topics( service::list_topics(&client).await } +/// 获取全部已保存的订阅配置(包含未启用配置)。 +#[tauri::command] +pub async fn mqtt_list_saved_topic_configs( + state: State<'_, Arc>, + connection_id: String, +) -> Result, String> { + let client = get_mqtt_client(&state, &connection_id).await?; + Ok(client.desired_topic_configs().await) +} + /// 获取 topic 树结构 #[tauri::command] pub async fn mqtt_get_topic_tree( diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 29e8c628bd..25917c6fec 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -2063,12 +2063,18 @@ pub fn run() { #[cfg(feature = "mq-admin")] commands::mqtt_cmd::mqtt_subscribe, #[cfg(feature = "mq-admin")] + commands::mqtt_cmd::mqtt_save_topic_config, + #[cfg(feature = "mq-admin")] commands::mqtt_cmd::mqtt_unsubscribe, #[cfg(feature = "mq-admin")] + commands::mqtt_cmd::mqtt_delete_topic_config, + #[cfg(feature = "mq-admin")] commands::mqtt_cmd::mqtt_publish, #[cfg(feature = "mq-admin")] commands::mqtt_cmd::mqtt_list_topics, #[cfg(feature = "mq-admin")] + commands::mqtt_cmd::mqtt_list_saved_topic_configs, + #[cfg(feature = "mq-admin")] commands::mqtt_cmd::mqtt_get_topic_tree, #[cfg(feature = "mq-admin")] commands::mqtt_cmd::mqtt_get_messages,