import { fetchEventSource } from '@microsoft/fetch-event-source' import { ElMessage } from 'element-plus' /** * SSE 连接客户端 * * 只负责连接生命周期(建连、断开、重连、心跳),不包含任何业务逻辑。 * 收到消息后通过 onMessage 回调抛出,由业务层自行分发。 */ export class SseClient { private abortController: AbortController | null = null private retryCount = 0 private _connected = false private _connecting = false /** 重连配置 */ private readonly maxRetries: number private readonly baseRetryDelayMs: number private readonly maxRetryDelayMs: number constructor( private readonly url: string, private readonly getToken: () => string | null, private readonly callbacks: SseClientCallbacks, opts?: SseClientOptions ) { this.maxRetries = opts?.maxRetries ?? 5 this.baseRetryDelayMs = opts?.baseRetryDelayMs ?? 3000 this.maxRetryDelayMs = opts?.maxRetryDelayMs ?? 30000 } get connected(): boolean { return this._connected } get connecting(): boolean { return this._connecting } /** * 建立 SSE 连接 * * 幂等调用:已连接或正在连接则跳过。 * 连接断开时自动指数退避重连,最多 maxRetries 次。 */ async connect(): Promise { const token = this.getToken() if (!token) { this.callbacks.onError?.('缺少 accessToken') return } // 幂等:相同 token 已连接或正在连接则跳过 if (this._connected || this._connecting) { return } this.disconnect() this._connecting = true this.abortController = new AbortController() try { await fetchEventSource(this.url, { headers: { saToken: token }, signal: this.abortController.signal, openWhenHidden: true, onopen: async (response) => { const contentType = response.headers.get('content-type') || '' if (response.status === 401) { throw new SseHttpError('SSE 鉴权失败', response) } if (!response.ok) { throw new SseHttpError(`SSE 连接失败: ${response.status}`, response) } if (!contentType.includes('text/event-stream')) { throw new SseHttpError(`非法的响应类型: ${contentType || '空'}`, response) } this._connecting = false this._connected = true this.retryCount = 0 this.callbacks.onConnected?.() }, onmessage: (message) => { if (!message.event || !message.data) { return } this.callbacks.onMessage(message.event, message.data) }, onclose: () => { this._connecting = false this._connected = false }, onerror: (error) => { this._connecting = false this._connected = false // 401 → 通知外部刷新 token if (isSseHttpError(error) && error.response.status === 401) { this.callbacks.onUnauthorized?.() throw error } this.retryCount += 1 if (this.retryCount >= this.maxRetries) { this.callbacks.onError?.('实时连接异常,请刷新页面') throw error } const delay = Math.min( this.baseRetryDelayMs * Math.pow(2, this.retryCount - 1), this.maxRetryDelayMs ) return delay } }) } catch (error) { // fetchEventSource 仅在重连耗尽或 401 时 throw,状态已在 onerror 中处理 } } /** 断开连接(主动关闭,不重连) */ disconnect(): void { if (this.abortController) { this.abortController.abort() this.abortController = null } this.retryCount = 0 this._connected = false this._connecting = false } } // ==================== 类型定义 ==================== export interface SseClientCallbacks { /** 连接成功 */ onConnected?: () => void /** 收到消息 (eventName, rawData) */ onMessage: (event: string, data: string) => void /** 连接异常(重连耗尽或 token 缺失) */ onError?: (message: string) => void /** 收到 401,需刷新 token */ onUnauthorized?: () => void } export interface SseClientOptions { maxRetries?: number baseRetryDelayMs?: number maxRetryDelayMs?: number } // ==================== 内部工具 ==================== class SseHttpError extends Error { constructor( message: string, readonly response: Response ) { super(message) this.name = 'SseHttpError' } } function isSseHttpError(error: unknown): error is SseHttpError { return error instanceof SseHttpError }