|
@@ -16,12 +16,14 @@ const MQTT_GLOBAL_TOPIC = 'MODULE/TASK/STATUS/GLOBAL';
|
|
|
const options = {
|
|
const options = {
|
|
|
clean: true, // true: 清除会话, false: 保留会话
|
|
clean: true, // true: 清除会话, false: 保留会话
|
|
|
connectTimeout: 4000, // 超时时间
|
|
connectTimeout: 4000, // 超时时间
|
|
|
- // 认证信息
|
|
|
|
|
- // clientId: 'mqttx_58db45cc7777',
|
|
|
|
|
username: 'mqttx1112',
|
|
username: 'mqttx1112',
|
|
|
password: '',
|
|
password: '',
|
|
|
|
|
+ keepalive: 120, // 心跳间隔 120 秒(更宽松,减少超时)
|
|
|
|
|
+ reconnectPeriod: 5000, // 断线后 5 秒自动重连
|
|
|
|
|
+ clientId: `mqtt_${Math.random().toString(16).substring(2, 10)}`, // 唯一客户端 ID
|
|
|
};
|
|
};
|
|
|
-let mqttClient;
|
|
|
|
|
|
|
+let mqttClient: any = null;
|
|
|
|
|
+let isConnecting = false; // 连接状态标志,防止重复调用
|
|
|
|
|
|
|
|
const handleMqttMessage = (message: MqttMessage) => {
|
|
const handleMqttMessage = (message: MqttMessage) => {
|
|
|
switch (message.status) {
|
|
switch (message.status) {
|
|
@@ -58,12 +60,46 @@ const handleMqttMessage = (message: MqttMessage) => {
|
|
|
|
|
|
|
|
const startListening = async () => {
|
|
const startListening = async () => {
|
|
|
try {
|
|
try {
|
|
|
|
|
+ // 防止重复调用
|
|
|
|
|
+ if (isConnecting) {
|
|
|
|
|
+ console.log('[mqttService] 连接正在进行中,忽略重复调用');
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
const MQTT_BROKER_URL = getMqttBrokerUrl();
|
|
const MQTT_BROKER_URL = getMqttBrokerUrl();
|
|
|
- console.log(`[mqttService] startListening with broker: ${MQTT_BROKER_URL}`);
|
|
|
|
|
|
|
+ const timestamp = () => new Date().toISOString();
|
|
|
|
|
+
|
|
|
|
|
+ console.log(`[${timestamp()}] [mqttService] startListening with broker: ${MQTT_BROKER_URL}`);
|
|
|
|
|
+
|
|
|
|
|
+ // 清理旧连接
|
|
|
|
|
+ if (mqttClient) {
|
|
|
|
|
+ console.log(`[${timestamp()}] [mqttService] 检测到旧连接,正在清理...`);
|
|
|
|
|
+ try {
|
|
|
|
|
+ // 移除所有事件监听器,防止内存泄漏
|
|
|
|
|
+ mqttClient.removeAllListeners();
|
|
|
|
|
+ // 强制关闭连接
|
|
|
|
|
+ mqttClient.end(true);
|
|
|
|
|
+ mqttClient = null;
|
|
|
|
|
+ console.log(`[${timestamp()}] [mqttService] 旧连接已清理`);
|
|
|
|
|
+ } catch (cleanupError) {
|
|
|
|
|
+ console.error(`[${timestamp()}] [mqttService] 清理旧连接时出错:`, cleanupError);
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
|
|
|
|
|
|
|
+ isConnecting = true;
|
|
|
|
|
+
|
|
|
mqttClient = mqtt.connect(MQTT_BROKER_URL, options);
|
|
mqttClient = mqtt.connect(MQTT_BROKER_URL, options);
|
|
|
- mqttClient.on('connect', () => {
|
|
|
|
|
- console.log('[mqttService] Connected to MQTT broker');
|
|
|
|
|
|
|
+ mqttClient.on('connect', (connack) => {
|
|
|
|
|
+ isConnecting = false;
|
|
|
|
|
+ const timestamp = () => new Date().toISOString();
|
|
|
|
|
+ console.log(`[${timestamp()}] [mqttService] ✅ Connected to MQTT broker`, {
|
|
|
|
|
+ sessionPresent: connack.sessionPresent,
|
|
|
|
|
+ returnCode: connack.returnCode
|
|
|
|
|
+ });
|
|
|
|
|
+
|
|
|
|
|
+ // 发送连接成功事件到应用层
|
|
|
|
|
+ emitter.emit('MQTT_CONNECTED');
|
|
|
|
|
+
|
|
|
mqttClient.subscribe(MQTT_TOPIC, (err) => {
|
|
mqttClient.subscribe(MQTT_TOPIC, (err) => {
|
|
|
if (err) {
|
|
if (err) {
|
|
|
console.error('[mqttService] Failed to subscribe to topic', err);
|
|
console.error('[mqttService] Failed to subscribe to topic', err);
|
|
@@ -76,8 +112,48 @@ const startListening = async () => {
|
|
|
});
|
|
});
|
|
|
});
|
|
});
|
|
|
|
|
|
|
|
|
|
+ // 重连事件
|
|
|
|
|
+ mqttClient.on('reconnect', () => {
|
|
|
|
|
+ const timestamp = () => new Date().toISOString();
|
|
|
|
|
+ console.log(`[${timestamp()}] [mqttService] 🔄 正在尝试重连...`);
|
|
|
|
|
+ emitter.emit('MQTT_RECONNECTING');
|
|
|
|
|
+ });
|
|
|
|
|
+
|
|
|
|
|
+ // 连接关闭事件
|
|
|
|
|
+ mqttClient.on('close', () => {
|
|
|
|
|
+ const timestamp = () => new Date().toISOString();
|
|
|
|
|
+ console.log(`[${timestamp()}] [mqttService] 🚪 连接已关闭`);
|
|
|
|
|
+ emitter.emit('MQTT_DISCONNECTED');
|
|
|
|
|
+ });
|
|
|
|
|
+
|
|
|
|
|
+ // 离线事件
|
|
|
|
|
+ mqttClient.on('offline', () => {
|
|
|
|
|
+ const timestamp = () => new Date().toISOString();
|
|
|
|
|
+ console.log(`[${timestamp()}] [mqttService] 📡 客户端离线`);
|
|
|
|
|
+ emitter.emit('MQTT_OFFLINE');
|
|
|
|
|
+ });
|
|
|
|
|
+
|
|
|
|
|
+ // 断开连接事件
|
|
|
|
|
+ mqttClient.on('disconnect', (packet) => {
|
|
|
|
|
+ const timestamp = () => new Date().toISOString();
|
|
|
|
|
+ console.log(`[${timestamp()}] [mqttService] ⚠️ 主动断开连接`, packet);
|
|
|
|
|
+ });
|
|
|
|
|
+
|
|
|
|
|
+ // 错误事件
|
|
|
mqttClient.on('error', (error) => {
|
|
mqttClient.on('error', (error) => {
|
|
|
- console.error('[mqttService] 连接失败:', error);
|
|
|
|
|
|
|
+ isConnecting = false;
|
|
|
|
|
+ const timestamp = () => new Date().toISOString();
|
|
|
|
|
+ console.error(`[${timestamp()}] [mqttService] ❌ 连接失败:`, {
|
|
|
|
|
+ message: error.message,
|
|
|
|
|
+ code: error.code,
|
|
|
|
|
+ stack: error.stack?.split('\n')[0] // 只记录第一行堆栈
|
|
|
|
|
+ });
|
|
|
|
|
+
|
|
|
|
|
+ // 发送错误事件到应用层
|
|
|
|
|
+ emitter.emit('MQTT_ERROR', {
|
|
|
|
|
+ message: error.message,
|
|
|
|
|
+ code: error.code
|
|
|
|
|
+ });
|
|
|
});
|
|
});
|
|
|
|
|
|
|
|
mqttClient.on('message', (topic, message) => {
|
|
mqttClient.on('message', (topic, message) => {
|
|
@@ -100,7 +176,30 @@ const startListening = async () => {
|
|
|
};
|
|
};
|
|
|
|
|
|
|
|
const stopListening = () => {
|
|
const stopListening = () => {
|
|
|
- mqttClient.end();
|
|
|
|
|
- console.log('Stopped listening to MQTT broker');
|
|
|
|
|
|
|
+ const timestamp = () => new Date().toISOString();
|
|
|
|
|
+
|
|
|
|
|
+ if (!mqttClient) {
|
|
|
|
|
+ console.log(`[${timestamp()}] [mqttService] 没有活跃的连接需要关闭`);
|
|
|
|
|
+ return;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ console.log(`[${timestamp()}] [mqttService] 正在停止监听...`);
|
|
|
|
|
+
|
|
|
|
|
+ try {
|
|
|
|
|
+ // 移除所有事件监听器
|
|
|
|
|
+ mqttClient.removeAllListeners();
|
|
|
|
|
+ // 优雅关闭连接
|
|
|
|
|
+ mqttClient.end(false, () => {
|
|
|
|
|
+ console.log(`[${timestamp()}] [mqttService] ✅ 已停止监听 MQTT broker`);
|
|
|
|
|
+ mqttClient = null;
|
|
|
|
|
+ isConnecting = false;
|
|
|
|
|
+ });
|
|
|
|
|
+ } catch (error) {
|
|
|
|
|
+ console.error(`[${timestamp()}] [mqttService] 停止监听时出错:`, error);
|
|
|
|
|
+ // 强制关闭
|
|
|
|
|
+ mqttClient.end(true);
|
|
|
|
|
+ mqttClient = null;
|
|
|
|
|
+ isConnecting = false;
|
|
|
|
|
+ }
|
|
|
};
|
|
};
|
|
|
export { mqttClient, handleMqttMessage, startListening, stopListening };
|
|
export { mqttClient, handleMqttMessage, startListening, stopListening };
|