/** * Talk 实时通道 · Socket.IO 连接管理(单例) * * 生命周期: * - connectTalkSocket() 拉取 SocketToken 后连接 /talk 命名空间(SocketPath 由后端下发) * - 断线由 socket.io 自动重连(1s 起,最大 5s);重连成功后重新鉴权并重启心跳 * - talk:error 且 code===401 时重新拉取 SocketToken 并重连一次 * - 页面销毁调用 disconnectTalkSocket() 断开并清理订阅 * * 事件分发:服务端响应统一 { code, msg, data }。成功(code===200)时向订阅者 * 仅传递 data;talk:error 或非成功码时传递完整响应,便于上层读取 code/msg。 */ import { io, type Socket } from 'socket.io-client' import { getTalkSocketToken } from '@/api/talk' /** 服务端 → 客户端事件名 */ export type TalkServerEvent = | 'talk:ready' | 'talk:message' | 'talk:read' | 'talk:typing' | 'talk:conversation' | 'talk:pong' | 'talk:error' /** 事件处理器(payload 为已解包数据;talk:error 为原始 { code, msg }) */ export type TalkEventHandler = (payload: unknown) => void /** 服务端事件全集(连接时统一挂载分发) */ const SERVER_EVENTS: TalkServerEvent[] = [ 'talk:ready', 'talk:message', 'talk:read', 'talk:typing', 'talk:conversation', 'talk:pong', 'talk:error', ] /** 心跳间隔(毫秒) */ const HEARTBEAT_INTERVAL = 25000 /** 单例连接与令牌 */ let SocketInstance: Socket | null = null let CurrentToken = '' /** 心跳定时器 */ let HeartbeatTimer: ReturnType | null = null /** 是否已针对 401 重连过一次(防止无限刷新令牌) */ let ReauthAttempted = false /** 本次连接周期是否已提示过连接失败(重连期间不重复弹提示) */ let ConnectErrorReported = false /** 连接中 Promise(并发调用去重) */ let Connecting: Promise | null = null /** 事件订阅表(与连接解耦,可在连接前注册) */ const EventHandlers = new Map>() /** 是否为统一响应包装 */ function IsWrapped(payload: unknown): payload is { code: number; msg?: string; data?: unknown } { return !!payload && typeof payload === 'object' && 'code' in (payload as Record) } /** 分发服务端事件到已注册处理器 */ function Dispatch(event: TalkServerEvent, payload: unknown): void { // 401:令牌过期,刷新后重连一次 if (event === 'talk:error' && IsWrapped(payload) && payload.code === 401) { void HandleUnauthorized() } let out: unknown = payload if (IsWrapped(payload) && event !== 'talk:error' && payload.code === 200) { out = payload.data } const set = EventHandlers.get(event) if (!set) return for (const handler of set) { try { handler(out) } catch { // 单个处理器异常不影响其他订阅者 } } } /** 启动心跳(talk:ping) */ function StartHeartbeat(socket: Socket): void { StopHeartbeat() HeartbeatTimer = setInterval(() => { socket.emit('talk:ping') }, HEARTBEAT_INTERVAL) } /** 停止心跳 */ function StopHeartbeat(): void { if (HeartbeatTimer) { clearInterval(HeartbeatTimer) HeartbeatTimer = null } } /** 401 处理:重新拉取 SocketToken 并重连一次 */ async function HandleUnauthorized(): Promise { if (ReauthAttempted || !SocketInstance) return ReauthAttempted = true try { const token = await getTalkSocketToken() CurrentToken = token.Token SocketInstance.auth = { Token: token.Token } SocketInstance.disconnect() SocketInstance.connect() } catch { // 令牌刷新失败:保留自动重连,等待后续 connect_error 再试 } } /** 挂载连接层核心监听(连接态、重连、心跳、事件分发) */ function BindCoreEvents(socket: Socket): void { socket.on('connect', () => { ReauthAttempted = false ConnectErrorReported = false // 命名空间鉴权:auth.Token 已携带,仍按契约显式发一次 talk:auth socket.emit('talk:auth', { Token: CurrentToken }) StartHeartbeat(socket) }) socket.on('disconnect', () => { StopHeartbeat() }) // 连接建立失败(如 CORS 被拒、网络不可达):把原因透给页面,避免"什么都看不到" socket.on('connect_error', (error: Error) => { if (ConnectErrorReported) return ConnectErrorReported = true Dispatch('talk:error', { code: 0, msg: `Talk 实时通道连接失败:${error?.message || '未知原因'}(消息将不会实时到达)`, }) }) // 重连前刷新 auth,保证使用最新令牌 socket.io.on('reconnect_attempt', () => { socket.auth = { Token: CurrentToken } }) for (const event of SERVER_EVENTS) { socket.on(event, (payload: unknown) => Dispatch(event, payload)) } } /** * 连接 Talk 实时通道(幂等):已连接则直接返回;连接中则复用同一 Promise。 * 失败返回 null(socket.io 仍按配置自动重连)。 */ export async function connectTalkSocket(): Promise { if (SocketInstance?.connected) return SocketInstance if (Connecting) return Connecting Connecting = (async () => { try { const token = await getTalkSocketToken() if (!token?.Token) return SocketInstance CurrentToken = token.Token // 已有实例:仅刷新令牌并按需重连,避免重复创建连接 if (SocketInstance) { SocketInstance.auth = { Token: token.Token } if (!SocketInstance.connected) SocketInstance.connect() return SocketInstance } const socket = io(token.SocketPath || '/talk', { path: '/socket.io', withCredentials: true, transports: ['websocket', 'polling'], auth: { Token: token.Token }, reconnection: true, reconnectionDelay: 1000, reconnectionDelayMax: 5000, }) BindCoreEvents(socket) SocketInstance = socket return socket } catch { return null } finally { Connecting = null } })() return Connecting } /** 订阅服务端事件,返回取消订阅函数 */ export function subscribeTalkEvent(event: TalkServerEvent, handler: TalkEventHandler): () => void { let set = EventHandlers.get(event) if (!set) { set = new Set() EventHandlers.set(event, set) } set.add(handler) const target = set return () => { target.delete(handler) } } /** 发送客户端事件(未连接时返回 false,由调用方走 REST 兜底) */ export function emitTalkEvent(event: string, payload?: unknown): boolean { if (!SocketInstance?.connected) return false SocketInstance.emit(event, payload) return true } /** 当前实时通道是否已连接 */ export function isTalkSocketConnected(): boolean { return !!SocketInstance?.connected } /** 断开并清理(页面销毁时调用) */ export function disconnectTalkSocket(): void { StopHeartbeat() EventHandlers.clear() if (SocketInstance) { SocketInstance.removeAllListeners() SocketInstance.disconnect() SocketInstance = null } ReauthAttempted = false ConnectErrorReported = false CurrentToken = '' Connecting = null }