1

api/middleware/mqtt/mqttClient.js

@@ -8,15 +8,45 @@ class MqttSyncClient {
     constructor(brokerUrl, options = {}) {
         this.client = mqtt.connect(brokerUrl, options)
         this.isConnected = false
+        /** @type {string[]} 需在每次 connect（含重连）后向 Broker 幂等订阅的主题 */
+        this._maintainedTopics = []
+        /** 最近一次收到任意 `response` 主题消息的时间（用于超时日志关联） */
+        this.lastResponseAt = null
 
         // 使用 Map 结构优化消息监听器，按 topic 分组
         this.messageListeners = new Map(); // Map<topic, Set<listener>>
         this.globalListeners = new Set(); // 全局监听器（监听所有 topic）
 
+        const ts = () => new Date().toISOString()
+        const markDisconnected = (reason) => {
+            this.isConnected = false
+            console.warn(`[MQTT] ${ts()} 连接不可用 reason=${reason}`)
+        }
+
         this.client.on('connect', () => {
             this.isConnected = true
+            console.log(`[MQTT] ${ts()} 服务端已连接（含重连后的 connect）`)
+            this._resubscribeMaintainedTopics()
+        })
 
-            console.log('MQTT 服务端已连接')
+        this.client.on('reconnect', () => {
+            console.log(`[MQTT] ${ts()} 正在重连 Broker...`)
+        })
+
+        this.client.on('offline', () => {
+            markDisconnected('offline')
+        })
+
+        this.client.on('disconnect', () => {
+            markDisconnected('disconnect')
+        })
+
+        this.client.on('close', () => {
+            markDisconnected('close')
+        })
+
+        this.client.on('end', () => {
+            markDisconnected('end')
         })
 
         this.client.on('message', (topic, message) => {
@@ -29,6 +59,9 @@ class MqttSyncClient {
                 return;
             }
 
+            if (topic === 'response') {
+                this.lastResponseAt = Date.now()
+            }
 
             // 1. 触发该 topic 的专用监听器
             const topicListeners = this.messageListeners.get(topic);
@@ -56,18 +89,52 @@ class MqttSyncClient {
         })
 
         this.client.on('error', (err) => {
-            console.warn('[MQTT] Error:', err.message)
+            console.warn(`[MQTT] ${ts()} Error:`, err && err.message ? err.message : err)
         })
     }
 
+    /**
+     * 与 mqtt.js 原生 connected 对齐，供单例健康检查
+     */
+    isBrokerConnected() {
+        return !!(this.client && this.client.connected)
+    }
+
+    /**
+     * 注册需在每次 connect 后向 Broker 重新声明订阅的主题（不重复注册消息监听器）
+     * @param {string[]} topics
+     */
+    setMaintainedTopics(topics) {
+        this._maintainedTopics = Array.isArray(topics) ? [...topics] : []
+    }
+
+    _resubscribeMaintainedTopics() {
+        if (!this._maintainedTopics.length) return
+        if (!this.client || !this.client.connected) return
+        const ts = new Date().toISOString()
+        for (const topic of this._maintainedTopics) {
+            this.client.subscribe(topic, { qos: 0 }, (err, granted) => {
+                if (err) {
+                    console.warn(`[MQTT] ${ts} ensureSubscriptions 订阅失败 topic=${topic}`, err.message || err)
+                } else {
+                    console.log(`[MQTT] ${ts} ensureSubscriptions 已订阅 topic=${topic}`, granted)
+                }
+            })
+        }
+    }
+
     waitForConnect(timeout = 5000) {
         return new Promise((resolve, reject) => {
-            if (this.isConnected) return resolve()
+            if (this.isBrokerConnected()) {
+                this.isConnected = true
+                return resolve()
+            }
             const timer = setTimeout(() => {
                 reject(new Error('MQTT connect timeout'))
             }, timeout)
             const check = () => {
-                if (this.isConnected) {
+                if (this.isBrokerConnected()) {
+                    this.isConnected = true
                     clearTimeout(timer)
                     resolve()
                 } else {
@@ -113,7 +180,6 @@ class MqttSyncClient {
                         resolve(granted)
                     }
                 }
-                1
             })
         })
     }
@@ -143,7 +209,12 @@ class MqttSyncClient {
 
             const timer = setTimeout(() => {
                 this.removeMessageListener(onMessage);
-                reject(new Error('Timeout waiting for response'));
+                const last = this.lastResponseAt
+                const extra = last
+                    ? ` lastResponseAt=${new Date(last).toISOString()} brokerConnected=${this.isBrokerConnected()}`
+                    : ` brokerConnected=${this.isBrokerConnected()}`
+                console.warn(`[MQTT] ${new Date().toISOString()} publishAndWait 超时 uuid=${uuid} topic=request_${sn_code}${extra}`)
+                reject(new Error('Timeout waiting for response' + (last ? `; lastResponseAt=${new Date(last).toISOString()}` : '')));
             }, timeout);
 
             const onMessage = (topic, message) => {
@@ -242,6 +313,7 @@ class MqttSyncClient {
     }
 
     end(force = false) {
+        this.isConnected = false
         this.client.end(force)
     }
 }