From 00212eaa8ef610073df9dfc589c234252e7bf239 Mon Sep 17 00:00:00 2001
From: YuShifan <894402575bt@gmail.com>
Date: Wed, 16 Oct 2024 18:22:14 +0800
Subject: [PATCH 1/6] feat(desktop): sync tree data to message table

---
 src/utils/messageQueue.ts      | 66 ++++++++++++++++++++++++++++++++++
 src/utils/topicTree.ts         |  1 -
 src/views/viewer/TopicTree.vue | 31 ++++++++++++----
 3 files changed, 91 insertions(+), 7 deletions(-)
 create mode 100644 src/utils/messageQueue.ts

diff --git a/src/utils/messageQueue.ts b/src/utils/messageQueue.ts
new file mode 100644
index 000000000..6fb764f40
--- /dev/null
+++ b/src/utils/messageQueue.ts
@@ -0,0 +1,66 @@
+import { Subject, timer, Observable } from 'rxjs'
+import { buffer, filter, mergeMap, share } from 'rxjs/operators'
+import { IPublishPacket } from 'mqtt-packet/types'
+import { getNowDate } from '@/utils/time'
+import { getMessageId } from '@/utils/idGenerator'
+
+export class MessageQueue {
+  private messageSubject = new Subject<{ packet: IPublishPacket; id: string }>()
+  private messageObservable: Observable<{ messages: MessageModel[]; connectionId: string }>
+  private processedMessages = new Set<string>()
+
+  constructor(private bufferTime: number = 1000, private maxBufferSize: number = 100) {
+    this.messageObservable = this.initMessageProcessing()
+  }
+
+  queueMessage(packet: IPublishPacket, id: string) {
+    this.messageSubject.next({ packet, id })
+  }
+
+  public getMessageObservable(): Observable<{ messages: MessageModel[]; connectionId: string }> {
+    return this.messageObservable
+  }
+
+  private initMessageProcessing(): Observable<{ messages: MessageModel[]; connectionId: string }> {
+    return this.messageSubject.pipe(
+      buffer(timer(0, this.bufferTime)),
+      filter((messages) => messages.length > 0),
+      mergeMap((messages) => {
+        const groupedMessages = new Map<string, MessageModel[]>()
+
+        messages.slice(0, this.maxBufferSize).forEach(({ packet, id }) => {
+          const messageKey = `${id}-${packet.topic}-${packet.payload}`
+          if (!this.processedMessages.has(messageKey)) {
+            if (!groupedMessages.has(id)) {
+              groupedMessages.set(id, [])
+            }
+            groupedMessages.get(id)!.push({
+              id: getMessageId(),
+              topic: packet.topic,
+              payload: packet.payload.toString(),
+              qos: packet.qos,
+              retain: packet.retain,
+              out: false,
+              createAt: getNowDate(),
+              properties: packet.properties,
+            })
+            this.processedMessages.add(messageKey)
+          }
+        })
+
+        if (this.processedMessages.size > 10000) {
+          const oldestEntries = Array.from(this.processedMessages).slice(0, 5000)
+          oldestEntries.forEach((entry) => this.processedMessages.delete(entry))
+        }
+
+        return Array.from(groupedMessages.entries()).map(([connectionId, messages]) => ({
+          messages,
+          connectionId,
+        }))
+      }),
+      share(),
+    )
+  }
+}
+
+export const messageQueue = new MessageQueue()
diff --git a/src/utils/topicTree.ts b/src/utils/topicTree.ts
index bebcc54a4..71574b2a0 100644
--- a/src/utils/topicTree.ts
+++ b/src/utils/topicTree.ts
@@ -11,7 +11,6 @@ import time from './time'
  * @param {ConnectionModel} rawData.connectionInfo - The connection information.
  * @returns {TopicTreeData[]} The updated topic tree data structure.
  */
