873 lines
33 KiB
JavaScript
873 lines
33 KiB
JavaScript
/**
|
||
* 网关采集服务（应用级常驻，单 enabled 设备）
|
||
*
|
||
* 职责：对当前唯一 enabled 的设备，自动连接 BLE 并持续接收 notify 帧，
|
||
* 按「测点链路」上报：topic = /{productId}/{sn}/{measurePointId}/measure/post
|
||
* - MQTT 连接由「我的」页独立建立（产品凭证），本服务只复用其连接状态
|
||
* - 采集模式：1=手动确认（数据需点击确认才上报）、2=自动上传（默认）
|
||
* - MQTT 不可用时帧进离线缓存（带测点ID），恢复后自动补传
|
||
*
|
||
* 页面通过 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 measurePoint = require('../utils/measure-point.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 静默阈值（超过则回前台时探活）
|
||
RESUME_PROBE_MS: 5000, // 后台超过该时长 → 回前台必须探活
|
||
};
|
||
|
||
class Gateway {
|
||
constructor() {
|
||
this._phase = GATEWAY_PHASE.IDLE;
|
||
this._device = null; // 正在采集的设备对象
|
||
this._bleReady = false;
|
||
// 初始取真实状态：MQTT 可能在本单例构造前就已连上（事件回调是一次性的）
|
||
this._mqttReady = mqttClient.isConnected();
|
||
this._mqttConnecting = 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() {
|
||
this._syncTarget();
|
||
return this._chain(() => this._ensureStartedNow());
|
||
}
|
||
|
||
/**
|
||
* 确保 MQTT 已连接（自动连接的唯一出口，幂等）
|
||
*
|
||
* 「我的」页录入的产品凭证是持久化的，采集器又要常驻 —— 因此冷启动、回前台、
|
||
* 网络恢复时都应自动把 MQTT 拉起来，而不是等用户手点「连接」。
|
||
* 唯一例外是用户主动断开（mqttClient.userDisconnected）：那是明确意图，不覆盖。
|
||
* @returns {Promise<boolean>} 是否已连接
|
||
*/
|
||
ensureMqtt() {
|
||
return this._chain(() => this._ensureMqttNow());
|
||
}
|
||
|
||
/**
|
||
* 启动采集（对目标设备）。不同设备会自动先拆旧会话并重置展示数据。
|
||
* @param {Object} device - 设备配置（serialNumber/productId/productSecret/deviceId/serviceUUID/enabled...）
|
||
*/
|
||
start(device) {
|
||
this._syncTarget();
|
||
return this._chain(() => this._startNow(device || this._device));
|
||
}
|
||
|
||
/**
|
||
* 停止采集（保留已上屏数据与统计，便于回看；拆设备时才 _resetFeed）。
|
||
*/
|
||
stop() {
|
||
this._syncTarget();
|
||
return this._chain(() => this._stopNow());
|
||
}
|
||
|
||
/** 切后台（onHide）：记录时间，不 teardown */
|
||
suspend() {
|
||
this._hiddenAt = Date.now();
|
||
this._clearInitRetry();
|
||
}
|
||
|
||
/** 回前台（onShow）：补起连接、探测静默并重连、补传离线 */
|
||
async resume() {
|
||
const bgMs = this._hiddenAt ? Date.now() - this._hiddenAt : 0;
|
||
this._hiddenAt = 0;
|
||
this._flush();
|
||
|
||
if (!this._hasUser()) return;
|
||
|
||
// MQTT 与 BLE 平权自愈：冷启动/回前台/网络恢复都补一次自动连接，
|
||
// 否则"自动上传"会在用户没进「我的」页时静默退化成离线堆积。
|
||
// MQTT 是应用级连接，设备开关不影响它
|
||
this._ensureMqttNow();
|
||
|
||
const enabled = storage.getEnabledDevice();
|
||
// 无目标 → 若在跑则停；目标变化/未就绪 → 交给队列处理
|
||
if (!enabled || !this._device || this._device.serialNumber !== enabled.serialNumber || !this._bleReady) {
|
||
return this.ensureStarted();
|
||
}
|
||
|
||
// 回前台（后台超阈值）或长时间静默 → 必须真正探活一次，再决定是否重连
|
||
const silentMs = Date.now() - (this._lastDataAt || Date.now());
|
||
if (bgMs > DEFAULTS.RESUME_PROBE_MS || silentMs > DEFAULTS.STALL_MS) {
|
||
console.log('[GW] 后台时长:', bgMs, 'ms, 静默时长:', silentMs, 'ms');
|
||
|
||
if (!bleService.isReady()) {
|
||
// 状态已不是就绪：直接重连（重连中/失败由 ble 内部按退避自愈）
|
||
console.warn('[GW] BLE 未就绪，强制重连');
|
||
return this._reconnectBle();
|
||
}
|
||
|
||
// 静默计时器只说明"这段时间没收到包"，它区分不了"设备没在测量（静默正常）"
|
||
// 与"链路已被系统收走"；回前台尤其危险 —— native 常在切后台时静默断链且
|
||
// 不派发状态回调，只看静默时长就会漏判（状态显示"已连接"却再也收不到数据）。
|
||
// 唯一可靠的判据是 GATT 读一次：探活失败才拆链重建，通过则原样复用
|
||
// （探活是单次读，不做周期轮询，不会把健康链路反复中断）。
|
||
const alive = await bleService.checkLinkAlive();
|
||
if (!this._device) return;
|
||
if (alive) {
|
||
console.log('[GW] BLE 探活通过，复用现有链路');
|
||
} else {
|
||
console.warn('[GW] BLE 链路已死，强制重连');
|
||
return this._reconnectBle();
|
||
}
|
||
}
|
||
|
||
// MQTT 若断开：底层 lib 会按 reconnectPeriod 自连，重连成功后
|
||
// 经 _onMqttState(CONNECTED) 自动补传离线，此处不重复 connect 以免建重复连接。
|
||
if (this._mqttReady) this._flushOffline();
|
||
}
|
||
|
||
/**
|
||
* BLE 重连（失败不上屏为错误）
|
||
* ble 内部已按指数退避自动重排重连，此处若把瞬时失败写成 lastError，
|
||
* 链路恢复后详情页的 {{lastError}} 会一直残留（"已连上却还显示报错"）
|
||
* @returns {Promise}
|
||
*/
|
||
_reconnectBle() {
|
||
return bleService.forceReconnect().catch((err) => {
|
||
console.error('[GW] BLE 重连失败(将自动重试):', err && err.message);
|
||
});
|
||
}
|
||
|
||
/** 清空展示数据与统计（不触碰离线缓存） */
|
||
clearHistory() {
|
||
this._pending = [];
|
||
this._dataList = [];
|
||
this._stats = { received: 0, uploaded: 0, failed: 0 };
|
||
this._lastDataAt = 0;
|
||
this._notify();
|
||
}
|
||
|
||
/** 手动补传离线缓存（当前设备） */
|
||
flushOffline() {
|
||
return this._flushOffline();
|
||
}
|
||
|
||
/**
|
||
* 手动重连 BLE（详情页 BLE 状态点击触发）
|
||
*
|
||
* 与"关开关再开"的区别：走复位蓝牙适配器（close → open）彻底清掉 native 残留连接，
|
||
* 这是"连续连接超时后自动重连卡死"场景的确定性出口；
|
||
* 关开关只是 stop + start，native 半开连接仍在，会一直 already connect / 超时。
|
||
*/
|
||
retryBle() {
|
||
return this._chain(async () => {
|
||
if (!this._hasUser()) {
|
||
return { ok: false, message: '未登录' };
|
||
}
|
||
const device = this._device;
|
||
if (!device) {
|
||
return { ok: false, message: '设备未启用采集，请先在“我的设备”页开启' };
|
||
}
|
||
// 详情页的 canRetryBle 只看"是否当前采集设备"，设备被停用而拆链尚未收尾时
|
||
// 仍可能点到这里 —— 此处按唯一真相源再挡一次，避免给已关闭的设备重连
|
||
if (!this._isEnabled(device.serialNumber)) {
|
||
return { ok: false, message: '设备已停用，请先在“我的设备”页开启' };
|
||
}
|
||
|
||
this._clearInitRetry();
|
||
this._lastError = null;
|
||
this._bleReady = false;
|
||
this._setPhase(GATEWAY_PHASE.STARTING);
|
||
this._notify();
|
||
|
||
try {
|
||
await bleService.hardReconnect(device.deviceId, device.serviceUUID || null);
|
||
this._bleReady = true;
|
||
this._lastError = null;
|
||
this._setPhase(GATEWAY_PHASE.RUNNING);
|
||
console.log('[GW] 手动重连成功:', device.serialNumber);
|
||
this._notify();
|
||
return { ok: true, message: '已重新连接' };
|
||
} catch (err) {
|
||
console.error('[GW] 手动重连失败:', err);
|
||
const message = (err && err.message) || '未知错误';
|
||
this._lastError = '重连失败: ' + message;
|
||
this._setPhase(GATEWAY_PHASE.ERROR);
|
||
this._scheduleInitRetry();
|
||
this._notify();
|
||
return { ok: false, message };
|
||
}
|
||
});
|
||
}
|
||
|
||
/**
|
||
* 手动重连 MQTT（详情页 MQTT 状态点击触发）
|
||
*
|
||
* MQTT 是账号/产品级连接（凭证在「我的」页配置，clientId = S&{sn}&{productId}&{userId}），
|
||
* 与"当前采集设备"无关，因此不要求 _device 存在；
|
||
* 也不需要重新录入凭证，用已保存的配置重建即可。
|
||
*
|
||
* 失败只经返回值/toast 呈现，不写 _lastError：lastError 的清理时机绑在 BLE 状态上，
|
||
* MQTT 失败写进去会出现"BLE 一恢复就把 MQTT 报错擦掉"的不可预期呈现。
|
||
*/
|
||
retryMqtt() {
|
||
return this._chain(async () => {
|
||
if (!this._hasUser()) {
|
||
return { ok: false, message: '未登录' };
|
||
}
|
||
|
||
const cfg = storage.getMqttConfig();
|
||
if (!cfg || !cfg.serialNumber || cfg.productId == null || cfg.productId === '' || !cfg.productSecret) {
|
||
return { ok: false, message: '未配置 MQTT 凭证，请到「我的」页配置' };
|
||
}
|
||
|
||
const creds = auth.getMqttCredentials(cfg.serialNumber, cfg.productId, cfg.productSecret);
|
||
if (!creds) {
|
||
return { ok: false, message: '生成 MQTT 凭证失败，请重新登录' };
|
||
}
|
||
|
||
try {
|
||
// 硬重连：关旧客户端 → 按凭证重建；成功后 mqttClient 会回调 _onMqttState(CONNECTED)，
|
||
// 由它统一置 _mqttReady 并触发离线补传（此处不重复写状态，避免两个真相源）
|
||
await mqttClient.reconnect(creds);
|
||
console.log('[GW] 手动重连 MQTT 成功:', creds.clientId);
|
||
return { ok: true, message: '已重新连接' };
|
||
} catch (err) {
|
||
console.error('[GW] 手动重连 MQTT 失败:', err);
|
||
return { ok: false, message: (err && err.message) || '未知错误' };
|
||
}
|
||
});
|
||
}
|
||
|
||
/* ===================== 页面订阅 ===================== */
|
||
|
||
/**
|
||
* 订阅网关快照
|
||
* @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(),
|
||
// 区分"用户主动断开"与"被动掉线"：前者不该显示"点击重连"（点不点由用户决定），
|
||
// 后者才是需要一键恢复的异常态
|
||
mqttUserDisconnected: mqttClient.userDisconnected === true,
|
||
bleState: bleService.getState(),
|
||
// 浅拷贝元素，确保 setData diff 能识别 uploaded 等字段变化并触发视图刷新
|
||
dataList: this._dataList.map(it => ({ ...it })),
|
||
stats: { ...this._stats },
|
||
offlineCount: this._offlineCount,
|
||
lastError: this._lastError,
|
||
// 测点链路上下文
|
||
measurePointId: measurePoint.getMeasurePointId(),
|
||
collectMode: measurePoint.getCollectMode(),
|
||
};
|
||
}
|
||
|
||
/** 当前采集设备（浅拷贝） */
|
||
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;
|
||
// MQTT 是**应用级**连接（凭证在「我的」页、clientId 按产品/账号），
|
||
// 与"设备开关"无关：这里照旧无条件幂等补连，不因无启用设备而跳过
|
||
this._ensureMqttNow();
|
||
const enabled = storage.getEnabledDevice();
|
||
// 无启用设备 = 采集关闭：只收敛 BLE，不碰 MQTT
|
||
if (!enabled) {
|
||
await this._stopNow();
|
||
return;
|
||
}
|
||
await this._startNow(enabled);
|
||
}
|
||
|
||
/**
|
||
* 自动连接 MQTT 的内部实现
|
||
*
|
||
* 三重短路，保证幂等且不做无谓建连：
|
||
* 1. 未登录 / 用户主动断开 → 不连
|
||
* 2. 已持有客户端（已连接或正在自愈）→ 不连（canAutoConnect）
|
||
* 3. 无已保存凭证 → 不连（并保持静默，配置入口在「我的」页）
|
||
* @returns {Promise<boolean>} 是否已连接
|
||
*/
|
||
async _ensureMqttNow() {
|
||
if (!this._hasUser()) return false;
|
||
if (!mqttClient.canAutoConnect()) return mqttClient.isConnected();
|
||
if (this._mqttConnecting) return mqttClient.isConnected();
|
||
|
||
const cfg = storage.getMqttConfig();
|
||
if (!cfg || !cfg.serialNumber || cfg.productId == null || cfg.productId === '' || !cfg.productSecret) {
|
||
return false;
|
||
}
|
||
const creds = auth.getMqttCredentials(cfg.serialNumber, cfg.productId, cfg.productSecret);
|
||
if (!creds) return false;
|
||
|
||
this._mqttConnecting = true;
|
||
try {
|
||
await mqttClient.connect(creds);
|
||
console.log('[GW] MQTT 自动连接成功:', creds.clientId);
|
||
return true;
|
||
} catch (err) {
|
||
// 失败不阻断：底层会按 reconnectPeriod 自愈；连接成功会经
|
||
// _onMqttState(CONNECTED) 置位并触发离线补传
|
||
console.warn('[GW] MQTT 自动连接失败(底层将自愈重试):', err && err.message);
|
||
return false;
|
||
} finally {
|
||
this._mqttConnecting = false;
|
||
}
|
||
}
|
||
|
||
async _startNow(device) {
|
||
if (!device || !device.serialNumber) return;
|
||
// 目标必须是"唯一启用设备"：启用开关是采集的唯一真相源，
|
||
// 任何入口（扫描页 gateway.start / 失败重试）都不得替一个已停用的设备建链
|
||
if (!this._isEnabled(device.serialNumber)) {
|
||
console.log('[GW] 目标设备未启用，不启动采集:', device.serialNumber);
|
||
// 只有"它正是当前目标"时才需要收敛（清掉这个过期会话）；
|
||
// 否则只是别人传了个已停用的设备进来，不能连累当前正在跑的采集
|
||
if (this._device && this._device.serialNumber === device.serialNumber) {
|
||
await this._stopNow();
|
||
}
|
||
return;
|
||
}
|
||
this._clearInitRetry();
|
||
// 启用设备 = 开始采集：顺手确保 MQTT 在（幂等短路），
|
||
// 否则首次扫码后必须回「我的」页点一次连接才能上传
|
||
this._ensureMqttNow();
|
||
|
||
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
|
||
&& this._isEnabled(device.serialNumber));
|
||
|
||
try {
|
||
// 1. BLE：连接目标（适配器由 bleService.connectDevice 内部保证已打开，幂等；
|
||
// 已就绪且同设备则直接复用，不做无谓建链）
|
||
if (!alive()) return;
|
||
if (!bleService.isReady() || bleService.deviceId !== device.deviceId) {
|
||
// connectDevice 幂等：同设备在途连接会复用同一 Promise，
|
||
// 不再需要"判断是否连接中 → 等待 → 二次补连"的兜底分支
|
||
await bleService.connectDevice(device.deviceId, device.serviceUUID || null);
|
||
}
|
||
if (!alive()) return;
|
||
this._bleReady = true;
|
||
this._lastError = null; // 链路已就绪：清掉历史错误，避免"已连上仍显示报错"
|
||
this._setPhase(GATEWAY_PHASE.RUNNING);
|
||
this._notify();
|
||
// MQTT 由「我的」页独立连接：此处不再主动连接，
|
||
// 连接状态经 mqttClient.onStateChange 回调同步到 _mqttReady
|
||
console.log('[GW] 采集器运行中(BLE):', device.serialNumber);
|
||
} catch (err) {
|
||
if (!alive()) return;
|
||
// 切设备/主动断开导致在途连接作废：不是失败，不上屏也不重试
|
||
if (err && err.cancelled) return;
|
||
console.error('[GW] 启动失败:', err);
|
||
|
||
// 提供更友好的错误提示
|
||
let errorMsg = '启动失败';
|
||
if (err.message) {
|
||
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();
|
||
// 错误也要推给订阅者：此前只置位不通知，"异常"相位永远到不了页面
|
||
// （mine 的「异常」标签与 detail 的 {{lastError}} 都靠这次通知）
|
||
this._notify();
|
||
}
|
||
}
|
||
|
||
async _teardown() {
|
||
this._clearInitRetry();
|
||
this._flush(); // 先处理缓冲（此时 _device 仍在）
|
||
this._bleReady = false;
|
||
// MQTT 连接由「我的」页建立并持有，网关只复用其状态：
|
||
// 此处不得 disconnect（否则切设备/停止采集会把 MQTT 拆掉，且 _boot 不会重连），
|
||
// _mqttReady 也保持真实值（由 _onMqttState 驱动），退出登录统一由 app.clearLoginInfo 断开
|
||
this._device = null; // 先置空，避免断开回调把 phase 打回 starting
|
||
try { await bleService.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);
|
||
// 链路已恢复：清掉历史错误，否则一次瞬时失败会一直挂在详情页 {{lastError}} 上
|
||
this._lastError = null;
|
||
} 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 mode = measurePoint.getCollectMode();
|
||
const measurePointId = measurePoint.getMeasurePointId();
|
||
// 上报身份在收帧批次内快照：手持采集仪逐点测量，确认/补传必须用「采集当时」的测点，
|
||
// 否则「A点采完→切到B点→再确认」会把 A 点的数据归属到 B 点
|
||
const productId = this._resolveProductId();
|
||
const reportSn = this._resolveSerialNumber();
|
||
const offline = [];
|
||
for (const item of batch) {
|
||
item.mode = mode;
|
||
item.measurePointId = measurePointId;
|
||
item.productId = productId;
|
||
item.reportSn = reportSn;
|
||
if (mode === measurePoint.COLLECT_MODE.MANUAL) {
|
||
// 手动确认模式：待确认，不自动上报、不进离线缓存（确认后由 confirmData 处理）
|
||
item.confirmed = false;
|
||
continue;
|
||
}
|
||
// 自动模式
|
||
if (!measurePoint.isValidMeasurePointId(measurePointId)) {
|
||
// 测点ID 缺失或非法：不上报不缓存，标记原因（topic 第3段非法会导致服务端分段错位）
|
||
item.pendingReason = measurePointId ? '测点ID格式非法' : '未设置测点ID';
|
||
continue;
|
||
}
|
||
if (this._mqttReady && device && device.enabled) {
|
||
this._upload(item); // fire-and-forget
|
||
} else {
|
||
item.offline = true;
|
||
offline.push(this._toOfflineItem(item, device));
|
||
}
|
||
}
|
||
if (offline.length > 0) {
|
||
storage.cachePendingDataBatch(offline);
|
||
this._refreshOfflineCount();
|
||
}
|
||
this._notify();
|
||
}
|
||
|
||
/**
|
||
* 解析上报用产品ID：优先取“我的”页 MQTT 配置（单一数据源），
|
||
* 设备配置不再录入产品信息，无 MQTT 配置时回退设备自身 productId（兼容旧数据）
|
||
* @returns {number|string|null}
|
||
*/
|
||
_resolveProductId() {
|
||
const cfg = storage.getMqttConfig();
|
||
if (cfg && cfg.productId != null && cfg.productId !== '') {
|
||
return cfg.productId;
|
||
}
|
||
return this._device ? this._device.productId : null;
|
||
}
|
||
|
||
/**
|
||
* 解析上报用设备编号：优先取“我的”页 MQTT 配置的 SN（与 clientId 依据一致），
|
||
* 回退设备自身 serialNumber（BLE 名称自动生成的编号，用于本地标识）
|
||
*
|
||
* 不统一会导致：topic 里的 SN 与 MQTT 凭证 SN 不一致，后端按 topic 的 SN 查不到设备
|
||
* @returns {string|null}
|
||
*/
|
||
_resolveSerialNumber() {
|
||
const cfg = storage.getMqttConfig();
|
||
if (cfg && cfg.serialNumber) {
|
||
return cfg.serialNumber;
|
||
}
|
||
return this._device ? this._device.serialNumber : null;
|
||
}
|
||
|
||
/**
|
||
* 手动确认上报（模式1）：确认单条数据并上传
|
||
* @param {string} itemId - 数据项 id
|
||
* @returns {Promise}
|
||
*/
|
||
confirmData(itemId) {
|
||
const item = this._dataList.find(it => it.id === itemId);
|
||
if (!item) {
|
||
return Promise.reject(new Error('数据不存在'));
|
||
}
|
||
if (item.uploaded || item.confirmed) {
|
||
return Promise.resolve();
|
||
}
|
||
|
||
const device = this._device;
|
||
// 归属一律取收帧快照：手动确认可能发生在切换测点之后
|
||
const measurePointId = item.measurePointId || '';
|
||
if (!measurePoint.isValidMeasurePointId(measurePointId)) {
|
||
item.pendingReason = measurePointId ? '测点ID格式非法' : '未设置测点ID';
|
||
this._notify();
|
||
return Promise.reject(new Error('请先填写合法测点ID(纯数字, 1-64位)'));
|
||
}
|
||
// 产品ID 允许回退：手动模式下常出现「先采集、后到我的页配凭证、再确认」
|
||
const productId = item.productId || this._resolveProductId();
|
||
if (productId == null || productId === '') {
|
||
item.pendingReason = '未配置产品ID';
|
||
this._notify();
|
||
return Promise.reject(new Error('请先在“我的”页配置 MQTT 产品凭证'));
|
||
}
|
||
|
||
item.confirmed = true;
|
||
item.pendingReason = null;
|
||
|
||
if (this._mqttReady && device && device.enabled) {
|
||
this._upload(item);
|
||
} else if (device) {
|
||
// MQTT 未连接：确认后进离线缓存，待补传
|
||
storage.cachePendingData(this._toOfflineItem(item, device));
|
||
this._refreshOfflineCount();
|
||
item.offline = true;
|
||
}
|
||
this._notify();
|
||
return Promise.resolve();
|
||
}
|
||
|
||
_upload(item) {
|
||
const device = this._device;
|
||
if (!device) return;
|
||
const measurePointId = item.measurePointId || '';
|
||
if (!measurePoint.isValidMeasurePointId(measurePointId)) {
|
||
item.pendingReason = measurePointId ? '测点ID格式非法' : '未设置测点ID';
|
||
this._notify();
|
||
return;
|
||
}
|
||
const productId = item.productId || this._resolveProductId();
|
||
if (productId == null || productId === '') {
|
||
item.pendingReason = '未配置产品ID';
|
||
this._notify();
|
||
return;
|
||
}
|
||
const topic = auth.buildMeasurePostTopic(
|
||
productId,
|
||
item.reportSn || this._resolveSerialNumber(),
|
||
measurePointId
|
||
);
|
||
mqttClient.publish(topic, item.hex)
|
||
.then(() => {
|
||
item.uploaded = true;
|
||
item.offline = false;
|
||
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();
|
||
item.offline = true;
|
||
}
|
||
this._notify();
|
||
});
|
||
}
|
||
|
||
_toOfflineItem(item, device) {
|
||
if (!device) return null;
|
||
return {
|
||
deviceId: device.serialNumber, // 设备身份（本地标识，用于缓存归属匹配）
|
||
reportSn: item.reportSn || this._resolveSerialNumber(), // 上报身份（补传时用于构建 topic）
|
||
measurePointId: item.measurePointId || '', // 采集当时的测点，补传不得重新解析
|
||
hex: item.hex,
|
||
dataId: item.id,
|
||
};
|
||
}
|
||
|
||
/** 补传当前设备的离线缓存（成功后仅移除本设备项，其他设备残留保留） */
|
||
async _flushOffline() {
|
||
if (this._flushingOffline) return;
|
||
this._flushingOffline = true;
|
||
try {
|
||
const device = this._device;
|
||
const productId = this._resolveProductId();
|
||
if (!device || !this._mqttReady || !productId) return;
|
||
const list = storage.getPendingData();
|
||
if (!list || list.length === 0) return;
|
||
|
||
const remaining = [];
|
||
for (const it of list) {
|
||
if (it.deviceId === device.serialNumber) {
|
||
if (!measurePoint.isValidMeasurePointId(it.measurePointId)) {
|
||
// 旧格式缓存（无测点ID）或非法测点ID 无法按测点上报，保留待人工清理
|
||
console.warn('[GW] 离线缓存测点ID缺失或非法，跳过:', it.measurePointId, it.hex);
|
||
remaining.push(it);
|
||
continue;
|
||
}
|
||
// 补传用缓存项当时的上报身份，缺失则回退当前解析结果
|
||
const sn = it.reportSn || this._resolveSerialNumber() || device.serialNumber;
|
||
const topic = auth.buildMeasurePostTopic(productId, sn, it.measurePointId);
|
||
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);
|
||
}
|
||
|
||
/**
|
||
* 目标设备当前是否仍是"唯一启用设备"
|
||
*
|
||
* `enabled` 是采集开关的唯一真相源：设备一旦停用就不得再建链、不得再排自动重试。
|
||
* @param {string} serialNumber
|
||
* @returns {boolean}
|
||
*/
|
||
_isEnabled(serialNumber) {
|
||
if (!serialNumber) return false;
|
||
const enabled = storage.getEnabledDevice();
|
||
return !!(enabled && enabled.serialNumber === serialNumber);
|
||
}
|
||
|
||
/**
|
||
* 同步"存储中的启用设备"与"网关当前目标"，不一致时**立即**中止在途 BLE 工作
|
||
*
|
||
* 必须同步执行（不能只靠队列里的 _stopNow）：_stopNow 要排在在途的 _boot 之后，
|
||
* 最坏要等一次建链超时（10s）+ 一次适配器复位，而这期间在途流程还会继续
|
||
* 复位适配器、上报启动失败、排自动重试 —— 表现就是"设备都关了还在一直连蓝牙"。
|
||
* bleService.disconnect() 的置位、序号自增与计时器清理都是同步的，
|
||
* 立刻作废在途连接流程并停掉自动重连；队列里的 _teardown 只负责收尾。
|
||
*/
|
||
_syncTarget() {
|
||
const enabled = storage.getEnabledDevice();
|
||
const next = enabled ? enabled.serialNumber : null;
|
||
const current = this._device ? this._device.serialNumber : null;
|
||
if (!current || current === next) return;
|
||
console.log('[GW] 目标设备已停用，立即中止连接:', current);
|
||
this._clearInitRetry();
|
||
bleService.disconnect().catch(() => { /* 无链路/已断开：无需处理 */ });
|
||
}
|
||
|
||
_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] 启动失败，自动重试...');
|
||
// 重试必须重新解析"唯一启用设备"：直接 start(this._device) 会绕过启用开关，
|
||
// 设备已被关掉时仍会继续重连。ensureStarted 在无启用设备时会直接收敛到停止
|
||
this.ensureStarted();
|
||
}, 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');
|
||
const ms = date.getMilliseconds().toString().padStart(3, '0');
|
||
return `${h}:${m}:${s}.${ms}`;
|
||
}
|
||
}
|
||
|
||
// 导出单例
|
||
const gateway = new Gateway();
|
||
|
||
module.exports = {
|
||
GATEWAY_PHASE,
|
||
GATEWAY_PHASE_TEXT,
|
||
Gateway,
|
||
gateway,
|
||
};
|