514 lines
18 KiB
JavaScript
514 lines
18 KiB
JavaScript
/**
|
||
* MQTT 客户端服务（包装层）
|
||
*
|
||
* ⚠️ 底层是自研的 `libs/mqtt-wx.min.js`，**不是 mqtt.js**，因此本层不得按 mqtt.js 的语义写：
|
||
* - `end(force)` 的 force 无意义 → 本实现一律"终止性断开"
|
||
* - 没有 `client.reconnecting` 属性（lib 侧已补 getter，但不要靠它做分支判断）
|
||
* - `connect()` 的 Promise 只作为辅助通道，事件才是主通道
|
||
*
|
||
* ⚠️ 曾因错误假设导致的真实故障（已修，勿回退）：
|
||
* 1. `!client.reconnecting` 恒为真 → 任何 error 都拆客户端 → 幽灵客户端
|
||
* 2. 旧实例回调未做实例校验 → 幽灵写入包装器状态 → UI"已连接"但 publish 全失败
|
||
* 3. 建连没有超时出口 → 「连接」按钮永久 loading
|
||
*
|
||
* 本层职责：把裸连接包装成"有状态、有确定语义"的服务
|
||
* - 单一真相源：所有回调做实例校验，旧实例的迟到回调一律丢弃
|
||
* - 彻底终止：换设备/断开时 end() 清掉旧客户端，不留自续重连的幽灵
|
||
* - 有出口：建连失败/超时都能结算，UI 不会永久 loading
|
||
* - 用户意图：userDisconnected 表达"用户主动断开"，自动连接不得覆盖
|
||
*/
|
||
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;
|
||
// 当前连接的 clientId（用于检测连接复用）
|
||
this.currentClientId = null;
|
||
// 用户主动断开意图：置位后自动连接（启动/回前台/网络恢复）不得拉起它
|
||
this.userDisconnected = false;
|
||
// 建连去重：同一 clientId 在途建连复用同一 Promise（避免并发建出两个连接）
|
||
this._connectInFlight = null;
|
||
this._connectInFlightClientId = null;
|
||
}
|
||
|
||
/**
|
||
* 设置消息回调
|
||
* @param {Function} callback - (topic, message) => void
|
||
*/
|
||
onMessage(callback) {
|
||
this.onMessageCallback = callback;
|
||
}
|
||
|
||
/**
|
||
* 设置状态变化回调
|
||
* @param {Function} callback - (state) => void
|
||
*/
|
||
onStateChange(callback) {
|
||
this.onStateChangeCallback = callback;
|
||
}
|
||
|
||
/**
|
||
* 更新状态并通知
|
||
* 去重：重连循环里 CONNECTING / DISCONNECTED 会反复回摆，
|
||
* 重复通知会带动网关重复补传离线，属无谓开销
|
||
* @param {string} state
|
||
*/
|
||
_setState(state) {
|
||
if (this.state === state) return;
|
||
this.state = state;
|
||
if (this.onStateChangeCallback) {
|
||
this.onStateChangeCallback(state);
|
||
}
|
||
console.log('[MQTT] 状态变更:', state);
|
||
}
|
||
|
||
/**
|
||
* 建连前校验网络可用性
|
||
* 必须 await：wx.getNetworkType 是异步回调，若放在建连之后，
|
||
* 检查会晚于 mqtt.connect 执行（断网时仍会建连，且 reject 落在已建连的 promise 上）
|
||
* @returns {Promise}
|
||
*/
|
||
_ensureNetwork() {
|
||
return new Promise((resolve, reject) => {
|
||
wx.getNetworkType({
|
||
success: (res) => {
|
||
console.log('[MQTT] 网络类型:', res.networkType);
|
||
if (res.networkType === 'none') {
|
||
reject(new Error('网络未连接，请检查网络设置'));
|
||
} else {
|
||
resolve();
|
||
}
|
||
},
|
||
fail: () => {
|
||
console.warn('[MQTT] 无法获取网络状态');
|
||
resolve();
|
||
},
|
||
});
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 是否值得发起自动连接
|
||
* 持有客户端即视为"已连接或正在自愈"，不需要再插一脚；
|
||
* 用户主动断开则明确不连。
|
||
* @returns {boolean}
|
||
*/
|
||
canAutoConnect() {
|
||
return !this.userDisconnected && !this.client;
|
||
}
|
||
|
||
/** 清除"用户主动断开"意图（登录 / 重新登录后允许自动连接） */
|
||
allowAutoConnect() {
|
||
this.userDisconnected = false;
|
||
}
|
||
|
||
/**
|
||
* 彻底终止一个客户端实例
|
||
*
|
||
* 不做这件事会留下"幽灵客户端"：对象已不可达，但内部还挂着自续重连定时器，
|
||
* 用同一 clientId 连回来会被 broker 当成新会话踢掉（clean session 只保留一个会话），
|
||
* 表现为状态周期性跳变。
|
||
* @param {Object} client
|
||
*/
|
||
_teardownClient(client) {
|
||
if (!client) return;
|
||
try {
|
||
client.end(true);
|
||
} catch (e) {
|
||
console.warn('[MQTT] 关闭旧客户端异常:', e && e.message);
|
||
}
|
||
if (this.client === client) {
|
||
this.client = null;
|
||
this.currentClientId = null;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 连接 MQTT Broker
|
||
*
|
||
* 失败语义：建连阶段的失败**不拆客户端** —— 底层会按 reconnectPeriod 自愈，
|
||
* 现场无人值守时也能自己连回来（工业现场网络抖动是常态）。
|
||
* 凭证类拒绝由底层置终止标记，避免以 5s 间隔无限撞 broker。
|
||
*
|
||
* @param {Object} options - { clientId, username, password, server }
|
||
* @returns {Promise} 建连结果（失败后底层仍在自愈重连，UI 状态由事件驱动）
|
||
*/
|
||
connect(options) {
|
||
const newClientId = options.clientId;
|
||
|
||
// 同一 clientId 在途建连：复用，绝不并发建出第二个连接
|
||
if (this._connectInFlight && this._connectInFlightClientId === newClientId) {
|
||
console.log('[MQTT] 建连已在进行中，复用本次请求, clientId:', newClientId);
|
||
return this._connectInFlight;
|
||
}
|
||
|
||
const promise = this._ensureNetwork().then(() => new Promise((resolve, reject) => {
|
||
let settled = false;
|
||
let guard = null;
|
||
// 结算闸门：成功 / 失败 / 超时只允许结算一次
|
||
const finish = (fn, arg) => {
|
||
if (settled) return;
|
||
settled = true;
|
||
if (guard) clearTimeout(guard);
|
||
fn(arg);
|
||
};
|
||
|
||
// 同一设备且连接仍可用：幂等复用
|
||
if (this.client && this.client.connected && this.currentClientId === newClientId) {
|
||
console.log('[MQTT] 已连接, 复用连接, clientId:', newClientId);
|
||
finish(resolve);
|
||
return;
|
||
}
|
||
|
||
// 已有客户端：先彻底终止，杜绝"幽灵客户端 + 同 clientId 互踢"
|
||
//
|
||
// ⚠️ 此处**不得**清空 subscriptions：订阅表的生命周期属于"业务意图"而非"某次连接"。
|
||
// 清空后 connect 成功事件里的 _resubscribe() 只能拿到一张空表，订阅被永久丢失；
|
||
// 而换 clientId 重连恰恰是最需要自动恢复订阅的场景。
|
||
// 唯一该清订阅的是 disconnect()（用户明确断开，不再需要这些回调）。
|
||
if (this.client) {
|
||
console.log('[MQTT] 终止旧客户端:', this.currentClientId, '->', newClientId);
|
||
this._teardownClient(this.client);
|
||
}
|
||
|
||
this._setState(MQTT_STATE.CONNECTING);
|
||
this.userDisconnected = false;
|
||
|
||
const server = options.server || envConfig.mqttServer;
|
||
// 连接参数允许被调用方覆盖（调优/测试用）：不传则用工业现场默认值
|
||
const connectOptions = {
|
||
clientId: options.clientId,
|
||
username: options.username,
|
||
password: options.password,
|
||
clean: true,
|
||
keepalive: options.keepalive != null ? options.keepalive : 30,
|
||
connectTimeout: options.connectTimeout != null ? options.connectTimeout : 30000,
|
||
reconnectPeriod: options.reconnectPeriod != null ? options.reconnectPeriod : 5000,
|
||
// 以下为可靠性参数，透传给底层（未传则用底层默认值）：
|
||
// pingTimeout 心跳判死时限 / pubackTimeout QoS1 重发间隔 / maxInflightRetry 重试上限
|
||
pingTimeout: options.pingTimeout,
|
||
pubackTimeout: options.pubackTimeout,
|
||
maxInflightRetry: options.maxInflightRetry,
|
||
maxInflight: options.maxInflight,
|
||
};
|
||
|
||
// 小程序真机只允许 wss:// / wxs://，ws:// 仅在开发者工具可用
|
||
if (/^ws:\/\//.test(server)) {
|
||
console.warn('[MQTT] ws:// 仅开发者工具可用，真机请改用 wss:// 或 wxs://');
|
||
}
|
||
|
||
console.log('[MQTT] 正在连接:', server, 'clientId:', options.clientId);
|
||
|
||
// 包装层兜底超时：没有这个出口，「连接」按钮会永久停在 loading
|
||
guard = setTimeout(() => {
|
||
console.warn('[MQTT] 连接超时(包装层兜底)');
|
||
this._setState(MQTT_STATE.DISCONNECTED);
|
||
finish(reject, new Error('MQTT 连接超时，请检查网络或服务器地址'));
|
||
}, (connectOptions.connectTimeout || 30000) + 5000);
|
||
|
||
let client;
|
||
try {
|
||
client = mqtt.connect(server, connectOptions);
|
||
} catch (err) {
|
||
console.error('[MQTT] 创建客户端失败:', err);
|
||
this._setState(MQTT_STATE.DISCONNECTED);
|
||
finish(reject, new Error('创建 MQTT 客户端失败: ' + (err.message || '未知错误')));
|
||
return;
|
||
}
|
||
|
||
this.client = client;
|
||
this.currentClientId = options.clientId;
|
||
this.reconnectAttempts = 0;
|
||
|
||
// ⭐ 实例校验：只有"当前客户端"的回调能写状态。
|
||
// 旧实例（已 end 的幽灵）的迟到回调一律丢弃，否则它会覆写状态真相源
|
||
const isCurrent = () => this.client === client;
|
||
|
||
client.on('connect', () => {
|
||
if (!isCurrent()) return;
|
||
console.log('[MQTT] 连接成功');
|
||
this._setState(MQTT_STATE.CONNECTED);
|
||
this.reconnectAttempts = 0;
|
||
this._resubscribe();
|
||
finish(resolve);
|
||
});
|
||
|
||
client.on('error', (err) => {
|
||
if (!isCurrent()) return;
|
||
console.error('[MQTT] 连接错误:', err);
|
||
const errMsg = (err && err.message) || '未知错误';
|
||
|
||
let userMsg = 'MQTT 连接失败';
|
||
if (errMsg.includes('ECONNREFUSED')) {
|
||
userMsg = 'MQTT 服务器拒绝连接，请检查服务器配置';
|
||
} else if (errMsg.includes('ENOTFOUND')) {
|
||
userMsg = 'MQTT 服务器地址解析失败，请检查地址';
|
||
} else if (errMsg.includes('ETIMEDOUT') || errMsg.includes('timeout') || errMsg.includes('超时')) {
|
||
userMsg = 'MQTT 连接超时，请检查网络或服务器状态';
|
||
} else if (errMsg.includes('WebSocket')) {
|
||
userMsg = 'WebSocket 连接失败，请检查协议（wss:// 或 wxs://）';
|
||
} else if (errMsg.includes('Bad username') || errMsg.includes('Not authorized')) {
|
||
userMsg = 'MQTT 产品凭证不正确，请核对设备编号 / 产品ID / 产品密钥';
|
||
}
|
||
|
||
// 建连阶段的失败：只结算，不拆客户端 —— 让底层继续自愈重连。
|
||
// 建连之后（已 resolve）的会话中抖动不 reject，仅记日志
|
||
if (!settled) {
|
||
this._setState(MQTT_STATE.DISCONNECTED);
|
||
finish(reject, new Error(userMsg + ': ' + errMsg));
|
||
}
|
||
});
|
||
|
||
client.on('close', () => {
|
||
if (!isCurrent()) return;
|
||
console.log('[MQTT] 连接关闭');
|
||
this._setState(MQTT_STATE.DISCONNECTED);
|
||
});
|
||
|
||
client.on('reconnect', () => {
|
||
if (!isCurrent()) return;
|
||
this.reconnectAttempts++;
|
||
console.log('[MQTT] 正在重连, 尝试次数:', this.reconnectAttempts);
|
||
this._setState(MQTT_STATE.CONNECTING);
|
||
});
|
||
|
||
client.on('message', (topic, message) => {
|
||
if (!isCurrent()) return;
|
||
const msgStr = message.toString();
|
||
console.log('[MQTT] 收到消息, topic:', topic, 'message:', msgStr);
|
||
if (this.onMessageCallback) {
|
||
this.onMessageCallback(topic, msgStr);
|
||
}
|
||
if (this.subscriptions.has(topic)) {
|
||
this.subscriptions.get(topic)(topic, msgStr);
|
||
}
|
||
});
|
||
}));
|
||
|
||
this._connectInFlight = promise;
|
||
this._connectInFlightClientId = newClientId;
|
||
const clear = () => {
|
||
if (this._connectInFlight === promise) {
|
||
this._connectInFlight = null;
|
||
this._connectInFlightClientId = null;
|
||
}
|
||
};
|
||
promise.then(clear, clear);
|
||
return promise;
|
||
}
|
||
|
||
/**
|
||
* 硬重连：先彻底关掉现有客户端（连带停掉底层自动重连），再按凭证重新建连
|
||
*
|
||
* 手动重连必须是"点了就一定有结果"的确定性出口：底层可能停在僵尸 socket 上，
|
||
* 只有 end() 清干净再建才是确定的。订阅表保留，建连成功后由 _resubscribe 恢复。
|
||
* @param {Object} options - { clientId, username, password, server }
|
||
* @returns {Promise}
|
||
*/
|
||
reconnect(options) {
|
||
this._teardownClient(this.client);
|
||
this._setState(MQTT_STATE.DISCONNECTED);
|
||
return this.connect(options);
|
||
}
|
||
|
||
/**
|
||
* 发布消息
|
||
* QoS1 由底层等待 PUBACK 才算成功（超时会按 dup 重发，重试用尽才回调错误），
|
||
* 因此这里的成功回调代表"broker 已确认收到"
|
||
*
|
||
* ⚠️ 必须有超时出口：底层的成功/失败**完全依赖回调**，而回调要等 PUBACK 或重试用尽。
|
||
* 一旦某条消息卡在在途队列里不结算，这个 Promise 就永久 pending —— 网关的离线补传
|
||
* （_flushOffline 串行 await publish，且被 _flushingOffline 闸门保护）会被它一次性
|
||
* 锁死，之后所有补传静默失效直到重启小程序。BLE 侧每个 GATT 操作都做了超时兜底，
|
||
* 这里不能是唯一的例外。
|
||
* @param {string} topic - 主题
|
||
* @param {string} message - 消息内容
|
||
* @param {Object} options - 发布选项 {qos: 0|1|2, retain: boolean, publishTimeout: number}
|
||
* @returns {Promise}
|
||
*/
|
||
publish(topic, message, options = { qos: 1 }) {
|
||
const timeout = options.publishTimeout != null ? options.publishTimeout : 15000;
|
||
return new Promise((resolve, reject) => {
|
||
if (!this.client || !this.client.connected) {
|
||
reject(new Error('MQTT 未连接'));
|
||
return;
|
||
}
|
||
|
||
let settled = false;
|
||
const timer = setTimeout(() => {
|
||
if (settled) return;
|
||
settled = true;
|
||
console.warn('[MQTT] 发布超时, topic:', topic);
|
||
reject(new Error('发布超时'));
|
||
}, timeout);
|
||
|
||
this.client.publish(topic, message, options, (err) => {
|
||
if (settled) return;
|
||
settled = true;
|
||
clearTimeout(timer);
|
||
if (err) {
|
||
console.error('[MQTT] 发布失败:', err);
|
||
reject(new Error('发布失败'));
|
||
} else {
|
||
console.log('[MQTT] 发布成功, topic:', topic);
|
||
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) {
|
||
this.subscriptions.delete(topic);
|
||
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);
|
||
});
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 主动断开连接（终止性操作）
|
||
*
|
||
* 置 userDisconnected = true 表达"用户意图"：网关的自动连接不再把它拉起来，
|
||
* 直到用户再次手动连接（或被显式 reconnect）。
|
||
* @returns {Promise}
|
||
*/
|
||
disconnect() {
|
||
this.userDisconnected = true;
|
||
const client = this.client;
|
||
|
||
return new Promise((resolve) => {
|
||
if (!client) {
|
||
this._setState(MQTT_STATE.DISCONNECTED);
|
||
resolve();
|
||
return;
|
||
}
|
||
|
||
this.subscriptions.clear();
|
||
try {
|
||
client.end(false, () => {
|
||
if (this.client === client) {
|
||
this._setState(MQTT_STATE.DISCONNECTED);
|
||
this.client = null;
|
||
this.currentClientId = null;
|
||
}
|
||
console.log('[MQTT] 已断开连接');
|
||
resolve();
|
||
});
|
||
} catch (e) {
|
||
console.warn('[MQTT] 断开异常，强制清理:', e && e.message);
|
||
this.client = null;
|
||
this.currentClientId = null;
|
||
this._setState(MQTT_STATE.DISCONNECTED);
|
||
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,
|
||
};
|