-
 export function updateTopicTreeData(
   currentTree: TopicTreeData[],
   rawData: {
diff --git a/src/views/viewer/TopicTree.vue b/src/views/viewer/TopicTree.vue
index 2b8c86df0..345fd9422 100644
--- a/src/views/viewer/TopicTree.vue
+++ b/src/views/viewer/TopicTree.vue
@@ -23,6 +23,9 @@ import TreeView from '@/components/widgets/TreeView.vue'
 import { updateTopicTreeData } from '@/utils/topicTree'
 import { IPublishPacket } from 'mqtt-packet/types'
 import TreeNodeInfo from '@/components/widgets/TreeNodeInfo.vue'
+import { ignoreQoS0Message } from '@/utils/mqttUtils'
+import { messageQueue } from '@/utils/messageQueue'
+import useServices from '@/database/useServices'
 
 @Component({
   components: {
@@ -40,19 +43,35 @@ export default class TopicTree extends Vue {
       packet,
       connectionInfo,
     })
+    this.queueMessage(packet, connectionInfo.id as string)
   }
 
-  created() {
-    globalEventBus.on('packetReceive', this.handlePacketReceive)
-  }
-
-  beforeDestroy() {
-    globalEventBus.off('packetReceive', this.handlePacketReceive)
+  private queueMessage(packet: IPublishPacket, id: string) {
+    if (packet.cmd !== 'publish' || ignoreQoS0Message(packet.qos)) return
+    messageQueue.queueMessage(packet, id)
   }
 
   private handleNodeClick(data: TopicTreeData) {
     this.selectedNode = data
   }
+
+  private created() {
+    globalEventBus.on('packetReceive', this.handlePacketReceive)
+    const { messageService } = useServices()
+    messageQueue.getMessageObservable().subscribe(async ({ messages, connectionId }) => {
+      if (messages.length === 0) return
+      try {
+        await messageService.importMsgsToConnection(messages, connectionId)
+        this.$log.info(`Topic Tree: Processed ${messages.length} messages for connection ${connectionId}`)
+      } catch (error) {
+        this.$log.error(`Topic Tree: Error processing messages: ${(error as Error).toString()}`)
+      }
+    })
+  }
+
+  private beforeDestroy() {
+    globalEventBus.off('packetReceive', this.handlePacketReceive)
+  }
 }
 </script>
 

From c7ba20a7ff081e449cad1178102504048fccc22a Mon Sep 17 00:00:00 2001
From: YuShifan <894402575bt@gmail.com>
Date: Thu, 17 Oct 2024 01:04:44 +0800
Subject: [PATCH 2/6] feat(desktop): update mqtt message queue

---
 src/types/global.d.ts          |  7 +++-
 src/utils/messageQueue.ts      | 66 -------------------------------
 src/utils/mqttMessageQueue.ts  | 71 ++++++++++++++++++++++++++++++++++
 src/views/viewer/TopicTree.vue | 43 ++++++++++++++++----
 4 files changed, 113 insertions(+), 74 deletions(-)
 delete mode 100644 src/utils/messageQueue.ts
 create mode 100644 src/utils/mqttMessageQueue.ts

diff --git a/src/types/global.d.ts b/src/types/global.d.ts
index 34a862998..5e12456b7 100644
--- a/src/types/global.d.ts
+++ b/src/types/global.d.ts
@@ -1,6 +1,6 @@
 import Vue from 'vue'
 import { TranslateResult } from 'vue-i18n'
-import { MqttClient } from 'mqtt'
+import { IPublishPacket, MqttClient } from 'mqtt'
 
 declare global {
   type $TSFixed = any
@@ -460,4 +460,9 @@ declare global {
     connectionInfo?: ConnectionModel
     children?: TopicTreeData[]
   }
+  interface QueuedMessage {
+    connectionId: string
+    packet: IPublishPacket
+    timestamp: number
+  }
 }
diff --git a/src/utils/messageQueue.ts b/src/utils/messageQueue.ts
deleted file mode 100644
index 6fb764f40..000000000
--- a/src/utils/messageQueue.ts
+++ /dev/null
@@ -1,66 +0,0 @@
-import { Subject, timer, Observable } from 'rxjs'
-import { buffer, filter, mergeMap, share } from 'rxjs/operators'
-import { IPublishPacket } from 'mqtt-packet/types'
-import { getNowDate } from '@/utils/time'
-import { getMessageId } from '@/utils/idGenerator'
-
-export class MessageQueue {
-  private messageSubject = new Subject<{ packet: IPublishPacket; id: string }>()
-  private messageObservable: Observable<{ messages: MessageModel[]; connectionId: string }>
-  private processedMessages = new Set<string>()
-
-  constructor(private bufferTime: number = 1000, private maxBufferSize: number = 100) {
-    this.messageObservable = this.initMessageProcessing()
-  }
-
-  queueMessage(packet: IPublishPacket, id: string) {
-    this.messageSubject.next({ packet, id })
-  }
-
-  public getMessageObservable(): Observable<{ messages: MessageModel[]; connectionId: string }> {
-    return this.messageObservable
-  }
-
-  private initMessageProcessing(): Observable<{ messages: MessageModel[]; connectionId: string }> {
-    return this.messageSubject.pipe(
-      buffer(timer(0, this.bufferTime)),
-      filter((messages) => messages.length > 0),
-      mergeMap((messages) => {
-        const groupedMessages = new Map<string, MessageModel[]>()
-
-        messages.slice(0, this.maxBufferSize).forEach(({ packet, id }) => {
-          const messageKey = `${id}-${packet.topic}-${packet.payload}`
-          if (!this.processedMessages.has(messageKey)) {
-            if (!groupedMessages.has(id)) {
-              groupedMessages.set(id, [])
-            }
-            groupedMessages.get(id)!.push({
-              id: getMessageId(),
-              topic: packet.topic,
-              payload: packet.payload.toString(),
-              qos: packet.qos,
-              retain: packet.retain,
-              out: false,
-              createAt: getNowDate(),
-              properties: packet.properties,
-            })
-            this.processedMessages.add(messageKey)
-          }
-        })
-
-        if (this.processedMessages.size > 10000) {
-          const oldestEntries = Array.from(this.processedMessages).slice(0, 5000)
-          oldestEntries.forEach((entry) => this.processedMessages.delete(entry))
-        }
-
-        return Array.from(groupedMessages.entries()).map(([connectionId, messages]) => ({
-          messages,
-          connectionId,
-        }))
-      }),
-      share(),
-    )
-  }
-}
-
-export const messageQueue = new MessageQueue()
diff --git a/src/utils/mqttMessageQueue.ts b/src/utils/mqttMessageQueue.ts
new file mode 100644
index 000000000..d392d597b
--- /dev/null
+++ b/src/utils/mqttMessageQueue.ts
@@ -0,0 +1,71 @@
+import { Subject, Observable } from 'rxjs'
+import { bufferTime, filter, mergeMap, share } from 'rxjs/operators'
+import { IPublishPacket } from 'mqtt-packet/types'
+
+/**
+ * MessageQueue class for handling MQTT message buffering and processing.
+ *
+ * This class provides functionality to queue MQTT messages, buffer them for a specified time,
+ * and process them in batches. It uses RxJS observables for efficient message handling.
+ */
+export class MessageQueue {
+  private messageSubject = new Subject<QueuedMessage>()
+  private messageObservable: Observable<{ messages: QueuedMessage[]; connectionId: string }>
+
+  /**
+   * Creates a new MessageQueue instance.
+   * @param bufferTime The time in milliseconds to buffer messages before processing. Default is 500ms.
+   */
+  constructor(private bufferTime: number = 500) {
+    this.messageObservable = this.initMessageProcessing()
+  }
+
+  /**
+   * Queues a new MQTT message.
+   * @param packet The MQTT publish packet to queue.
+   * @param connectionId The ID of the connection associated with this message.
+   */
+  queueMessage(packet: IPublishPacket, connectionId: string) {
+    this.messageSubject.next({
+      connectionId,
+      packet,
+      timestamp: Date.now(),
+    })
+  }
+
+  /**
+   * Returns the observable for processed message batches.
+   * @returns An Observable that emits batches of messages grouped by connectionId.
+   */
+  getMessageObservable(): Observable<{ messages: QueuedMessage[]; connectionId: string }> {
+    return this.messageObservable
+  }
+
+  /**
+   * Initializes the message processing pipeline.
+   * @returns An Observable that emits batches of messages grouped by connectionId.
+   */
+  private initMessageProcessing(): Observable<{ messages: QueuedMessage[]; connectionId: string }> {
+    return this.messageSubject.pipe(
+      bufferTime(this.bufferTime),
+      filter((messages) => messages.length > 0),
+      mergeMap((messages) => {
+        const groupedMessages = messages.reduce((acc, message) => {
+          if (!acc[message.connectionId]) {
+            acc[message.connectionId] = []
+          }
+          acc[message.connectionId].push(message)
+          return acc
+        }, {} as Record<string, QueuedMessage[]>)
+
+        return Object.entries(groupedMessages).map(([connectionId, messages]) => ({
+          messages,
+          connectionId,
+        }))
+      }),
+      share(),
+    )
+  }
+}
+
+export const messageQueue = new MessageQueue()
diff --git a/src/views/viewer/TopicTree.vue b/src/views/viewer/TopicTree.vue
index 345fd9422..d170c3c63 100644
--- a/src/views/viewer/TopicTree.vue
+++ b/src/views/viewer/TopicTree.vue
@@ -24,8 +24,11 @@ import { updateTopicTreeData } from '@/utils/topicTree'
 import { IPublishPacket } from 'mqtt-packet/types'
 import TreeNodeInfo from '@/components/widgets/TreeNodeInfo.vue'
 import { ignoreQoS0Message } from '@/utils/mqttUtils'
-import { messageQueue } from '@/utils/messageQueue'
+import { messageQueue } from '@/utils/mqttMessageQueue'
 import useServices from '@/database/useServices'
+import { getMessageId } from '@/utils/idGenerator'
+import { Subscription } from 'rxjs'
+import { getNowDate } from '@/utils/time'
 
 @Component({
   components: {
@@ -38,6 +41,12 @@ export default class TopicTree extends Vue {
 
   private selectedNode: TopicTreeData | null = null
 
+  private subscription: Subscription | null = null
+
+  private handleNodeClick(data: TopicTreeData) {
+    this.selectedNode = data
+  }
+
   private handlePacketReceive(packet: IPublishPacket, connectionInfo: ConnectionModel) {
     this.data = updateTopicTreeData(this.data, {
       packet,
@@ -51,17 +60,26 @@ export default class TopicTree extends Vue {
     messageQueue.queueMessage(packet, id)
   }
 
-  private handleNodeClick(data: TopicTreeData) {
-    this.selectedNode = data
+  private generateMessage(m: QueuedMessage): MessageModel {
+    return {
+      id: getMessageId(),
+      topic: m.packet.topic,
+      payload: m.packet.payload.toString(),
+      qos: m.packet.qos,
+      retain: m.packet.retain,
+      out: false,
+      createAt: getNowDate(),
+      properties: m.packet.properties,
+    }
   }
 
-  private created() {
-    globalEventBus.on('packetReceive', this.handlePacketReceive)
+  private async storeMessages() {
     const { messageService } = useServices()
-    messageQueue.getMessageObservable().subscribe(async ({ messages, connectionId }) => {
+    this.subscription = messageQueue.getMessageObservable().subscribe(async ({ messages, connectionId }) => {
       if (messages.length === 0) return
       try {
-        await messageService.importMsgsToConnection(messages, connectionId)
+        const processedMessages = messages.map((m) => this.generateMessage(m))
+        await messageService.importMsgsToConnection(processedMessages, connectionId)
         this.$log.info(`Topic Tree: Processed ${messages.length} messages for connection ${connectionId}`)
       } catch (error) {
         this.$log.error(`Topic Tree: Error processing messages: ${(error as Error).toString()}`)
@@ -69,8 +87,19 @@ export default class TopicTree extends Vue {
     })
   }
 
+  private created() {
+    globalEventBus.on('packetReceive', this.handlePacketReceive)
+    if (!this.subscription) {
+      this.storeMessages()
+    }
+  }
+
   private beforeDestroy() {
     globalEventBus.off('packetReceive', this.handlePacketReceive)
+    if (this.subscription) {
+      this.subscription.unsubscribe()
+      this.subscription = null
+    }
   }
 }
 </script>

From fe820fe30c5dd10fede4070cdd1b31be5f1b3a1d Mon Sep 17 00:00:00 2001
From: YuShifan <894402575bt@gmail.com>
Date: Thu, 17 Oct 2024 01:18:39 +0800
Subject: [PATCH 3/6] feat(desktop): log info improvement

---
 src/components/widgets/TreeView.vue         |  6 ++--
 src/views/connections/ConnectionsDetail.vue | 34 ++++++++++++---------
 src/views/viewer/TopicTree.vue              |  7 +++--
 3 files changed, 28 insertions(+), 19 deletions(-)

diff --git a/src/components/widgets/TreeView.vue b/src/components/widgets/TreeView.vue
index a5978627b..3c6b54161 100644
--- a/src/components/widgets/TreeView.vue
+++ b/src/components/widgets/TreeView.vue
@@ -36,9 +36,9 @@
       <template #default="{ node, data }">
         <span class="custom-tree-node">
           <span class="tree-node-info">
-            <span>{{ node.label }}</span>
-            <span v-if="data.connectionInfo && data.connectionInfo.name">
-              &nbsp;- [{{ data.connectionInfo.name }}]
+            <span>
+              <span v-if="data.connectionInfo && data.connectionInfo.name">{{ data.connectionInfo.name }}@</span
+              >{{ node.label }}
             </span>
             <el-tag v-if="!checkPayloadEmpty(data.latestMessage)" size="mini" class="value-tag ml-2">
               {{ data.latestMessage }}
diff --git a/src/views/connections/ConnectionsDetail.vue b/src/views/connections/ConnectionsDetail.vue
index cb5d17e51..05948ad36 100644
--- a/src/views/connections/ConnectionsDetail.vue
+++ b/src/views/connections/ConnectionsDetail.vue
@@ -704,7 +704,7 @@ export default class ConnectionsDetail extends Vue {
       clearInterval(this.sendTimeId)
       this.sendTimeId = null
       this.$message.success(this.$tc('connections.stopTimedMessage'))
-      this.$log.info(`Timed messages sending stopped for ${this.record.name}`)
+      this.$log.info(`Timed messages sending stopped for ${this.record.name}@${this.record.host}`)
     }
   }
 
@@ -977,7 +977,7 @@ export default class ConnectionsDetail extends Vue {
     if (this.record.id) {
       const { messageService } = useServices()
       await messageService.cleanInConnection(this.record.id)
-      this.$log.info(`History connection messages were cleaned for ${this.record.name}`)
+      this.$log.info(`History connection messages were cleaned for ${this.record.name}@${this.record.host}`)
     }
   }
 
@@ -1085,7 +1085,7 @@ export default class ConnectionsDetail extends Vue {
     })
     this.setShowClientInfo(false)
     this.$emit('reload', false, false, this.handleReSubTopics)
-    this.$log.info(`Successful connection for ${this.record.name}, MQTT.js onConnect trigger`)
+    this.$log.info(`Successful connection for ${this.record.name}@${this.record.host}, MQTT.js onConnect trigger`)
   }
 
   // Error callback
@@ -1096,7 +1096,9 @@ export default class ConnectionsDetail extends Vue {
     }
     this.forceCloseTheConnection()
     this.notifyMsgWithCopilot(msgTitle)
-    this.$log.error(`Connection for ${this.record.name} failed, MQTT.js onError trigger, Error: ${error.stack}`)
+    this.$log.error(
+      `Connection for ${this.record.name}@${this.record.host} failed, MQTT.js onError trigger, Error: ${error.stack}`,
+    )
     this.$emit('reload')
   }
 
@@ -1119,7 +1121,7 @@ export default class ConnectionsDetail extends Vue {
         this.forceCloseTheConnection()
       } else {
         this.$log.info(
-          `Retrying connection for ${this.record.name}, attempt: [${this.reTryConnectTimes}/${this.maxReconnectTimes}]`,
+          `Retrying connection for ${this.record.name}@${this.record.host}, attempt: [${this.reTryConnectTimes}/${this.maxReconnectTimes}]`,
         )
         this.connectLoading = true
         this.$notify({
@@ -1135,7 +1137,7 @@ export default class ConnectionsDetail extends Vue {
 
   // Close connection callback
   private onClose() {
-    this.$log.info(`Connection for ${this.record.name} closed, MQTT.js onClose trigger`)
+    this.$log.info(`Connection for ${this.record.name}@${this.record.host} closed, MQTT.js onClose trigger`)
     this.connectLoading = false
   }
 
@@ -1221,7 +1223,9 @@ export default class ConnectionsDetail extends Vue {
     if (this.client.reconnecting && this.client.connected === false) {
       this.client.reconnecting = false
       this.forceCloseTheConnection()
-      this.$log.warn(`MQTTX force stopped reconnecting for ${this.record.name} (Client ID: ${this.record.clientId})`)
+      this.$log.warn(
+        `MQTTX force stopped reconnecting for ${this.record.name}@${this.record.host} - Client ID: ${this.record.clientId}`,
+      )
     }
   }
 
@@ -1258,14 +1262,14 @@ export default class ConnectionsDetail extends Vue {
     }
     this.setScript({ currentScript })
     this.$message.success(this.$tc('script.startScript'))
-    this.$log.info(`Script set successfully for ${this.record.name}`)
+    this.$log.info(`Script set successfully for ${this.record.name}@${this.record.host}`)
   }
 
   // Remove script
   private removeScript() {
     this.setScript({ currentScript: null })
     this.$message.success(this.$tc('script.stopScirpt'))
-    this.$log.info(`Script removed successfully from ${this.record.name}`)
+    this.$log.info(`Script removed successfully from ${this.record.name}@${this.record.host}`)
   }
 
   /*
@@ -1386,9 +1390,11 @@ export default class ConnectionsDetail extends Vue {
         const isFromActiveTopic = this.msgType !== 'publish' && this.activeTopic && isActiveTopicMessages
         const isFromNotActiveTopic = this.msgType !== 'publish' && !this.activeTopic
         if (isFromActiveTopic || isFromNotActiveTopic) {
-          let receivedLog = `Message arrived for ${this.record.name} with topic: "${topic}". Message ID: "${
-            message.id
-          }", payload: ${jsonStringify(message.payload)}. MQTT.js onMessageArrived trigger`
+          let receivedLog = `Message arrived for ${this.record.name}@${
+            this.record.host
+          } with topic: "${topic}". Message ID: "${message.id}", payload: ${jsonStringify(
+            message.payload,
+          )}. MQTT.js onMessageArrived trigger`
           this.$log.info(receivedLog)
         }
       } else {
@@ -1550,7 +1556,7 @@ export default class ConnectionsDetail extends Vue {
   private notifyTimedMessageSuccess() {
     this.$message.success(`${this.$t('connections.startTimedMessage')}${this.sendFrequency}`)
     this.$log.info(
-      `Timed message for ${this.record.name} started successfully with a frequency of ${this.sendFrequency} seconds.`,
+      `Timed message for ${this.record.name}@${this.record.host} started successfully with a frequency of ${this.sendFrequency} seconds.`,
     )
   }
 
@@ -1686,7 +1692,7 @@ export default class ConnectionsDetail extends Vue {
     this.notifyMsgWithCopilot(errorMsg)
     this.stopTimedSend()
     this.$log.error(
-      `Failed to publish message for ${this.record.name}. Error: ${errorMsg}. Stack trace: ${error.stack}`,
+      `Failed to publish message for ${this.record.name}@${this.record.host}. Error: ${errorMsg}. Stack trace: ${error.stack}`,
     )
   }
 
diff --git a/src/views/viewer/TopicTree.vue b/src/views/viewer/TopicTree.vue
index d170c3c63..61d72a065 100644
--- a/src/views/viewer/TopicTree.vue
+++ b/src/views/viewer/TopicTree.vue
@@ -52,6 +52,9 @@ export default class TopicTree extends Vue {
       packet,
       connectionInfo,
     })
+    this.$log.info(
+      `Topic Tree: Updated tree data for connection ${connectionInfo.name}@${connectionInfo.host}. Topic: ${packet.topic}, QoS: ${packet.qos}`,
+    )
     this.queueMessage(packet, connectionInfo.id as string)
   }
 
@@ -80,9 +83,9 @@ export default class TopicTree extends Vue {
       try {
         const processedMessages = messages.map((m) => this.generateMessage(m))
         await messageService.importMsgsToConnection(processedMessages, connectionId)
-        this.$log.info(`Topic Tree: Processed ${messages.length} messages for connection ${connectionId}`)
+        this.$log.info(`Topic Tree: Processed and stored ${messages.length} messages for connection ${connectionId}`)
       } catch (error) {
-        this.$log.error(`Topic Tree: Error processing messages: ${(error as Error).toString()}`)
+        this.$log.error(`Topic Tree: Error processing and storing messages: ${(error as Error).toString()}`)
       }
     })
   }

From 3e28b7946b48fa091a6f6db632e69994c9486d7e Mon Sep 17 00:00:00 2001
From: YuShifan <894402575bt@gmail.com>
Date: Thu, 17 Oct 2024 01:26:17 +0800
Subject: [PATCH 4/6] fix(desktop): fix json big int issue on tree node info

---
 src/components/widgets/TreeNodeInfo.vue | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/src/components/widgets/TreeNodeInfo.vue b/src/components/widgets/TreeNodeInfo.vue
index f7f3a0ea2..8e57ebce5 100644
--- a/src/components/widgets/TreeNodeInfo.vue
+++ b/src/components/widgets/TreeNodeInfo.vue
@@ -74,6 +74,7 @@ import { Component, Vue, Prop } from 'vue-property-decorator'
 import { Getter } from 'vuex-class'
 import { findSubTopics, findFullTopicPath, isPayloadEmpty } from '@/utils/topicTree'
 import Prism from 'prismjs'
+import { jsonStringify, jsonParse } from '@/utils/jsonUtils'
 
 @Component
 export default class TreeNodeInfo extends Vue {
@@ -89,7 +90,7 @@ export default class TreeNodeInfo extends Vue {
   get latestMessage(): string {
     const message = this.node.latestMessage || ''
     if (this.payloadFormat === 'json') {
-      return JSON.stringify(JSON.parse(message.toString()), null, 2)
+      return jsonStringify(jsonParse(message.toString()), null, 2)
     }
     return message.toString()
   }

From 47f5e3c8a729bddc7a7e72c8a0f8c5d3a5e6dd45 Mon Sep 17 00:00:00 2001
From: YuShifan <894402575bt@gmail.com>
Date: Thu, 17 Oct 2024 10:44:25 +0800
Subject: [PATCH 5/6] test(desktop): add message queue test case

---
 src/views/viewer/TopicTree.vue            |   7 +-
 tests/unit/utils/mqttMessageQueue.spec.ts | 110 ++++++++++++++++++++++
 2 files changed, 115 insertions(+), 2 deletions(-)
 create mode 100644 tests/unit/utils/mqttMessageQueue.spec.ts

diff --git a/src/views/viewer/TopicTree.vue b/src/views/viewer/TopicTree.vue
index 61d72a065..39632eafd 100644
--- a/src/views/viewer/TopicTree.vue
+++ b/src/views/viewer/TopicTree.vue
@@ -52,10 +52,13 @@ export default class TopicTree extends Vue {
       packet,
       connectionInfo,
     })
+    this.queueMessage(packet, connectionInfo.id as string)
+    if (this.data.length === 0) return
     this.$log.info(
-      `Topic Tree: Updated tree data for connection ${connectionInfo.name}@${connectionInfo.host}. Topic: ${packet.topic}, QoS: ${packet.qos}`,
+      `Topic Tree: Updated tree data for connection ${connectionInfo.name}@${connectionInfo.host}. Topic: ${
+        packet.topic
+      }, QoS: ${packet.qos}${packet.payload ? `, Payload: ${packet.payload.toString()}` : ''}`,
     )
-    this.queueMessage(packet, connectionInfo.id as string)
   }
 
   private queueMessage(packet: IPublishPacket, id: string) {
diff --git a/tests/unit/utils/mqttMessageQueue.spec.ts b/tests/unit/utils/mqttMessageQueue.spec.ts
new file mode 100644
index 000000000..39e9b0a61
--- /dev/null
+++ b/tests/unit/utils/mqttMessageQueue.spec.ts
@@ -0,0 +1,110 @@
+import { expect } from 'chai'
+import { MessageQueue } from '@/utils/mqttMessageQueue'
+import { IPublishPacket } from 'mqtt-packet'
+import { take, toArray } from 'rxjs/operators'
+
+describe('MessageQueue', () => {
+  let messageQueue: MessageQueue
+
+  beforeEach(() => {
+    messageQueue = new MessageQueue(100)
+  })
+
+  it('should buffer and group messages by connectionId', (done) => {
+    const mockPacket1: IPublishPacket = {
+      cmd: 'publish',
+      topic: 'test/topic1',
+      payload: Buffer.from('message1'),
+      qos: 0,
+      dup: false,
+      retain: false,
+    }
+    const mockPacket2: IPublishPacket = {
+      cmd: 'publish',
+      topic: 'test/topic2',
+      payload: Buffer.from('message2'),
+      qos: 0,
+      dup: false,
+      retain: false,
+    }
+    const mockPacket3: IPublishPacket = {
+      cmd: 'publish',
+      topic: 'test/topic3',
+      payload: Buffer.from('message3'),
+      qos: 0,
+      dup: false,
+      retain: false,
+    }
+    messageQueue
+      .getMessageObservable()
+      .pipe(take(2), toArray())
+      .subscribe({
+        next: (results) => {
+          expect(results).to.have.length(2)
+
+          expect(results[0].connectionId).to.equal('connection1')
+          expect(results[0].messages).to.have.length(2)
+          expect(results[0].messages[0].packet).to.deep.equal(mockPacket1)
+          expect(results[0].messages[1].packet).to.deep.equal(mockPacket2)
+
+          expect(results[1].connectionId).to.equal('connection2')
+          expect(results[1].messages).to.have.length(1)
+          expect(results[1].messages[0].packet).to.deep.equal(mockPacket3)
+        },
+        complete: () => {
+          done()
+        },
+        error: (err) => {
+          done(err)
+        },
+      })
+
+    messageQueue.queueMessage(mockPacket1, 'connection1')
+    messageQueue.queueMessage(mockPacket2, 'connection1')
+    messageQueue.queueMessage(mockPacket3, 'connection2')
+  })
+
+  it('should not emit when no messages are queued', (done) => {
+    let emitted = false
+    messageQueue.getMessageObservable().subscribe({
+      next: () => {
+        emitted = true
+      },
+    })
+
+    setTimeout(() => {
+      expect(emitted).to.be.false
+      done()
+    }, 200)
+  })
+
+  it('should use the specified buffer time', (done) => {
+    const bufferTime = 300
+    const customMessageQueue = new MessageQueue(bufferTime)
+    const mockPacket: IPublishPacket = {
+      cmd: 'publish',
+      topic: 'test/topic',
+      payload: Buffer.from('message'),
+      qos: 0,
+      dup: false,
+      retain: false,
+    }
+
+    const start = Date.now()
+    customMessageQueue
+      .getMessageObservable()
+      .pipe(take(1))
+      .subscribe({
+        next: () => {
+          const elapsed = Date.now() - start
+          expect(elapsed).to.be.at.least(bufferTime)
+          done()
+        },
+        error: (err) => {
+          done(err)
+        },
+      })
+
+    customMessageQueue.queueMessage(mockPacket, 'connection1')
+  })
+})

From 7c26eb05b627e5eaadfa788034020c4e5b197117 Mon Sep 17 00:00:00 2001
From: YuShifan <894402575bt@gmail.com>
Date: Thu, 17 Oct 2024 10:56:18 +0800
Subject: [PATCH 6/6] refactor(desktop): tree data -> tree node

---
 src/components/widgets/TreeNodeInfo.vue |  8 +++---
 src/components/widgets/TreeView.vue     | 14 +++++-----
 src/types/global.d.ts                   |  4 +--
 src/utils/topicTree.ts                  | 24 ++++++++---------
 src/views/viewer/TopicTree.vue          | 10 +++----
 tests/unit/utils/topicTree.spec.ts      | 36 ++++++++++++-------------
 6 files changed, 48 insertions(+), 48 deletions(-)

diff --git a/src/components/widgets/TreeNodeInfo.vue b/src/components/widgets/TreeNodeInfo.vue
index 8e57ebce5..dfb2ef401 100644
--- a/src/components/widgets/TreeNodeInfo.vue
+++ b/src/components/widgets/TreeNodeInfo.vue
@@ -78,8 +78,8 @@ import { jsonStringify, jsonParse } from '@/utils/jsonUtils'
 
 @Component
 export default class TreeNodeInfo extends Vue {
-  @Prop() private node!: TopicTreeData
-  @Prop() private treeData!: TopicTreeData[]
+  @Prop() private node!: TopicTreeNode
+  @Prop() private treeData!: TopicTreeNode[]
 
   @Getter('currentTheme') private currentTheme!: Theme
 
@@ -128,11 +128,11 @@ export default class TreeNodeInfo extends Vue {
     return fullHost
   }
 
-  private getSubTopics(node: TopicTreeData): string[] {
+  private getSubTopics(node: TopicTreeNode): string[] {
     return findSubTopics(node)
   }
 
-  private getFullTopicPath(node: TopicTreeData): string {
+  private getFullTopicPath(node: TopicTreeNode): string {
     const fullPath = findFullTopicPath(this.treeData, node.id)
     if (!fullPath) return node.label
     return fullPath
diff --git a/src/components/widgets/TreeView.vue b/src/components/widgets/TreeView.vue
index 3c6b54161..612c2b8b3 100644
--- a/src/components/widgets/TreeView.vue
+++ b/src/components/widgets/TreeView.vue
@@ -62,7 +62,7 @@ import { getAllIDs, isPayloadEmpty } from '@/utils/topicTree'
 
 @Component
 export default class TreeView extends Vue {
-  @Prop({ default: () => [] }) public data!: TopicTreeData[]
+  @Prop({ default: () => [] }) public data!: TopicTreeNode[]
 
   @Getter('currentTheme') private currentTheme!: Theme
 
@@ -95,26 +95,26 @@ export default class TreeView extends Vue {
     label: 'label',
   }
 
-  private filterNode(value: string, data: TopicTreeData) {
+  private filterNode(value: string, data: TopicTreeNode) {
     if (!value) return true
     return data.label.toLowerCase().indexOf(value.toLowerCase()) !== -1
   }
 
-  private handleNodeClick(data: TopicTreeData) {
+  private handleNodeClick(data: TopicTreeNode) {
     this.$emit('node-click', data)
   }
 
-  private handleNodeExpand(data: TopicTreeData) {
+  private handleNodeExpand(data: TopicTreeNode) {
     if (!this.expandedKeys.includes(data.id)) {
       this.expandedKeys.push(data.id)
     }
   }
 
-  private handleNodeCollapse(data: TopicTreeData) {
+  private handleNodeCollapse(data: TopicTreeNode) {
     this.removeExpandedKeysRecursively(data)
   }
 
-  private removeExpandedKeysRecursively(node: TopicTreeData) {
+  private removeExpandedKeysRecursively(node: TopicTreeNode) {
     const index = this.expandedKeys.indexOf(node.id)
     if (index > -1) {
       this.expandedKeys.splice(index, 1)
@@ -129,7 +129,7 @@ export default class TreeView extends Vue {
   }
 
   private collapseAll() {
-    const collapseNode = (node: TopicTreeData) => {
+    const collapseNode = (node: TopicTreeNode) => {
       if (node.children && node.children.length > 0) {
         node.children.forEach(collapseNode)
       }
diff --git a/src/types/global.d.ts b/src/types/global.d.ts
index 5e12456b7..b3be44f02 100644
--- a/src/types/global.d.ts
+++ b/src/types/global.d.ts
@@ -448,7 +448,7 @@ declare global {
 
   type LogLevel = 'debug' | 'info' | 'warn' | 'error'
 
-  interface TopicTreeData {
+  interface TopicTreeNode {
     id: string
     label: string
     qos?: QoS
@@ -458,7 +458,7 @@ declare global {
     messageCount: number
     subTopicCount: number
     connectionInfo?: ConnectionModel
-    children?: TopicTreeData[]
+    children?: TopicTreeNode[]
   }
   interface QueuedMessage {
     connectionId: string
diff --git a/src/utils/topicTree.ts b/src/utils/topicTree.ts
index 71574b2a0..a8afec9e5 100644
--- a/src/utils/topicTree.ts
+++ b/src/utils/topicTree.ts
@@ -5,19 +5,19 @@ import time from './time'
 /**
  * Updates the topic tree data structure based on received MQTT packets.
  *
- * @param {TopicTreeData[]} currentTree - The current state of the topic tree.
+ * @param {TopicTreeNode[]} currentTree - The current state of the topic tree.
  * @param {Object} rawData - The raw data containing the MQTT packet and connection information.
  * @param {IPublishPacket} rawData.packet - The MQTT publish packet.
  * @param {ConnectionModel} rawData.connectionInfo - The connection information.
- * @returns {TopicTreeData[]} The updated topic tree data structure.
+ * @returns {TopicTreeNode[]} The updated topic tree data structure.
  */
-export function updateTopicTreeData(
-  currentTree: TopicTreeData[],
+export function updateTopicTreeNode(
+  currentTree: TopicTreeNode[],
   rawData: {
     packet: IPublishPacket
     connectionInfo: ConnectionModel
   },
-): TopicTreeData[] {
+): TopicTreeNode[] {
   const { packet, connectionInfo } = rawData
   if (packet.cmd !== 'publish') {
     return currentTree
@@ -51,7 +51,7 @@ export function updateTopicTreeData(
   for (let i = 0; i < topicLevels.length; i++) {
     const level = topicLevels[i]
     const childIndex = currentNode.children?.findIndex((n) => n.label === level) ?? -1
-    let childNode: TopicTreeData
+    let childNode: TopicTreeNode
 
     if (childIndex === -1) {
       currentId = `${currentId}-${currentNode.children?.length ?? 0 + 1}`
@@ -99,7 +99,7 @@ export function updateTopicTreeData(
  * @param node - The current node in the topic tree to update.
  * @returns The total number of subtopics for the current node and its children.
  */
-export function updateSubTopicCounts(node: TopicTreeData): number {
+export function updateSubTopicCounts(node: TopicTreeNode): number {
   if (!node.children || node.children.length === 0) {
     node.subTopicCount = 0
     return 1
@@ -120,7 +120,7 @@ export function updateSubTopicCounts(node: TopicTreeData): number {
  * @param isRoot - A boolean indicating whether the current node is the root node.
  * @returns An array of strings representing the subtopics.
  */
-export function findSubTopics(node: TopicTreeData, isRoot: boolean = true): string[] {
+export function findSubTopics(node: TopicTreeNode, isRoot: boolean = true): string[] {
   let subTopics: string[] = []
   if (!isRoot && node.label) {
     subTopics.push(node.label)
@@ -140,8 +140,8 @@ export function findSubTopics(node: TopicTreeData, isRoot: boolean = true): stri
  * @param targetId - The ID of the node to find the full path for.
  * @returns The full topic path as a string, or null if the node is not found.
  */
-export function findFullTopicPath(treeData: TopicTreeData[], targetId: string): string | null {
-  function findPath(node: TopicTreeData, currentPath: string[]): string[] | null {
+export function findFullTopicPath(treeData: TopicTreeNode[], targetId: string): string | null {
+  function findPath(node: TopicTreeNode, currentPath: string[]): string[] | null {
     if (node.id === targetId) {
       return currentPath
     }
@@ -169,10 +169,10 @@ export function findFullTopicPath(treeData: TopicTreeData[], targetId: string):
 /**
  * Retrieves all IDs from the given topic tree nodes and their children.
  *
- * @param nodes - An array of TopicTreeData representing the topic tree nodes.
+ * @param nodes - An array of TopicTreeNode representing the topic tree nodes.
  * @returns An array of strings containing all IDs from the nodes and their children.
  */
-export function getAllIDs(nodes: TopicTreeData[]): string[] {
+export function getAllIDs(nodes: TopicTreeNode[]): string[] {
   let ids: string[] = []
   for (const node of nodes) {
     ids.push(node.id)
diff --git a/src/views/viewer/TopicTree.vue b/src/views/viewer/TopicTree.vue
index 39632eafd..863025761 100644
--- a/src/views/viewer/TopicTree.vue
+++ b/src/views/viewer/TopicTree.vue
@@ -20,7 +20,7 @@
 import { Component, Vue } from 'vue-property-decorator'
 import { globalEventBus } from '@/utils/globalEventBus'
 import TreeView from '@/components/widgets/TreeView.vue'
-import { updateTopicTreeData } from '@/utils/topicTree'
+import { updateTopicTreeNode } from '@/utils/topicTree'
 import { IPublishPacket } from 'mqtt-packet/types'
 import TreeNodeInfo from '@/components/widgets/TreeNodeInfo.vue'
 import { ignoreQoS0Message } from '@/utils/mqttUtils'
@@ -37,18 +37,18 @@ import { getNowDate } from '@/utils/time'
   },
 })
 export default class TopicTree extends Vue {
-  private data: TopicTreeData[] = []
+  private data: TopicTreeNode[] = []
 
-  private selectedNode: TopicTreeData | null = null
+  private selectedNode: TopicTreeNode | null = null
 
   private subscription: Subscription | null = null
 
-  private handleNodeClick(data: TopicTreeData) {
+  private handleNodeClick(data: TopicTreeNode) {
     this.selectedNode = data
   }
 
   private handlePacketReceive(packet: IPublishPacket, connectionInfo: ConnectionModel) {
-    this.data = updateTopicTreeData(this.data, {
+    this.data = updateTopicTreeNode(this.data, {
       packet,
       connectionInfo,
     })
diff --git a/tests/unit/utils/topicTree.spec.ts b/tests/unit/utils/topicTree.spec.ts
index f856c0c6c..ed8037ac4 100644
--- a/tests/unit/utils/topicTree.spec.ts
+++ b/tests/unit/utils/topicTree.spec.ts
@@ -1,6 +1,6 @@
 import { expect } from 'chai'
 import {
-  updateTopicTreeData,
+  updateTopicTreeNode,
   updateSubTopicCounts,
   findSubTopics,
   findFullTopicPath,
@@ -45,7 +45,7 @@ const connectionInfo: ConnectionModel = {
 
 describe('Topic Tree Functions', () => {
   it('should update topic tree data with IDs', () => {
-    const currentTree: TopicTreeData[] = []
+    const currentTree: TopicTreeNode[] = []
     const packet1: IPublishPacket = {
       cmd: 'publish',
       topic: 'sensors/temperature/living-room',
@@ -63,8 +63,8 @@ describe('Topic Tree Functions', () => {
       retain: false,
     }
 
-    let updatedTree = updateTopicTreeData(currentTree, { packet: packet1, connectionInfo })
-    updatedTree = updateTopicTreeData(updatedTree, { packet: packet2, connectionInfo })
+    let updatedTree = updateTopicTreeNode(currentTree, { packet: packet1, connectionInfo })
+    updatedTree = updateTopicTreeNode(updatedTree, { packet: packet2, connectionInfo })
 
     expect(updatedTree).to.have.lengthOf(1)
     expect(updatedTree[0].id).to.equal('test_connection_id')
@@ -84,7 +84,7 @@ describe('Topic Tree Functions', () => {
   })
 
   it('should update existing tree node and add new siblings', () => {
-    const currentTree: TopicTreeData[] = [
+    const currentTree: TopicTreeNode[] = [
       {
         id: 'test_connection_id',
         label: 'broker.emqx.io',
@@ -112,8 +112,8 @@ describe('Topic Tree Functions', () => {
       retain: false,
     }
 
-    let updatedTree = updateTopicTreeData(currentTree, { packet: packet1, connectionInfo })
-    updatedTree = updateTopicTreeData(updatedTree, { packet: packet2, connectionInfo })
+    let updatedTree = updateTopicTreeNode(currentTree, { packet: packet1, connectionInfo })
+    updatedTree = updateTopicTreeNode(updatedTree, { packet: packet2, connectionInfo })
 
     expect(updatedTree[0].messageCount).to.equal(2)
     expect(updatedTree[0].children?.[0].id).to.equal('test_connection_id-0')
@@ -128,7 +128,7 @@ describe('Topic Tree Functions', () => {
   })
 
   it('should update sub topic counts', () => {
-    const tree: TopicTreeData = {
+    const tree: TopicTreeNode = {
       id: 'root',
       label: 'root',
       latestMessage: '',
@@ -173,7 +173,7 @@ describe('Topic Tree Functions', () => {
   })
 
   it('should find sub topics', () => {
-    const tree: TopicTreeData = {
+    const tree: TopicTreeNode = {
       id: 'root',
       label: 'root',
       latestMessage: '',
@@ -199,7 +199,7 @@ describe('Topic Tree Functions', () => {
   })
 
   it('should find full topic path', () => {
-    const tree: TopicTreeData[] = [
+    const tree: TopicTreeNode[] = [
       {
         id: 'test_connection_id',
         label: 'broker.emqx.io',
@@ -243,7 +243,7 @@ describe('Topic Tree Functions', () => {
   })
 
   it('should get all IDs', () => {
-    const tree: TopicTreeData[] = [
+    const tree: TopicTreeNode[] = [
       {
         id: 'root',
         label: 'root',
@@ -278,7 +278,7 @@ describe('Topic Tree Functions', () => {
   })
 
   it('should get all IDs for a complex tree', () => {
-    const tree: TopicTreeData[] = [
+    const tree: TopicTreeNode[] = [
       {
         id: 'test_connection_id',
         label: 'root',
@@ -344,14 +344,14 @@ describe('Topic Tree Functions', () => {
   })
 
   it('should get all IDs for empty tree', () => {
-    const tree: TopicTreeData[] = []
+    const tree: TopicTreeNode[] = []
 
     const ids = getAllIDs(tree)
     expect(ids).to.deep.equal([])
   })
 
   it('should update existing tree node', () => {
-    const currentTree: TopicTreeData[] = [
+    const currentTree: TopicTreeNode[] = [
       {
         id: 'test_connection_id',
         label: 'broker.emqx.io',
@@ -371,7 +371,7 @@ describe('Topic Tree Functions', () => {
       retain: false,
     }
 
-    const updatedTree = updateTopicTreeData(currentTree, { packet, connectionInfo })
+    const updatedTree = updateTopicTreeNode(currentTree, { packet, connectionInfo })
 
     expect(updatedTree[0].messageCount).to.equal(1)
     expect(updatedTree[0].children?.[0].id).to.equal('test_connection_id-0')
@@ -382,16 +382,16 @@ describe('Topic Tree Functions', () => {
   })
 
   it('should not update tree for non-publish packets', () => {
-    const currentTree: TopicTreeData[] = []
+    const currentTree: TopicTreeNode[] = []
     const packet = { cmd: 'subscribe' } as any
 
-    const updatedTree = updateTopicTreeData(currentTree, { packet, connectionInfo })
+    const updatedTree = updateTopicTreeNode(currentTree, { packet, connectionInfo })
 
     expect(updatedTree).to.deep.equal(currentTree)
   })
 
   it('should return null for non-existent topic path', () => {
-    const tree: TopicTreeData[] = [
+    const tree: TopicTreeNode[] = [
       {
         id: 'test_connection_id',
         label: 'broker.emqx.io',