/** * WebSocket 连接管理服务 * * 单例模式,管理与后端的 WebSocket 长连接,提供: * - JWT Token 认证连接 * - 心跳保活(30s 间隔) * - 断线自动重连(指数退避 1s→2s→4s→8s→30s max) * - 事件分发:on(event, callback) / off(event, callback) * * 对应后端:GET /ws?token=xxx */ import { BASE_URL } from '@/utils/request' import { getToken } from '@/utils/storage' // WebSocket 基础地址推导: // - 若 BASE_URL 非空(生产 CDN 直连场景),将其协议从 http(s) 替换成 ws(s) // - 若 BASE_URL 为空(开发 vite proxy / 生产同源 Nginx),用当前页面 origin 推导同源 ws(s) // 这样 WebSocket 天然随页面协议走:HTTPS 页面用 wss,HTTP 页面用 ws,避免混合内容被浏览器阻断 const resolveWsBase = () => { if (BASE_URL) return BASE_URL.replace(/^http/, 'ws') // #ifdef H5 if (typeof window !== 'undefined' && window.location) { const wsProto = window.location.protocol === 'https:' ? 'wss:' : 'ws:' return `${wsProto}//${window.location.host}` } // #endif return '' } const WS_BASE = resolveWsBase() const HEARTBEAT_INTERVAL = 30000 const RECONNECT_BASE_DELAY = 1000 const RECONNECT_MAX_DELAY = 30000 const MAX_RECONNECT_ATTEMPTS = Infinity const ACK_DEFAULT_TIMEOUT_MS = 10000 /** 异常重连检测:10s 窗口内 >=3 次断连即认为异常抖动(常见于同账号多 tab 互踢) */ const FLAPPING_WINDOW_MS = 10000 const FLAPPING_THRESHOLD = 3 /** 两次 flapping 警告之间最小间隔,避免持续抖动时每秒触发 */ const FLAPPING_NOTIFY_COOLDOWN_MS = 15000 class WebSocketService { constructor() { this.ws = null this.listeners = new Map() this.heartbeatTimer = null this.reconnectTimer = null this.reconnectAttempts = 0 this.isManualClose = false this.seq = 0 /** * 等待 ACK 的请求索引:Map * Task 9 引入,为 meeting.* 事件提供 Promise 化 ACK 等待 */ this.pendingAcks = new Map() /** 近期断连时间戳数组,用于 flapping 检测 */ this.recentCloseAt = [] /** 上次 flapping 警告时间,用于冷却 */ this.lastFlappingNotifyAt = 0 } /** * 建立 WebSocket 连接 * @param {string} [token] - JWT Token,不传则从本地存储获取 */ connect(token) { if (this.ws && (this.ws.readyState === WebSocket.OPEN || this.ws.readyState === WebSocket.CONNECTING)) { return } const accessToken = token || getToken() if (!accessToken) { console.warn('[WS] 无 Token,无法连接') return } this.isManualClose = false const url = `${WS_BASE}/ws?token=${accessToken}` try { // #ifdef H5 this.ws = new WebSocket(url) // #endif // #ifndef H5 this.ws = uni.connectSocket({ url, complete: () => {} }) // #endif this._bindEvents() } catch (err) { console.error('[WS] 连接创建失败', err) this._scheduleReconnect() } } /** 主动断开连接 */ disconnect() { this.isManualClose = true this._clearTimers() if (this.ws) { // #ifdef H5 this.ws.close() // #endif // #ifndef H5 uni.closeSocket() // #endif this.ws = null } } /** * 发送消息 * @param {string} event - 事件类型 * @param {Object} [data] - 业务数据 * @returns {number} 消息序号 */ send(event, data = {}) { // #ifdef H5 if (!this.ws || this.ws.readyState !== WebSocket.OPEN) { console.warn('[WS] 连接未就绪,无法发送') return -1 } // #endif // #ifndef H5 if (!this.ws) { console.warn('[WS] 连接未就绪,无法发送') return -1 } // #endif this.seq++ const msg = JSON.stringify({ event, seq: this.seq, data, time: new Date().toISOString() }) // #ifdef H5 this.ws.send(msg) // #endif // #ifndef H5 uni.sendSocketMessage({ data: msg }) // #endif return this.seq } /** * 发送消息并等待服务端 ACK(Task 9 引入) * * 服务端对 C→S 事件以相同 event 名 + 对应 seq 返回 ACK 报文: * { event, seq, data: { code, message, data? } } * * 约定: * - code === 0 表示成功,Promise resolve 为 data.data(业务 payload) * - code !== 0 表示业务失败,Promise reject 为 { code, message } * - 超时 reject 为 { code: -1, message: 'ACK 超时' } * - 连接未就绪 reject 为 { code: -1, message: '连接未就绪' } * * 典型用法: * const info = await wsService.sendWithAck('meeting.transport.create', { room_code, direction: 'send' }) * * @param {string} event - 事件名 * @param {Object} [data] - 业务 payload * @param {number} [timeoutMS=10000] - 超时时间 * @returns {Promise} ACK 的 data 字段 */ sendWithAck(event, data = {}, timeoutMS = ACK_DEFAULT_TIMEOUT_MS) { return new Promise((resolve, reject) => { const seq = this.send(event, data) if (seq === -1) { reject({ code: -1, message: '连接未就绪' }) return } const timer = setTimeout(() => { this.pendingAcks.delete(seq) reject({ code: -1, message: `ACK 超时 (${event})` }) }, timeoutMS) this.pendingAcks.set(seq, { resolve, reject, timer, event }) }) } /** * 当前连接是否处于可发送状态(H5 下要求 readyState === OPEN) * 与 send() 的就绪判断保持一致 * @returns {boolean} */ isConnected() { // #ifdef H5 return !!(this.ws && this.ws.readyState === WebSocket.OPEN) // #endif // #ifndef H5 return !!this.ws // #endif } /** * 等待 WebSocket 进入可用(OPEN)状态 * * 场景:SSO / web-view 登录后 connect() 为异步建链(CONNECTING→OPEN)。 * 设备预览结束立即入会时,首个 sendWithAck 可能早于 socket OPEN, * 导致 send() 返回 -1 → 业务侧抛 "连接未就绪"。 * 本方法在未连接时主动发起一次连接(connect 对 OPEN/CONNECTING 幂等), * 并等待 _connected 事件;已连接则立即 resolve;超时 reject。 * * @param {number} [timeoutMS=10000] - 最长等待时间 * @returns {Promise} */ waitForConnected(timeoutMS = 10000) { if (this.isConnected()) return Promise.resolve(true) // 未建链 / 已断开:主动发起一次连接(幂等,不会重复建链) this.connect() return new Promise((resolve, reject) => { let settled = false let timer = null let poll = null const cleanup = () => { this.off('_connected', onConnected) if (timer) clearTimeout(timer) if (poll) clearInterval(poll) } const onConnected = () => { if (settled) return settled = true cleanup() resolve(true) } timer = setTimeout(() => { if (settled) return settled = true cleanup() reject({ code: -1, message: '连接会议服务器超时' }) }, timeoutMS) // 轮询兜底:防止 _onOpen 在 listener 注册前已触发而错过事件 poll = setInterval(() => { if (this.isConnected()) onConnected() }, 100) this.on('_connected', onConnected) }) } /** * 注册事件监听 * @param {string} event - 事件类型 * @param {Function} callback - 回调函数 */ on(event, callback) { if (!this.listeners.has(event)) { this.listeners.set(event, new Set()) } this.listeners.get(event).add(callback) } /** * 移除事件监听 * @param {string} event - 事件类型 * @param {Function} callback - 回调函数 */ off(event, callback) { const cbs = this.listeners.get(event) if (cbs) { cbs.delete(callback) } } /** 绑定 WebSocket 事件处理 */ _bindEvents() { // #ifdef H5 this.ws.onopen = () => this._onOpen() this.ws.onclose = (e) => this._onClose(e) this.ws.onerror = (e) => this._onError(e) this.ws.onmessage = (e) => this._onMessage(e.data) // #endif // #ifndef H5 uni.onSocketOpen(() => this._onOpen()) uni.onSocketClose((e) => this._onClose(e)) uni.onSocketError((e) => this._onError(e)) uni.onSocketMessage((e) => this._onMessage(e.data)) // #endif } _onOpen() { console.log('[WS] 连接成功') this.reconnectAttempts = 0 this._startHeartbeat() this._emit('_connected') } _onClose(e) { console.log('[WS] 连接关闭', e) this._clearTimers() this._rejectAllPendingAcks('连接已断开') this._emit('_disconnected', e) if (!this.isManualClose) { this._recordCloseForFlapping() this._scheduleReconnect() } } /** * 记录一次非主动断连,并在短窗口内多次断连时触发 _flapping 事件 * 典型场景:同账号多端登录互踢导致 WS 每秒被后端关闭 */ _recordCloseForFlapping() { const now = Date.now() this.recentCloseAt.push(now) // 丢掉窗口外的历史记录 this.recentCloseAt = this.recentCloseAt.filter((t) => now - t <= FLAPPING_WINDOW_MS) if (this.recentCloseAt.length < FLAPPING_THRESHOLD) return if (now - this.lastFlappingNotifyAt < FLAPPING_NOTIFY_COOLDOWN_MS) return this.lastFlappingNotifyAt = now const count = this.recentCloseAt.length console.warn(`[WS] 检测到异常重连抖动:${FLAPPING_WINDOW_MS / 1000}s 内断开 ${count} 次,可能同账号在其他 tab/设备登录`) this._emit('_flapping', { count, windowMs: FLAPPING_WINDOW_MS }) } /** * 连接断开时统一拒绝所有未完成的 ACK 等待(Task 9) * 避免业务层 await sendWithAck() 永久挂起 */ _rejectAllPendingAcks(reason) { if (this.pendingAcks.size === 0) return this.pendingAcks.forEach((pending, seq) => { clearTimeout(pending.timer) pending.reject({ code: -1, message: reason, event: pending.event }) }) this.pendingAcks.clear() } _onError(e) { console.error('[WS] 连接错误', e) } _onMessage(data) { try { const msg = JSON.parse(data) // Task 9:ACK 报文(event 以 ".ack" 结尾)会先走 pendingAcks 处理 // 后端 ws.NewResponse 固定在原 event 后追加 ".ack" // // 修复(2026-04-23):历史实现在命中 _handleAck 后直接 return, // 导致通过 on('xxx.ack') 订阅 ACK 事件的 listener(chat store 的 // _onSendACK / _onReadACK)永远收不到回调,表现为 // "发送方消息一直卡在 _sending=true,气泡下方不显示 已读/未读"。 // 两条通路互不冲突:sendWithAck() 按 seq 命中 pending; // on('xxx.ack') 按 event 命中 listener。此处改为并行分发。 if (typeof msg.event === 'string' && msg.event.endsWith('.ack')) { this._handleAck(msg) } this._emit(msg.event, msg) } catch (e) { console.warn('[WS] 消息解析失败', data) } } /** * 处理 ACK 报文,触发 pendingAcks 中对应 seq 的 resolve/reject * @param {Object} msg - { event: 'xxx.ack', seq, code, message, data? } */ _handleAck(msg) { const pending = this.pendingAcks.get(msg.seq) if (!pending) { return } this.pendingAcks.delete(msg.seq) clearTimeout(pending.timer) if (msg.code === 0) { pending.resolve(msg.data) } else { pending.reject({ code: msg.code, message: msg.message || 'WS 请求失败', event: pending.event }) } } /** 分发事件给注册的监听器 */ _emit(event, data) { const cbs = this.listeners.get(event) if (cbs) { cbs.forEach(cb => { try { cb(data) } catch (e) { console.error(`[WS] 事件处理器异常: ${event}`, e) } }) } } /** 启动心跳 */ _startHeartbeat() { this._clearTimers() this.heartbeatTimer = setInterval(() => { this.send('heartbeat') }, HEARTBEAT_INTERVAL) } /** 调度重连(指数退避) */ _scheduleReconnect() { if (this.reconnectAttempts >= MAX_RECONNECT_ATTEMPTS) { console.error('[WS] 超过最大重连次数') return } const delay = Math.min( RECONNECT_BASE_DELAY * Math.pow(2, this.reconnectAttempts), RECONNECT_MAX_DELAY ) console.log(`[WS] ${delay}ms 后重连 (第 ${this.reconnectAttempts + 1} 次)`) this.reconnectTimer = setTimeout(() => { this.reconnectAttempts++ this.connect() }, delay) } _clearTimers() { if (this.heartbeatTimer) { clearInterval(this.heartbeatTimer) this.heartbeatTimer = null } if (this.reconnectTimer) { clearTimeout(this.reconnectTimer) this.reconnectTimer = null } } } /** 全局单例 */ const wsService = new WebSocketService() export default wsService