317 lines
8.5 KiB
JavaScript
317 lines
8.5 KiB
JavaScript
/**
|
||
* MQTT 客户端服务
|
||
*/
|
||
const mqtt = require('../libs/mqtt-wx.min.js');
|
||
const envConfig = require('../env.config.js');
|
||
|
||
// MQTT 状态常量
|
||
const MQTT_STATE = {
|
||
DISCONNECTED: 'disconnected',
|
||
CONNECTING: 'connecting',
|
||
CONNECTED: 'connected',
|
||
};
|
||
|
||
class MqttClient {
|
||
constructor() {
|
||
this.client = null;
|
||
this.state = MQTT_STATE.DISCONNECTED;
|
||
this.subscriptions = new Map();
|
||
this.onMessageCallback = null;
|
||
this.onStateChangeCallback = null;
|
||
this.reconnectAttempts = 0;
|
||
this.maxReconnectAttempts = 5;
|
||
// 当前连接的 clientId（用于检测连接复用）
|
||
this.currentClientId = null;
|
||
}
|
||
|
||
/**
|
||
* 设置消息回调
|
||
* @param {Function} callback - (topic, message) => void
|
||
*/
|
||
onMessage(callback) {
|
||
this.onMessageCallback = callback;
|
||
}
|
||
|
||
/**
|
||
* 设置状态变化回调
|
||
* @param {Function} callback - (state) => void
|
||
*/
|
||
onStateChange(callback) {
|
||
this.onStateChangeCallback = callback;
|
||
}
|
||
|
||
/**
|
||
* 更新状态并通知
|
||
* @param {string} state
|
||
*/
|
||
_setState(state) {
|
||
this.state = state;
|
||
if (this.onStateChangeCallback) {
|
||
this.onStateChangeCallback(state);
|
||
}
|
||
console.log('[MQTT] 状态变更:', state);
|
||
}
|
||
|
||
/**
|
||
* 连接 MQTT Broker
|
||
* @param {Object} options - 连接配置
|
||
* @param {string} options.clientId - 客户端 ID
|
||
* @param {string} options.username - 用户名
|
||
* @param {string} options.password - 密码
|
||
* @param {string} options.server - 服务器地址（可选，默认使用 envConfig）
|
||
* @returns {Promise}
|
||
*/
|
||
connect(options) {
|
||
return new Promise((resolve, reject) => {
|
||
const newClientId = options.clientId;
|
||
|
||
// 检查是否需要重建连接（clientId 变化或未连接）
|
||
if (this.client && this.client.connected) {
|
||
if (this.currentClientId === newClientId) {
|
||
// 同一设备，复用连接
|
||
console.log('[MQTT] 已连接, 复用连接, clientId:', newClientId);
|
||
resolve();
|
||
return;
|
||
}
|
||
// 不同设备，先断开旧连接，防止数据串号
|
||
console.log('[MQTT] clientId 变化，断开旧连接:', this.currentClientId, '->', newClientId);
|
||
this.client.end(true); // 强制关闭，停止重连
|
||
this.client = null;
|
||
this.subscriptions.clear();
|
||
}
|
||
|
||
this._setState(MQTT_STATE.CONNECTING);
|
||
|
||
const server = options.server || envConfig.mqttServer;
|
||
const connectOptions = {
|
||
clientId: options.clientId,
|
||
username: options.username,
|
||
password: options.password,
|
||
clean: true,
|
||
keepalive: 30,
|
||
connectTimeout: 30000,
|
||
reconnectPeriod: 5000,
|
||
// protocol 从 URL 自动解析，无需手动设置
|
||
};
|
||
|
||
console.log('[MQTT] 正在连接:', server, 'clientId:', options.clientId);
|
||
|
||
try {
|
||
this.client = mqtt.connect(server, connectOptions);
|
||
this.currentClientId = options.clientId;
|
||
|
||
// 连接成功
|
||
this.client.on('connect', () => {
|
||
console.log('[MQTT] 连接成功');
|
||
this._setState(MQTT_STATE.CONNECTED);
|
||
this.reconnectAttempts = 0;
|
||
// 重新订阅之前的主题
|
||
this._resubscribe();
|
||
resolve();
|
||
});
|
||
|
||
// 连接错误
|
||
this.client.on('error', (err) => {
|
||
console.error('[MQTT] 连接错误:', err);
|
||
// 强制关闭客户端，停止重连
|
||
if (this.client) {
|
||
this.client.end(true);
|
||
this.client = null;
|
||
}
|
||
this._setState(MQTT_STATE.DISCONNECTED);
|
||
reject(new Error('MQTT 连接失败: ' + (err.message || '未知错误')));
|
||
});
|
||
|
||
// 断开连接
|
||
this.client.on('close', () => {
|
||
console.log('[MQTT] 连接关闭');
|
||
this._setState(MQTT_STATE.DISCONNECTED);
|
||
});
|
||
|
||
// 重连
|
||
this.client.on('reconnect', () => {
|
||
this.reconnectAttempts++;
|
||
console.log('[MQTT] 正在重连, 尝试次数:', this.reconnectAttempts);
|
||
this._setState(MQTT_STATE.CONNECTING);
|
||
if (this.reconnectAttempts > this.maxReconnectAttempts) {
|
||
console.log('[MQTT] 重连次数超限，停止重连');
|
||
this.disconnect();
|
||
}
|
||
});
|
||
|
||
// 收到消息
|
||
this.client.on('message', (topic, message) => {
|
||
const msgStr = message.toString();
|
||
console.log('[MQTT] 收到消息, topic:', topic, 'message:', msgStr);
|
||
if (this.onMessageCallback) {
|
||
this.onMessageCallback(topic, msgStr);
|
||
}
|
||
// 触发订阅回调
|
||
if (this.subscriptions.has(topic)) {
|
||
const callback = this.subscriptions.get(topic);
|
||
callback(topic, msgStr);
|
||
}
|
||
});
|
||
|
||
// 离线
|
||
this.client.on('offline', () => {
|
||
console.log('[MQTT] 离线');
|
||
this._setState(MQTT_STATE.DISCONNECTED);
|
||
});
|
||
|
||
} catch (err) {
|
||
console.error('[MQTT] 创建客户端失败:', err);
|
||
this._setState(MQTT_STATE.DISCONNECTED);
|
||
reject(new Error('创建 MQTT 客户端失败'));
|
||
}
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 发布消息
|
||
* @param {string} topic - 主题
|
||
* @param {string} message - 消息内容
|
||
* @param {Object} options - 发布选项 {qos: 0|1|2, retain: boolean}
|
||
* @returns {Promise}
|
||
*/
|
||
publish(topic, message, options = { qos: 1 }) {
|
||
return new Promise((resolve, reject) => {
|
||
if (!this.client || !this.client.connected) {
|
||
reject(new Error('MQTT 未连接'));
|
||
return;
|
||
}
|
||
|
||
this.client.publish(topic, message, options, (err) => {
|
||
if (err) {
|
||
console.error('[MQTT] 发布失败:', err);
|
||
reject(new Error('发布失败'));
|
||
} else {
|
||
console.log('[MQTT] 发布成功, topic:', topic, 'message:', message);
|
||
resolve();
|
||
}
|
||
});
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 订阅主题
|
||
* @param {string} topic - 主题
|
||
* @param {Function} callback - 消息回调（可选）
|
||
* @param {Object} options - 订阅选项 {qos: 0|1|2}
|
||
* @returns {Promise}
|
||
*/
|
||
subscribe(topic, callback = null, options = { qos: 1 }) {
|
||
return new Promise((resolve, reject) => {
|
||
if (!this.client || !this.client.connected) {
|
||
reject(new Error('MQTT 未连接'));
|
||
return;
|
||
}
|
||
|
||
this.client.subscribe(topic, options, (err) => {
|
||
if (err) {
|
||
console.error('[MQTT] 订阅失败:', err);
|
||
reject(new Error('订阅失败'));
|
||
} else {
|
||
console.log('[MQTT] 订阅成功:', topic);
|
||
if (callback) {
|
||
this.subscriptions.set(topic, callback);
|
||
}
|
||
resolve();
|
||
}
|
||
});
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 取消订阅
|
||
* @param {string} topic - 主题
|
||
* @returns {Promise}
|
||
*/
|
||
unsubscribe(topic) {
|
||
return new Promise((resolve, reject) => {
|
||
if (!this.client || !this.client.connected) {
|
||
resolve();
|
||
return;
|
||
}
|
||
|
||
this.client.unsubscribe(topic, (err) => {
|
||
if (err) {
|
||
console.error('[MQTT] 取消订阅失败:', err);
|
||
reject(new Error('取消订阅失败'));
|
||
} else {
|
||
console.log('[MQTT] 取消订阅成功:', topic);
|
||
this.subscriptions.delete(topic);
|
||
resolve();
|
||
}
|
||
});
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 重新订阅所有主题
|
||
*/
|
||
_resubscribe() {
|
||
if (this.subscriptions.size === 0) return;
|
||
|
||
this.subscriptions.forEach((callback, topic) => {
|
||
this.subscribe(topic, callback).catch(err => {
|
||
console.error('[MQTT] 重订阅失败:', topic, err);
|
||
});
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 断开连接
|
||
*/
|
||
disconnect() {
|
||
return new Promise((resolve) => {
|
||
if (!this.client) {
|
||
resolve();
|
||
return;
|
||
}
|
||
|
||
this.client.end(false, () => {
|
||
console.log('[MQTT] 已断开连接');
|
||
this._setState(MQTT_STATE.DISCONNECTED);
|
||
this.client = null;
|
||
this.currentClientId = null;
|
||
this.subscriptions.clear();
|
||
resolve();
|
||
});
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 获取当前状态
|
||
* @returns {string}
|
||
*/
|
||
getState() {
|
||
return this.state;
|
||
}
|
||
|
||
/**
|
||
* 是否已连接
|
||
* @returns {boolean}
|
||
*/
|
||
isConnected() {
|
||
return this.client && this.client.connected;
|
||
}
|
||
|
||
/**
|
||
* 清除消息回调（页面卸载时调用）
|
||
*/
|
||
clearCallbacks() {
|
||
this.onMessageCallback = null;
|
||
this.onStateChangeCallback = null;
|
||
console.log('[MQTT] 已清除回调');
|
||
}
|
||
}
|
||
|
||
// 导出单例
|
||
const mqttClient = new MqttClient();
|
||
|
||
module.exports = {
|
||
MQTT_STATE,
|
||
MqttClient,
|
||
mqttClient,
|
||
}; |