/** * 网关采集服务(应用级常驻,单 enabled 设备) * * 职责:对当前唯一 enabled 的设备,自动连接 BLE + MQTT, * 持续接收 BLE notify 帧并透传 publish 到 yack 平台, * 跨页面/跨 tab 运行;MQTT 不可用时帧进离线缓存,恢复后自动补传。 * * 页面通过 gateway.subscribe(cb) + gateway.getSnapshot() 订阅实时状态与数据, * 不再由页面各自持有 BLE/MQTT 连接。 * * 并发模型:所有启停/切换操作经 _chain() 串行排队,避免“旧 boot 未结束即切新设备” * 造成漏启/双连。数据通路(收帧→上屏/上报)独立于该队列。 */ const { bleService, BLE_STATE } = require('./ble.js'); const { mqttClient, MQTT_STATE } = require('./mqtt-client.js'); const auth = require('../utils/auth.js'); const storage = require('../utils/storage.js'); const { TIMEOUT, LIMITS, RETRY } = require('../utils/constants.js'); // 网关阶段 const GATEWAY_PHASE = { IDLE: 'idle', // 无目标设备 STARTING: 'starting', // 正在连接(BLE/MQTT 未就绪) RUNNING: 'running', // BLE 已就绪,正在采集(MQTT 可选) ERROR: 'error', // 启动失败,等待重试 STOPPED: 'stopped', // 已停止 }; const GATEWAY_PHASE_TEXT = { idle: '未启用', starting: '连接中', running: '采集中', error: '异常', stopped: '已停止', }; const DEFAULTS = { FLUSH_MS: TIMEOUT.FLUSH_INTERVAL, // 收帧上屏/上报节流 MAX_LIST: LIMITS.MAX_DATA_LIST, // 屏幕保留条数 INIT_RETRY_MS: RETRY.INIT_RETRY_INTERVAL,// 启动失败自动重试间隔 STALL_MS: 30000, // BLE 静默阈值 MQTT_CONNECT_TIMEOUT: TIMEOUT.MQTT_CONNECT, // MQTT 首连超时保护 }; class Gateway { constructor() { this._phase = GATEWAY_PHASE.IDLE; this._device = null; // 正在采集的设备对象 this._bleReady = false; this._mqttReady = false; // 数据与统计 this._pending = []; // 待上屏/上报缓冲 this._dataList = []; // 上屏列表(最新在前) this._stats = { received: 0, uploaded: 0, failed: 0 }; this._offlineCount = 0; // 诊断/生命周期 this._lastDataAt = 0; this._hiddenAt = 0; this._lastError = null; // 定时器与串行队列 this._flushTimer = null; this._initRetryTimer = null; this._flushingOffline = false; this._chainTail = Promise.resolve(); // 页面订阅 this._subs = new Set(); // 唯一数据来源:网关持有 BLE/MQTT 回调 bleService.onData((d) => this._onBleData(d)); bleService.onStateChange((s) => this._onBleState(s)); mqttClient.onStateChange((s) => this._onMqttState(s)); } /* ===================== 对外主流程(经串行队列) ===================== */ /** * 确保采集器运行:以“唯一 enabled 设备”为目标。 */ ensureStarted() { return this._chain(() => this._ensureStartedNow()); } /** * 启动采集(对目标设备)。不同设备会自动先拆旧会话并重置展示数据。 * @param {Object} device - 设备配置(serialNumber/productId/productSecret/deviceId/serviceUUID/enabled...) */ start(device) { return this._chain(() => this._startNow(device || this._device)); } /** * 停止采集(保留已上屏数据与统计,便于回看;拆设备时才 _resetFeed)。 */ stop() { return this._chain(() => this._stopNow()); } /** 切后台(onHide):记录时间,不 teardown */ suspend() { this._hiddenAt = Date.now(); this._clearInitRetry(); } /** 回前台(onShow):补起连接、探测静默并重连、补传离线 */ resume() { const bgMs = this._hiddenAt ? Date.now() - this._hiddenAt : 0; this._hiddenAt = 0; this._flush(); if (!this._hasUser()) return Promise.resolve(); const enabled = storage.getEnabledDevice(); // 无目标 → 若在跑则停;目标变化/未就绪 → 交给队列处理 if (!enabled || !this._device || this._device.serialNumber !== enabled.serialNumber || !this._bleReady) { return this.ensureStarted(); } // 已就绪但静默过久 / 后台超过 5s → 探测并强制重连 const silentMs = Date.now() - (this._lastDataAt || Date.now()); if (bgMs > 5000 || silentMs > DEFAULTS.STALL_MS) { console.log('[GW] 后台时长:', bgMs, 'ms, 静默时长:', silentMs, 'ms'); // 检查 BLE 连接是否真的断开了 if (!bleService.isReady()) { console.warn('[GW] BLE 未就绪,尝试重连'); return bleService.forceReconnect().catch((err) => { this._lastError = err.message || '重连失败'; this._notify(); console.error('[GW] BLE 重连失败:', err); }); } // 探测连接是否仍然活跃 if (!bleService.probeConnection(DEFAULTS.STALL_MS)) { console.warn('[GW] BLE 静默/掉线,强制重连'); return bleService.forceReconnect().catch((err) => { this._lastError = err.message || '重连失败'; this._notify(); console.error('[GW] BLE 重连失败:', err); }); } } // MQTT 若断开:底层 lib 会按 reconnectPeriod 自连,重连成功后 // 经 _onMqttState(CONNECTED) 自动补传离线,此处不重复 connect 以免建重复连接。 if (this._mqttReady) this._flushOffline(); return Promise.resolve(); } /** 清空展示数据与统计(不触碰离线缓存) */ clearHistory() { this._pending = []; this._dataList = []; this._stats = { received: 0, uploaded: 0, failed: 0 }; this._lastDataAt = 0; this._notify(); } /** 手动补传离线缓存(当前设备) */ flushOffline() { return this._flushOffline(); } /* ===================== 页面订阅 ===================== */ /** * 订阅网关快照 * @param {Function} cb - (snapshot) => void * @returns {Function} 退订函数 */ subscribe(cb) { if (typeof cb !== 'function') return () => {}; this._subs.add(cb); return () => this._subs.delete(cb); } /** * 当前快照 * @returns {Object} */ getSnapshot() { return { phase: this._phase, phaseText: GATEWAY_PHASE_TEXT[this._phase] || this._phase, device: this._device ? { ...this._device } : null, bleConnected: this._bleReady, mqttConnected: this._mqttReady, mqttState: mqttClient.getState(), bleState: bleService.getState(), // 浅拷贝元素,确保 setData diff 能识别 uploaded 等字段变化并触发视图刷新 dataList: this._dataList.map(it => ({ ...it })), stats: { ...this._stats }, offlineCount: this._offlineCount, lastError: this._lastError, }; } /** 当前采集设备(浅拷贝) */ getCurrentDevice() { return this._device ? { ...this._device } : null; } /* ===================== 内部:串行队列 ===================== */ _chain(task) { const result = this._chainTail.then(() => task()); // 队列尾不因某次操作失败而中断 this._chainTail = result.catch(() => {}); return result; } async _ensureStartedNow() { if (!this._hasUser()) return; const enabled = storage.getEnabledDevice(); if (!enabled) { await this._stopNow(); return; } await this._startNow(enabled); } async _startNow(device) { if (!device || !device.serialNumber) return; this._clearInitRetry(); const same = !!(this._device && this._device.serialNumber === device.serialNumber); if (same) { // 已在目标会话:BLE 就绪且已运行则幂等返回(MQTT 断开交给 lib 自连 + 失败重试) if (this._bleReady && this._phase === GATEWAY_PHASE.RUNNING) return; await this._boot(); return; } if (this._device) { // 切设备:先拆旧会话(保留旧 feed),再启动新会话并重置展示 await this._teardown(); this._resetFeed(); } this._device = device; this._lastError = null; if (this._phase !== GATEWAY_PHASE.STARTING) this._setPhase(GATEWAY_PHASE.STARTING); this._notify(); await this._boot(); } async _stopNow() { this._clearInitRetry(); if (!this._device && this._phase === GATEWAY_PHASE.STOPPED) return; await this._teardown(); } /* ===================== 内部:启动/停止 ===================== */ async _boot() { const device = this._device; if (!device) return; const alive = () => !!(this._device && this._device.serialNumber === device.serialNumber); try { // 1. BLE:确保适配器打开并连接目标(已就绪同设备则复用) await bleService.initBluetooth(); if (!alive()) return; if (!bleService.isReady() || bleService.deviceId !== device.deviceId) { // 若已有连接/自动重连在途,不重复发起,等其就绪或中止 if (bleService.isConnecting()) { await this._waitBleReady(); } else { await bleService.connectDevice(device.deviceId, device.serviceUUID || null); } if (!alive()) return; // 等待/连接后仍未就绪且无连接在途 → 显式再连一次 if (!bleService.isReady() && !bleService.isConnecting()) { await bleService.connectDevice(device.deviceId, device.serviceUUID || null); } } if (!alive()) return; this._bleReady = true; this._setPhase(GATEWAY_PHASE.RUNNING); this._notify(); // 2. MQTT:已连同一 clientId 时 connect() 内部复用;带超时保护 const creds = auth.getMqttCredentialsFromDevice(device); if (!creds) { throw new Error('获取 MQTT 凭证失败,请检查设备配置'); } console.log('[GW] 开始连接 MQTT...'); await this._connectMqtt(creds); if (!alive()) return; this._mqttReady = true; this._flushOffline(); console.log('[GW] 采集器运行中:', device.serialNumber); } catch (err) { if (!alive()) return; console.error('[GW] 启动失败:', err); // 提供更友好的错误提示 let errorMsg = '启动失败'; if (err.message) { if (err.message.includes('MQTT')) { errorMsg = err.message; } else if (err.message.includes('BLE') || err.message.includes('蓝牙')) { errorMsg = '蓝牙连接失败: ' + err.message; } else if (err.message.includes('凭证')) { errorMsg = err.message; } else { errorMsg = '启动失败: ' + err.message; } } this._lastError = errorMsg; this._setPhase(GATEWAY_PHASE.ERROR); this._scheduleInitRetry(); } } /** * MQTT 连接(带超时保护,避免首连卡死阻塞切换队列) */ _connectMqtt(creds) { return new Promise((resolve, reject) => { let timedOut = false; let timer = null; timer = setTimeout(() => { timedOut = true; timer = null; console.warn('[GW] MQTT 连接超时,断开重试'); mqttClient.disconnect().catch(() => {}); reject(new Error('MQTT 连接超时')); }, DEFAULTS.MQTT_CONNECT_TIMEOUT); mqttClient.connect(creds) .then(() => { if (timer) clearTimeout(timer); if (!timedOut) { resolve(); } }) .catch((err) => { if (timer) clearTimeout(timer); if (!timedOut) { reject(err); } }); }); } /** * 轮询等待 BLE 就绪或连接中止(避免与自动重连并发) * @param {number} timeoutMs * @returns {Promise} */ _waitBleReady(timeoutMs = 15000) { return new Promise((resolve) => { const start = Date.now(); const tick = () => { if (!this._device) { resolve(); return; } if (bleService.isReady() || !bleService.isConnecting() || Date.now() - start > timeoutMs) { resolve(); return; } setTimeout(tick, 200); }; tick(); }); } async _teardown() { this._clearInitRetry(); this._flush(); // 先处理缓冲(此时 _device 仍在) this._bleReady = false; this._mqttReady = false; this._device = null; // 先置空,避免断开回调把 phase 打回 starting try { await bleService.disconnect(); } catch (e) { /* 忽略 */ } try { await mqttClient.disconnect(); } catch (e) { /* 忽略 */ } this._lastError = null; this._setPhase(GATEWAY_PHASE.STOPPED); this._notify(); } _resetFeed() { this._pending = []; this._dataList = []; this._stats = { received: 0, uploaded: 0, failed: 0 }; this._lastDataAt = 0; this._notify(); } /* ===================== 内部:状态回接 ===================== */ _onBleState(s) { if (!this._device) return; const ready = s === BLE_STATE.READY || s === BLE_STATE.CONNECTED; this._bleReady = ready; if (ready) { if (this._phase !== GATEWAY_PHASE.RUNNING) this._setPhase(GATEWAY_PHASE.RUNNING); } else { // 目标会话仍在但 BLE 断开 → 自动重连中 if (this._phase !== GATEWAY_PHASE.ERROR) this._setPhase(GATEWAY_PHASE.STARTING); } this._notify(); } _onMqttState(s) { this._mqttReady = s === MQTT_STATE.CONNECTED; if (this._mqttReady) this._flushOffline(); this._notify(); } /* ===================== 内部:数据通路 ===================== */ _onBleData(d) { if (!this._device || !d || !d.hex) return; this._stats.received++; this._lastDataAt = Date.now(); this._pending.push({ id: `${Date.now()}_${Math.random().toString(36).slice(2, 8)}`, hex: d.hex, time: this._fmtTime(new Date()), uploaded: false, }); if (!this._flushTimer) { this._flushTimer = setTimeout(() => { this._flushTimer = null; this._flush(); }, DEFAULTS.FLUSH_MS); } } /** 节流上屏 + 上报/离线缓存,一次处理一批 */ _flush() { if (this._flushTimer) { clearTimeout(this._flushTimer); this._flushTimer = null; } const batch = this._pending; this._pending = []; if (batch.length === 0) return; this._dataList = batch.concat(this._dataList).slice(0, DEFAULTS.MAX_LIST); const device = this._device; const offline = []; for (const item of batch) { if (this._mqttReady && device && device.enabled) { this._upload(item); // fire-and-forget } else { offline.push(this._toOfflineItem(item, device)); } } if (offline.length > 0) { storage.cachePendingDataBatch(offline); this._refreshOfflineCount(); } this._notify(); } _upload(item) { const device = this._device; if (!device) return; const topic = auth.buildPropertyPostTopic(device.productId, device.serialNumber); mqttClient.publish(topic, item.hex) .then(() => { item.uploaded = true; this._stats.uploaded++; this._notify(); }) .catch((err) => { console.warn('[GW] 上报失败转离线:', err && err.message); this._stats.failed++; if (this._device) { storage.cachePendingData(this._toOfflineItem(item, this._device)); this._refreshOfflineCount(); } this._notify(); }); } _toOfflineItem(item, device) { if (!device) return null; return { deviceId: device.serialNumber, productId: device.productId, hex: item.hex, dataId: item.id, }; } /** 补传当前设备的离线缓存(成功后仅移除本设备项,其他设备残留保留) */ async _flushOffline() { if (this._flushingOffline) return; this._flushingOffline = true; try { const device = this._device; if (!device || !this._mqttReady) return; const list = storage.getPendingData(); if (!list || list.length === 0) return; const topic = auth.buildPropertyPostTopic(device.productId, device.serialNumber); const remaining = []; for (const it of list) { if (it.deviceId === device.serialNumber) { try { await mqttClient.publish(topic, it.hex); this._stats.uploaded++; } catch (err) { console.warn('[GW] 离线补传失败,保留:', it.hex, err && err.message); remaining.push(it); } } else { remaining.push(it); } } storage.setPendingData(remaining); this._refreshOfflineCount(); this._notify(); } finally { this._flushingOffline = false; } } /* ===================== 内部:工具 ===================== */ _hasUser() { const info = storage.getUserInfo(); return !!(info && info.userId); } _setPhase(p) { if (this._phase !== p) { this._phase = p; console.log('[GW] 阶段变更:', p); } } _notify() { const snap = this.getSnapshot(); this._subs.forEach((cb) => { try { cb(snap); } catch (e) { console.error('[GW] 订阅回调异常:', e); } }); } _refreshOfflineCount() { this._offlineCount = storage.getPendingData().length; } _scheduleInitRetry() { if (this._initRetryTimer) return; if (!this._device) return; this._initRetryTimer = setTimeout(() => { this._initRetryTimer = null; if (!this._device) return; console.log('[GW] 启动失败,自动重试...'); this.start(this._device); }, DEFAULTS.INIT_RETRY_MS); } _clearInitRetry() { if (this._initRetryTimer) { clearTimeout(this._initRetryTimer); this._initRetryTimer = null; } } _fmtTime(date) { const h = date.getHours().toString().padStart(2, '0'); const m = date.getMinutes().toString().padStart(2, '0'); const s = date.getSeconds().toString().padStart(2, '0'); return `${h}:${m}:${s}`; } } // 导出单例 const gateway = new Gateway(); module.exports = { GATEWAY_PHASE, GATEWAY_PHASE_TEXT, Gateway, gateway, };