feat(mqtt): optimize console and subscription sync

This commit is contained in:
jinmo
2026-08-08 11:56:32 +08:00
committed by GitHub
parent e49d86b101
commit bad7f6b472
22 changed files with 1814 additions and 229 deletions
@@ -1,9 +1,10 @@
<script setup lang="ts">
import { ref, computed, onMounted, onUnmounted } from "vue";
import { ref, computed, onMounted, onUnmounted, watch } from "vue";
import { useI18n } from "vue-i18n";
import { mqttGetBrokerInfo, mqttListTopics, mqttGetTopicTree, mqttGetMessages, mqttSubscribe, mqttUnsubscribe, mqttClearMessages } from "@/lib/backend/api";
import type { MqttBrokerInfo, MqttTopicNode, MqttMessage } from "@/types/mqtt";
import { mqttGetBrokerInfo, mqttListTopics, mqttListSavedTopicConfigs, mqttGetMessages, mqttSubscribe, mqttUnsubscribe, mqttSaveTopicConfig, mqttDeleteTopicConfig, mqttClearMessages } from "@/lib/backend/api";
import type { MqttBrokerInfo, MqttSavedTopic, MqttTopicNode, MqttMessage, MqttQoS } from "@/types/mqtt";
import { Button } from "@/components/ui/button";
import { Dialog, DialogContent, DialogFooter, DialogHeader, DialogTitle } from "@/components/ui/dialog";
import TopicTreeNode from "./TopicTreeNode.vue";
import MqttPublishPanel from "./MqttPublishDialog.vue";
import { decodePayload, PAYLOAD_ENCODINGS, PAYLOAD_ENCODING_LABELS, type PayloadEncoding } from "@/lib/mqtt/mqttPayloadCodec";
@@ -15,35 +16,65 @@ interface Props {
const props = defineProps<Props>();
const { t } = useI18n();
/* ========== State ========== */
const brokerInfo = ref<MqttBrokerInfo | null>(null);
const topicTree = ref<MqttTopicNode | null>(null);
const messages = ref<MqttMessage[]>([]);
const savedTopics = ref<MqttSavedTopic[]>([]);
const subscribedTopics = ref<[string, string][]>([]);
const messages = ref<MqttMessage[]>([]);
const noLocalSubscribe = ref(false);
const selectedTopic = ref<string>(props.initialTopic ?? "");
const loading = ref(true);
const error = ref<string | null>(null);
const pollingTimer = ref<ReturnType<typeof setInterval> | null>(null);
/* Payload 显示编码格式 */
const displayEncoding = ref<PayloadEncoding>("plaintext");
const topicSearch = ref("");
const showSubscriptionDialog = ref(false);
const savingSubscription = ref(false);
const formTopic = ref("");
const formQos = ref<MqttQoS>("atmostonce");
const formNoLocal = ref(false);
const formEnabled = ref(true);
const editingTopic = ref<string | null>(null);
/* ========== Computed ========== */
const connected = computed(() => brokerInfo.value?.connected ?? false);
const mqtt5 = computed(() => brokerInfo.value?.protocolVersion?.includes("5") ?? false);
function buildTopicTree(topics: MqttSavedTopic[]): MqttTopicNode {
const root: MqttTopicNode = { name: "root", fullPath: "", children: [], isLeaf: false };
for (const config of topics) {
const segments = config.topic.split("/");
let current = root;
for (let index = 0; index < segments.length; index += 1) {
const fullPath = segments.slice(0, index + 1).join("/");
let child = current.children.find((candidate) => candidate.name === segments[index]);
if (!child) {
child = { name: segments[index], fullPath, children: [], isLeaf: false };
current.children.push(child);
}
if (index === segments.length - 1) child.isLeaf = true;
current = child;
}
}
const sort = (node: MqttTopicNode) => {
node.children.sort((left, right) => left.name.localeCompare(right.name));
node.children.forEach(sort);
};
sort(root);
return root;
}
const topicTree = computed(() => buildTopicTree(savedTopics.value));
/* ========== 数据刷新 ========== */
async function refreshData() {
try {
const [info, topics, tree, msgs] = await Promise.all([
const [info, active, configs, msgs] = await Promise.all([
mqttGetBrokerInfo(props.connectionId) as Promise<MqttBrokerInfo>,
mqttListTopics(props.connectionId) as Promise<[string, string][]>,
mqttGetTopicTree(props.connectionId) as Promise<MqttTopicNode>,
mqttListSavedTopicConfigs(props.connectionId) as Promise<MqttSavedTopic[]>,
mqttGetMessages(props.connectionId, selectedTopic.value || undefined, 50) as Promise<MqttMessage[]>,
]);
brokerInfo.value = info;
subscribedTopics.value = topics;
topicTree.value = tree;
subscribedTopics.value = active;
savedTopics.value = configs.map((config) => ({ ...config, enabled: config.enabled !== false, noLocal: config.noLocal === true }));
messages.value = msgs;
error.value = null;
} catch (e) {
@@ -53,49 +84,117 @@ async function refreshData() {
}
}
async function handleSubscribe(topic: string, noLocal = noLocalSubscribe.value) {
async function handleSubscribe(topic: string, qos: MqttQoS = "atmostonce", noLocal = noLocalSubscribe.value) {
try {
await mqttSubscribe(props.connectionId, topic, "atmostonce", noLocal);
await mqttSubscribe(props.connectionId, topic, qos, noLocal);
selectedTopic.value = topic;
await refreshData();
} catch (e) {
error.value = String(e);
}
}
async function handleUnsubscribe(topics: string[]) {
const subscribed = new Set(subscribedTopics.value.map(([topic]) => topic));
const targets = [...new Set(topics)].filter((topic) => subscribed.has(topic));
if (targets.length === 0) {
error.value = t("connection.mqttUnsubscribeMissing");
async function handleToggleEnabled(config: MqttSavedTopic) {
try {
if (subscribedTopics.value.some(([topic]) => topic === config.topic)) {
await mqttUnsubscribe(props.connectionId, config.topic);
} else {
await handleSubscribe(config.topic, config.qos, config.noLocal === true);
return;
}
await refreshData();
} catch (e) {
error.value = String(e);
}
}
async function handleDelete(config: MqttSavedTopic) {
if (!window.confirm(t("connection.mqttDeleteSubscriptionConfirm", { topic: config.topic }))) return;
try {
if (subscribedTopics.value.some(([topic]) => topic === config.topic)) await mqttUnsubscribe(props.connectionId, config.topic);
await mqttDeleteTopicConfig(props.connectionId, config.topic);
if (selectedTopic.value === config.topic) selectedTopic.value = "";
await refreshData();
} catch (e) {
error.value = String(e);
}
}
function openSubscriptionDialog() {
editingTopic.value = null;
formTopic.value = "";
formQos.value = "atmostonce";
formNoLocal.value = false;
formEnabled.value = true;
error.value = null;
showSubscriptionDialog.value = true;
}
function editSubscription(config: MqttSavedTopic) {
editingTopic.value = config.topic;
formTopic.value = config.topic;
formQos.value = config.qos;
formNoLocal.value = config.noLocal === true;
formEnabled.value = config.enabled !== false;
error.value = null;
showSubscriptionDialog.value = true;
}
function validTopicFilter(topic: string): boolean {
if (!topic) return false;
const segments = topic.split("/");
return segments.every((segment, index) => {
if (segment === "#") return index === segments.length - 1;
return !segment.includes("#") && (!segment.includes("+") || segment === "+");
});
}
async function saveSubscription() {
const topic = formTopic.value.trim();
if (!validTopicFilter(topic)) {
error.value = t("connection.mqttSubscriptionTopicFilterInvalid");
return;
}
if (savedTopics.value.some((config) => config.topic === topic && config.topic !== editingTopic.value)) {
error.value = t("connection.mqttSubscriptionTopicFilterDuplicate");
return;
}
if (formNoLocal.value && !mqtt5.value) {
error.value = t("connection.mqttSubscriptionNoLocalProtocolError");
return;
}
const prompt = targets.length === 1 ? t("connection.mqttUnsubscribeConfirm", { topic: targets[0] }) : t("connection.mqttUnsubscribeGroupConfirm", { count: targets.length });
if (!window.confirm(prompt)) return;
const failures: string[] = [];
for (const topic of targets) {
try {
await mqttUnsubscribe(props.connectionId, topic);
} catch (e) {
failures.push(`${topic}:${String(e)}`);
}
}
if (targets.includes(selectedTopic.value)) selectedTopic.value = "";
await refreshData();
if (failures.length > 0) {
error.value = t("connection.mqttUnsubscribeFailed", { count: failures.length, errors: failures.join("; ") });
savingSubscription.value = true;
error.value = null;
const previousTopic = editingTopic.value;
const topicChanged = previousTopic != null && previousTopic !== topic;
const previousWasActive = previousTopic != null && subscribedTopics.value.some(([current]) => current === previousTopic);
try {
// Save the new configuration first so a broker failure cannot remove the
// previous configuration or leave the edited topic without a retry path.
await mqttSaveTopicConfig(props.connectionId, topic, formQos.value, formNoLocal.value, formEnabled.value);
if (formEnabled.value) await mqttSubscribe(props.connectionId, topic, formQos.value, formNoLocal.value);
else if (!topicChanged && previousWasActive) await mqttUnsubscribe(props.connectionId, topic);
if (topicChanged && previousWasActive && previousTopic) await mqttUnsubscribe(props.connectionId, previousTopic);
if (topicChanged && previousTopic) await mqttDeleteTopicConfig(props.connectionId, previousTopic);
selectedTopic.value = topic;
showSubscriptionDialog.value = false;
await refreshData();
} catch (e) {
error.value = t("connection.mqttSubscriptionOperationFailed", { error: String(e) });
await refreshData();
} finally {
savingSubscription.value = false;
}
}
function handleTopicClick(topic: string) {
selectedTopic.value = topic;
refreshData();
void refreshData();
}
function handleMessagePublished() {
refreshData();
void refreshData();
}
async function handleClearMessages() {
@@ -107,15 +206,13 @@ async function handleClearMessages() {
}
}
/* ========== 消息轮询 ========== */
function startPolling() {
stopPolling();
pollingTimer.value = setInterval(async () => {
try {
const msgs = (await mqttGetMessages(props.connectionId, selectedTopic.value || undefined, 50)) as MqttMessage[];
messages.value = msgs;
messages.value = (await mqttGetMessages(props.connectionId, selectedTopic.value || undefined, 50)) as MqttMessage[];
} catch {
/* 忽略轮询错误 */
/* ignore polling errors */
}
}, 3000);
}
@@ -127,33 +224,36 @@ function stopPolling() {
}
}
/* ========== 格式化消息 Payload ========== */
function formatMessagePayload(msg: MqttMessage): string {
if (displayEncoding.value === "plaintext" && msg.payloadText != null) {
return msg.payloadText;
}
if (displayEncoding.value === "plaintext" && msg.payloadText != null) return msg.payloadText;
return decodePayload(msg.payloadBase64, displayEncoding.value);
}
/* ========== 生命周期 ========== */
watch(
() => props.initialTopic,
(topic) => {
if (topic !== undefined && topic !== selectedTopic.value) {
selectedTopic.value = topic;
void refreshData();
}
},
);
onMounted(async () => {
await refreshData();
startPolling();
});
onUnmounted(() => {
stopPolling();
});
onUnmounted(stopPolling);
</script>
<template>
<div class="flex flex-col h-full bg-card text-card-foreground">
<!-- Header -->
<div class="flex items-center justify-between px-4 py-2 border-b shrink-0">
<div class="flex h-full flex-col bg-card text-card-foreground">
<div class="flex shrink-0 items-center justify-between border-b px-4 py-2">
<div class="flex items-center gap-3">
<span class="text-lg font-semibold">{{ t("connection.mqttConsoleTitle") }}</span>
<span v-if="brokerInfo" class="text-sm text-muted-foreground">{{ brokerInfo.brokerUrl }}</span>
<span v-if="brokerInfo" class="px-1.5 py-0.5 rounded text-xs font-medium" :class="connected ? 'bg-green-100 text-green-700 dark:bg-green-900/30 dark:text-green-400' : 'bg-red-100 text-red-700 dark:bg-red-900/30 dark:text-red-400'">
<span v-if="brokerInfo" class="rounded px-1.5 py-0.5 text-xs font-medium" :class="connected ? 'bg-green-100 text-green-700 dark:bg-green-900/30 dark:text-green-400' : 'bg-red-100 text-red-700 dark:bg-red-900/30 dark:text-red-400'">
{{ connected ? t("connection.mqttConnected") : t("connection.mqttDisconnected") }}
</span>
</div>
@@ -163,86 +263,129 @@ onUnmounted(() => {
</div>
</div>
<!-- Error banner -->
<div v-if="error" class="px-4 py-2 bg-red-50 dark:bg-red-950/30 text-red-700 dark:text-red-400 text-sm border-b">
<div v-if="error" class="border-b bg-red-50 px-4 py-2 text-sm text-red-700 dark:bg-red-950/30 dark:text-red-400">
{{ error }}
<button class="ml-2 underline" @click="error = null">{{ t("common.close") }}</button>
</div>
<!-- Body: 双栏布局 -->
<div class="flex-1 flex min-h-0">
<!-- 左侧:主题树 -->
<div class="w-72 border-r flex flex-col shrink-0">
<div class="px-3 py-2 text-xs font-semibold text-muted-foreground uppercase border-b">{{ t("connection.mqttTopics") }}</div>
<div class="flex-1 overflow-auto p-2">
<div v-if="loading" class="text-xs text-muted-foreground p-2">{{ t("common.loading") }}</div>
<div v-else-if="!topicTree?.children?.length" class="text-xs text-muted-foreground p-2">{{ t("connection.mqttNoTopicsHint") }}</div>
<TopicTreeNode v-for="child in topicTree?.children ?? []" :key="child.fullPath" :node="child" :depth="0" :selected-topic="selectedTopic" :subscribed-topics="subscribedTopics" @select="handleTopicClick" @subscribe="handleSubscribe" @unsubscribe="handleUnsubscribe" />
<div class="flex min-h-0 flex-1">
<div class="flex w-[22rem] shrink-0 flex-col border-r">
<div class="flex items-center justify-between border-b px-3 py-2">
<span class="text-xs font-semibold uppercase text-muted-foreground">{{ t("connection.mqttSubscriptionConfig") }}</span>
<div class="flex items-center gap-1">
<Button size="sm" variant="ghost" class="h-7 px-2 text-xs" @click="refreshData">{{ t("connection.mqttRefresh") }}</Button>
<Button size="sm" class="h-7 px-2 text-xs" @click="openSubscriptionDialog">+ {{ t("connection.mqttNewSubscription") }}</Button>
</div>
</div>
<div class="border-b p-2">
<input v-model="topicSearch" class="h-8 w-full rounded border bg-transparent px-2 text-xs outline-none focus-visible:ring-1 focus-visible:ring-ring" :placeholder="t('connection.mqttSearchSubscriptionsPlaceholder')" />
</div>
<div class="min-h-0 flex-1 overflow-auto">
<div v-if="loading" class="p-3 text-xs text-muted-foreground">{{ t("common.loading") }}</div>
<div v-else-if="!savedTopics.length" class="p-3 text-xs text-muted-foreground">{{ t("connection.mqttNoSavedSubscriptions") }}</div>
<TopicTreeNode
v-for="child in topicTree.children"
:key="child.fullPath"
:node="child"
:depth="0"
:selected-topic="selectedTopic"
:subscribed-topics="subscribedTopics"
:saved-topics="savedTopics"
:filter="topicSearch"
@select="handleTopicClick"
@toggle-enabled="handleToggleEnabled"
@edit="editSubscription"
@delete="handleDelete"
/>
</div>
<!-- 快速订阅 -->
<div class="border-t p-2">
<form
class="flex gap-1"
@submit.prevent="
(e) => {
const input = (e.target as HTMLFormElement).querySelector('input');
const topic = input?.value.trim();
if (topic && input) {
handleSubscribe(topic);
input.value = '';
}
}
"
>
<input class="flex-1 h-7 px-2 text-xs rounded border bg-transparent" :placeholder="t('connection.mqttSubscribePlaceholder')" />
<label class="flex items-center gap-1 text-[10px] whitespace-nowrap"><input v-model="noLocalSubscribe" type="checkbox" />{{ t("connection.mqttNoLocal") }}</label>
<Button size="sm" variant="ghost" class="h-7 text-xs" type="submit">{{ t("connection.mqttSubscribe") }}</Button>
<form class="flex gap-1" @submit.prevent="handleSubscribe((($event.target as HTMLFormElement).querySelector('input') as HTMLInputElement)?.value.trim() || '')">
<input class="h-7 min-w-0 flex-1 rounded border bg-transparent px-2 text-xs" :placeholder="t('connection.mqttSubscribePlaceholder')" />
<Button size="sm" variant="ghost" class="h-7 px-2 text-xs" type="submit">{{ t("connection.mqttQuickSubscribe") }}</Button>
</form>
<label class="mt-1 flex items-center gap-1 text-[10px] text-muted-foreground"><input v-model="noLocalSubscribe" type="checkbox" :disabled="!mqtt5" />{{ t("connection.mqttNoLocal") }}</label>
</div>
</div>
<!-- 右侧:消息列表 + 发布面板 -->
<div class="flex-1 flex flex-col min-w-0">
<!-- 消息区域 -->
<div class="flex-1 flex flex-col min-h-0">
<div class="px-3 py-2 text-xs font-semibold text-muted-foreground uppercase border-b flex items-center justify-between shrink-0">
<div class="flex min-w-0 flex-1 flex-col">
<div class="flex min-h-0 flex-1 flex-col">
<div class="flex shrink-0 items-center justify-between border-b px-3 py-2 text-xs font-semibold uppercase text-muted-foreground">
<span v-if="selectedTopic">{{ t("connection.mqttMessagesForTopic", { topic: selectedTopic }) }}</span>
<span v-else>{{ t("connection.mqttAllMessages") }}</span>
<div class="flex items-center gap-2">
<span class="font-normal">{{ messages.length }}</span>
<!-- Payload 编码格式选择 -->
<select v-model="displayEncoding" class="h-6 px-1.5 text-[11px] rounded border bg-transparent text-muted-foreground cursor-pointer outline-none">
<option v-for="enc in PAYLOAD_ENCODINGS" :key="enc" :value="enc">
{{ PAYLOAD_ENCODING_LABELS[enc] }}
</option>
<select v-model="displayEncoding" class="h-6 rounded border bg-transparent px-1.5 text-[11px] text-muted-foreground outline-none">
<option v-for="enc in PAYLOAD_ENCODINGS" :key="enc" :value="enc">{{ PAYLOAD_ENCODING_LABELS[enc] }}</option>
</select>
</div>
</div>
<div class="flex-1 overflow-auto flex flex-col">
<div v-if="messages.length === 0" class="text-xs text-muted-foreground p-4 text-center">{{ t("connection.mqttNoMessages") }}</div>
<div class="flex min-h-0 flex-1 flex-col overflow-auto">
<div v-if="messages.length === 0" class="p-4 text-center text-xs text-muted-foreground">{{ t("connection.mqttNoMessages") }}</div>
<div
v-for="(msg, i) in messages"
:key="i"
class="px-3 py-2 border-b last:border-b-0 cursor-pointer text-xs self-start w-full"
:class="msg.direction === 'sent' ? 'bg-emerald-50/60 dark:bg-emerald-950/20 hover:bg-emerald-100/60 dark:hover:bg-emerald-950/40 ml-auto max-w-[85%] rounded-l-md' : 'hover:bg-muted/50 mr-auto max-w-[85%] rounded-r-md'"
class="w-full cursor-pointer border-b px-3 py-2 text-xs"
:class="
msg.direction === 'sent'
? 'ml-auto max-w-[85%] rounded-l-md border-r-2 border-emerald-400 bg-emerald-50/70 hover:bg-emerald-100/70 dark:border-emerald-500 dark:bg-emerald-950/30 dark:hover:bg-emerald-950/50'
: 'mr-auto max-w-[85%] rounded-r-md border-l-2 border-blue-400 bg-blue-50/70 hover:bg-blue-100/70 dark:border-blue-500 dark:bg-blue-950/30 dark:hover:bg-blue-950/50'
"
@click="handleTopicClick(msg.topic)"
>
<div class="flex items-center gap-2 mb-0.5">
<span v-if="msg.direction === 'sent'" class="text-emerald-600 dark:text-emerald-400 text-[10px] font-bold shrink-0">{{ t("connection.mqttSent") }}</span>
<span v-else class="text-blue-600 dark:text-blue-400 text-[10px] font-bold shrink-0">{{ t("connection.mqttReceived") }}</span>
<span class="font-mono text-blue-600 dark:text-blue-400 font-medium truncate">{{ msg.topic }}</span>
<span class="text-muted-foreground/60 shrink-0">QoS{{ msg.qos }}</span>
<span v-if="msg.retain" class="text-amber-600 dark:text-amber-400 text-[10px] font-medium shrink-0">{{ t("connection.mqttRetained") }}</span>
<div class="mb-0.5 flex items-center gap-2">
<span v-if="msg.direction === 'sent'" class="shrink-0 text-[10px] font-bold text-emerald-600 dark:text-emerald-400">{{ t("connection.mqttSent") }}</span>
<span v-else class="shrink-0 text-[10px] font-bold text-blue-600 dark:text-blue-400">{{ t("connection.mqttReceived") }}</span>
<span class="truncate font-mono font-medium text-blue-700 dark:text-blue-300">{{ msg.topic }}</span>
<span class="shrink-0 text-muted-foreground">QoS{{ msg.qos }}</span>
<span v-if="msg.retain" class="shrink-0 text-[10px] font-medium text-amber-600 dark:text-amber-400">{{ t("connection.mqttRetained") }}</span>
</div>
<div class="text-muted-foreground font-mono whitespace-pre-wrap break-all">{{ formatMessagePayload(msg) }}</div>
<div class="text-muted-foreground/50 mt-0.5 text-[10px]">{{ new Date(msg.receivedAtMs).toLocaleTimeString() }}</div>
<div class="break-all whitespace-pre-wrap font-mono text-muted-foreground">{{ formatMessagePayload(msg) }}</div>
<div class="mt-0.5 text-[10px] text-muted-foreground/60">{{ new Date(msg.receivedAtMs).toLocaleTimeString() }}</div>
</div>
</div>
</div>
<!-- 底部:内联发布面板 -->
<MqttPublishPanel v-if="connected" :connection-id="connectionId" :initial-topic="selectedTopic" @published="handleMessagePublished" />
</div>
</div>
<Dialog v-model:open="showSubscriptionDialog">
<DialogContent class="sm:max-w-md">
<DialogHeader>
<DialogTitle>{{ editingTopic ? t("connection.mqttEditSubscription") : t("connection.mqttNewSubscription") }}</DialogTitle>
</DialogHeader>
<div class="grid gap-3">
<label class="grid gap-1.5 text-xs font-medium">
<span>{{ t("connection.mqttTopicFilter") }}</span>
<input v-model="formTopic" class="h-8 rounded border bg-transparent px-2 font-mono text-xs outline-none focus-visible:ring-1 focus-visible:ring-ring" :placeholder="t('connection.mqttTopicFilterPlaceholder')" :disabled="savingSubscription" @keydown.enter.prevent="saveSubscription" />
<span class="text-[11px] font-normal text-muted-foreground">{{ t("connection.mqttTopicFilterHint") }}</span>
</label>
<label class="grid gap-1.5 text-xs font-medium">
<span>QoS</span>
<select v-model="formQos" class="h-8 rounded border bg-transparent px-2 text-xs" :disabled="savingSubscription">
<option value="atmostonce">QoS 0 — {{ t("connection.mqttQosAtMostOnce") }}</option>
<option value="atleastonce">QoS 1 — {{ t("connection.mqttQosAtLeastOnce") }}</option>
<option value="exactlyonce">QoS 2 — {{ t("connection.mqttQosExactlyOnce") }}</option>
</select>
</label>
<label class="flex items-center gap-2 text-xs font-medium">
<input v-model="formNoLocal" type="checkbox" :disabled="savingSubscription || !mqtt5" />
<span>{{ t("connection.mqttNoLocal") }}(No Local)</span>
<span v-if="!mqtt5" class="font-normal text-muted-foreground">{{ t("connection.mqttNoLocalMqtt5Only") }}</span>
</label>
<label class="flex items-start gap-2 text-xs font-medium">
<input v-model="formEnabled" type="checkbox" :disabled="savingSubscription" />
<span
><span>{{ t("connection.mqttEnableSubscription") }}</span
><span class="block font-normal text-muted-foreground">{{ t("connection.mqttEnableSubscriptionHint") }}</span></span
>
</label>
<p v-if="error" class="text-xs text-destructive">{{ error }}</p>
</div>
<DialogFooter>
<Button variant="ghost" :disabled="savingSubscription" @click="showSubscriptionDialog = false">{{ t("common.cancel") }}</Button>
<Button :disabled="savingSubscription" @click="saveSubscription">{{ savingSubscription ? t("connection.mqttSaving") : t("common.save") }}</Button>
</DialogFooter>
</DialogContent>
</Dialog>
</div>
</template>
@@ -1,6 +1,6 @@
<script setup lang="ts">
import type { MqttTopicNode } from "@/types/mqtt";
import { ChevronRight, ChevronDown, Trash2 } from "@lucide/vue";
import type { MqttSavedTopic, MqttTopicNode } from "@/types/mqtt";
import { ChevronRight, ChevronDown, Pencil, Power, Trash2 } from "@lucide/vue";
import { computed, ref } from "vue";
import { useI18n } from "vue-i18n";
@@ -9,32 +9,28 @@ interface Props {
depth: number;
selectedTopic: string;
subscribedTopics: [string, string][];
savedTopics: MqttSavedTopic[];
filter?: string;
}
const props = defineProps<Props>();
const { t } = useI18n();
const emit = defineEmits<{
select: [topic: string];
subscribe: [topic: string];
unsubscribe: [topics: string[]];
toggleEnabled: [config: MqttSavedTopic];
edit: [config: MqttSavedTopic];
delete: [config: MqttSavedTopic];
}>();
const expanded = ref(props.depth < 2);
function isSubscribed(topic: string): boolean {
return props.subscribedTopics.some(([t]) => t === topic);
}
function collectSubscribedTopics(node: MqttTopicNode): string[] {
const topics = node.isLeaf && isSubscribed(node.fullPath) ? [node.fullPath] : [];
for (const child of node.children ?? []) {
topics.push(...collectSubscribedTopics(child));
}
return topics;
}
const topicsToUnsubscribe = computed(() => collectSubscribedTopics(props.node));
const unsubscribeLabel = computed(() => (topicsToUnsubscribe.value.length === 1 ? t("connection.mqttUnsubscribeTopic", { topic: topicsToUnsubscribe.value[0] }) : t("connection.mqttUnsubscribeGroup", { count: topicsToUnsubscribe.value.length })));
const savedConfig = computed(() => props.savedTopics.find((config) => config.topic === props.node.fullPath));
const active = computed(() => props.subscribedTopics.some(([topic]) => topic === props.node.fullPath));
const visible = computed(() => {
const filter = props.filter?.trim().toLowerCase();
if (!filter) return true;
const matches = (node: MqttTopicNode): boolean => node.fullPath.toLowerCase().includes(filter) || (node.children ?? []).some(matches);
return matches(props.node);
});
function toggle() {
expanded.value = !expanded.value;
@@ -42,39 +38,41 @@ function toggle() {
</script>
<template>
<div>
<div v-if="visible">
<div
class="flex items-center gap-1 py-0.5 pl-2 pr-1 rounded cursor-pointer text-xs group border-l-2 border-transparent"
:class="selectedTopic === node.fullPath ? 'bg-primary/25 text-foreground font-semibold border-l-primary shadow-sm ring-1 ring-primary/20' : 'hover:bg-muted/70 text-muted-foreground'"
:style="{ paddingLeft: `${depth * 12 + 4}px` }"
@click="node.isLeaf ? emit('select', node.fullPath) : toggle()"
class="group flex min-h-8 items-center gap-1 border-b border-border/50 py-1 pr-1 text-[13px] transition-colors"
:class="selectedTopic === node.fullPath ? 'border-l-2 border-l-primary bg-primary/10 pl-1 text-primary' : 'border-l-2 border-l-transparent pl-1 text-foreground/90 hover:bg-muted/60'"
:style="{ paddingLeft: `${depth * 14 + 4}px` }"
>
<!-- Expand/collapse toggle -->
<button v-if="node.children?.length" class="shrink-0 w-4 h-4 flex items-center justify-center text-muted-foreground" @click.stop="toggle">
<ChevronRight v-if="!expanded" class="h-3 w-3" />
<ChevronDown v-else class="h-3 w-3" />
<button v-if="node.children?.length" type="button" class="flex h-5 w-5 shrink-0 items-center justify-center rounded text-muted-foreground hover:bg-muted" @click.stop="toggle">
<ChevronRight v-if="!expanded" class="h-3.5 w-3.5" />
<ChevronDown v-else class="h-3.5 w-3.5" />
</button>
<span v-else class="w-4 shrink-0" />
<span v-else class="w-5 shrink-0" />
<!-- Topic name -->
<span class="truncate flex-1 font-mono" :class="selectedTopic === node.fullPath ? 'text-primary' : node.isLeaf ? 'font-medium text-blue-600 dark:text-blue-400' : ''">
{{ node.name }}
</span>
<button type="button" class="min-w-0 flex-1 truncate text-left font-mono font-medium" :title="node.fullPath" @click="node.isLeaf ? emit('select', node.fullPath) : toggle()">{{ node.name }}<span v-if="!node.isLeaf" class="text-muted-foreground/70">/</span></button>
<!-- 分组节点可一次取消其下全部订阅 -->
<button
v-if="topicsToUnsubscribe.length"
type="button"
class="flex h-5 w-5 shrink-0 items-center justify-center rounded text-red-500 opacity-0 transition-opacity hover:bg-accent group-hover:opacity-100 focus:opacity-100"
:title="unsubscribeLabel"
:aria-label="unsubscribeLabel"
@click.stop="emit('unsubscribe', topicsToUnsubscribe)"
>
<Trash2 class="h-3 w-3" />
</button>
<template v-if="savedConfig">
<span class="shrink-0 rounded border px-1 py-0.5 text-[10px] text-muted-foreground">QoS {{ savedConfig.qos === "atmostonce" ? 0 : savedConfig.qos === "atleastonce" ? 1 : 2 }}</span>
<span v-if="savedConfig.noLocal" class="shrink-0 rounded border border-blue-400/40 px-1 py-0.5 text-[10px] text-blue-600 dark:text-blue-400" :title="t('connection.mqttNoLocalMqtt5Only')">NL</span>
<span class="hidden shrink-0 text-[10px] text-muted-foreground sm:inline">{{ active ? t("connection.mqttSubscriptionStatusActive") : savedConfig.enabled ? t("connection.mqttSubscriptionStatusPending") : t("connection.mqttSubscriptionStatusDisabled") }}</span>
<button
type="button"
class="flex h-6 w-6 shrink-0 items-center justify-center rounded text-muted-foreground hover:bg-muted hover:text-foreground"
:title="active ? t('connection.mqttSubscriptionDisable') : t('connection.mqttSubscriptionEnable')"
@click.stop="emit('toggleEnabled', savedConfig)"
>
<Power class="h-3.5 w-3.5" :class="active ? 'text-emerald-600 dark:text-emerald-400' : ''" />
</button>
<button type="button" class="flex h-6 w-6 shrink-0 items-center justify-center rounded text-muted-foreground hover:bg-muted hover:text-foreground" :title="t('connection.mqttSubscriptionEdit')" @click.stop="emit('edit', savedConfig)">
<Pencil class="h-3.5 w-3.5" />
</button>
<button type="button" class="flex h-6 w-6 shrink-0 items-center justify-center rounded text-muted-foreground hover:bg-destructive/10 hover:text-destructive" :title="t('connection.mqttSubscriptionDelete')" @click.stop="emit('delete', savedConfig)">
<Trash2 class="h-3.5 w-3.5" />
</button>
</template>
</div>
<!-- Children -->
<TopicTreeNode
v-if="expanded"
v-for="child in node.children"
@@ -83,14 +81,16 @@ function toggle() {
:depth="depth + 1"
:selected-topic="selectedTopic"
:subscribed-topics="subscribedTopics"
:saved-topics="savedTopics"
:filter="filter"
@select="emit('select', $event)"
@subscribe="emit('subscribe', $event)"
@unsubscribe="emit('unsubscribe', $event)"
@toggle-enabled="emit('toggleEnabled', $event)"
@edit="emit('edit', $event)"
@delete="emit('delete', $event)"
/>
</div>
</template>
<script lang="ts">
// Recursive component needs explicit name in non-setup script block
export default { name: "TopicTreeNode" };
</script>
+24 -6
View File
@@ -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",
+24 -6
View File
@@ -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.",
+24 -6
View File
@@ -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.",
+24 -6
View File
@@ -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 を入力します。空白を含む値にもシェル引用符は不要です。",
+24 -6
View File
@@ -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.",
+24 -6
View File
@@ -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: "分类",
+24 -6
View File
@@ -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 引號。",
+3
View File
@@ -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");
+12
View File
@@ -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");
}
@@ -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 });
}
@@ -8,7 +8,7 @@ export type SidebarActivation = "single" | "double";
const dataNodeTypes = new Set<TreeNodeType>(["table", "view", "materialized_view"]);
const documentBrowserNodeTypes = new Set<TreeNodeType>(["mongo-collection", "mongo-bucket"]);
const toggleLeafNodeTypes = new Set<TreeNodeType>(["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<TreeNodeType>(["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<TreeNodeType>(["database", "schema", "object-browser"]);
const sourceNodeTypes = new Set<TreeNodeType>(["materialized_view", "procedure", "function", "trigger", "sequence", "synonym", "package", "package-body", "type", "type-body"]);
const savedSqlNodeTypes = new Set<TreeNodeType>(["saved-sql-file"]);
+4 -19
View File
@@ -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;
+2
View File
@@ -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 {
+172
View File
@@ -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<ConnectionConfig>,
/// 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<Vec<MqttSubscriptionSyncEntry>>,
/// 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<EncryptedSecretsBlob>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct MqttSubscriptionSyncEntry {
pub connection_id: String,
pub subscriptions: Vec<MqttSubscriptionSyncTopic>,
}
#[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<Vec<MqttSubscriptionSyncEntry>, 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<MqttSubscriptionSyncTopic> = 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());
+3 -3
View File
@@ -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);
+60 -9
View File
@@ -311,6 +311,8 @@ pub struct MqttClient {
granted_subscriptions: RwLock<Vec<(String, MqttQoS)>>,
/// 需要在无会话重连后恢复的 topic 集合
desired_subscriptions: RwLock<Vec<(String, MqttQoS)>>,
/// All persisted subscription configurations, including disabled entries.
saved_topic_configs: RwLock<Vec<MqttSavedTopic>>,
no_local_topics: RwLock<HashSet<String>>,
/// 当前连接是否已收到有效 CONNACK
connected: AtomicBool,
@@ -392,15 +394,24 @@ impl MqttClient {
backend: MqttBackendClient,
shutdown_notify: Arc<Notify>,
) -> Arc<Self> {
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<MqttSavedTopic> {
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<MqttMessage> {
+8
View File
@@ -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 消息服务质量
File diff suppressed because it is too large Load Diff
+34
View File
@@ -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<AppState>>,
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<AppState>>,
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<AppState>>,
connection_id: String,
) -> Result<Vec<MqttSavedTopic>, 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(
+6
View File
@@ -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,