222 lines
7.0 KiB
TypeScript
222 lines
7.0 KiB
TypeScript
/**
|
||
* 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: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:pong',
|
||
'talk:error',
|
||
]
|
||
|
||
/** 心跳间隔（毫秒） */
|
||
const HEARTBEAT_INTERVAL = 25000
|
||
|
||
/** 单例连接与令牌 */
|
||
let SocketInstance: Socket | null = null
|
||
let CurrentToken = ''
|
||
/** 心跳定时器 */
|
||
let HeartbeatTimer: ReturnType<typeof setInterval> | null = null
|
||
/** 是否已针对 401 重连过一次（防止无限刷新令牌） */
|
||
let ReauthAttempted = false
|
||
/** 本次连接周期是否已提示过连接失败（重连期间不重复弹提示） */
|
||
let ConnectErrorReported = false
|
||
/** 连接中 Promise（并发调用去重） */
|
||
let Connecting: Promise<Socket | null> | null = null
|
||
|
||
/** 事件订阅表（与连接解耦，可在连接前注册） */
|
||
const EventHandlers = new Map<TalkServerEvent, Set<TalkEventHandler>>()
|
||
|
||
/** 是否为统一响应包装 */
|
||
function IsWrapped(payload: unknown): payload is { code: number; msg?: string; data?: unknown } {
|
||
return !!payload && typeof payload === 'object' && 'code' in (payload as Record<string, unknown>)
|
||
}
|
||
|
||
/** 分发服务端事件到已注册处理器 */
|
||
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<void> {
|
||
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<Socket | null> {
|
||
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<TalkEventHandler>()
|
||
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
|
||
}
|