feat: 重新实现SseClient
This commit is contained in:
@@ -1,184 +1,83 @@
|
||||
import { ref } from 'vue'
|
||||
import { fetchEventSource } from '@microsoft/fetch-event-source'
|
||||
import { sseBus } from '@/stores/sseBus'
|
||||
import { SseClient } from '@/common/sse/SseClient'
|
||||
import type { SseEvents } from '@/stores/sseBus'
|
||||
import { sseBus } from '@/stores/sseBus'
|
||||
import { useAccountStore } from '@/stores/modules/account'
|
||||
|
||||
/** SSE 连接 URL */
|
||||
const SSE_URL = '/api/sse/subscribe'
|
||||
const MAX_RETRY_COUNT = 5
|
||||
const BASE_RETRY_DELAY_MS = 3000
|
||||
const MAX_RETRY_DELAY_MS = 30000
|
||||
|
||||
const isConnected = ref(false)
|
||||
let abortController: AbortController | null = null
|
||||
let currentToken: string | null = null
|
||||
let connecting = false
|
||||
let retryCount = 0
|
||||
/** 是否正在刷新 token（防并发） */
|
||||
let isRefreshing = false
|
||||
|
||||
class SseHttpError extends Error {
|
||||
constructor(
|
||||
message: string,
|
||||
readonly response: Response
|
||||
) {
|
||||
super(message)
|
||||
this.name = 'SseHttpError'
|
||||
// ==================== 模块级单例 ====================
|
||||
|
||||
const isConnected = ref(false)
|
||||
|
||||
const client = new SseClient(
|
||||
SSE_URL,
|
||||
() => useAccountStore().accessToken,
|
||||
{
|
||||
onConnected() {
|
||||
isConnected.value = true
|
||||
sseBus.emit('connected', undefined)
|
||||
},
|
||||
onMessage(event: string, data: string) {
|
||||
try {
|
||||
const payload = JSON.parse(data)
|
||||
sseBus.emit(event as keyof SseEvents, payload)
|
||||
} catch {
|
||||
console.warn('[SSE] 事件数据解析失败:', event, data)
|
||||
}
|
||||
},
|
||||
onError(message: string) {
|
||||
sseBus.emit('error', { message, retryCount: 0 })
|
||||
},
|
||||
onUnauthorized() {
|
||||
void refreshTokenAndReconnect()
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
// ==================== Token 刷新 ====================
|
||||
|
||||
async function refreshTokenAndReconnect(): Promise<void> {
|
||||
if (isRefreshing) return
|
||||
|
||||
const accountStore = useAccountStore()
|
||||
if (!accountStore.refreshToken) {
|
||||
sseBus.emit('error', { message: '登录已过期，请重新登录', retryCount: 0 })
|
||||
return
|
||||
}
|
||||
|
||||
isRefreshing = true
|
||||
try {
|
||||
const newToken = await accountStore.refreshAccessToken()
|
||||
if (newToken) {
|
||||
await client.connect()
|
||||
}
|
||||
} catch {
|
||||
sseBus.emit('error', { message: '实时连接异常，请刷新页面', retryCount: 0 })
|
||||
} finally {
|
||||
isRefreshing = false
|
||||
}
|
||||
}
|
||||
|
||||
function isSseHttpError(error: unknown): error is SseHttpError {
|
||||
return error instanceof SseHttpError
|
||||
}
|
||||
|
||||
function getRetryDelay(): number {
|
||||
return Math.min(BASE_RETRY_DELAY_MS * Math.pow(2, retryCount - 1), MAX_RETRY_DELAY_MS)
|
||||
}
|
||||
|
||||
function resetConnectionState(): void {
|
||||
connecting = false
|
||||
isConnected.value = false
|
||||
}
|
||||
// ==================== Composable ====================
|
||||
|
||||
/**
|
||||
* Global SSE connection manager.
|
||||
* 全局 SSE 连接管理
|
||||
*
|
||||
* Module-level state guarantees a single active connection across all callers.
|
||||
* SseClient 只管连接生命周期，业务消息通过 sseBus 分发。
|
||||
* 组件只需 `useSse().connect()` / `useSse().disconnect()`。
|
||||
*/
|
||||
export function useSse() {
|
||||
async function connect(): Promise<void> {
|
||||
const accountStore = useAccountStore()
|
||||
const token = accountStore.accessToken
|
||||
|
||||
if (!token) {
|
||||
console.warn('[SSE] Missing accessToken, skip connection')
|
||||
return
|
||||
}
|
||||
|
||||
if (currentToken === token && (isConnected.value || connecting)) {
|
||||
return
|
||||
}
|
||||
|
||||
abort()
|
||||
connecting = true
|
||||
currentToken = token
|
||||
abortController = new AbortController()
|
||||
|
||||
await fetchEventSource(SSE_URL, {
|
||||
headers: { saToken: token },
|
||||
signal: abortController.signal,
|
||||
openWhenHidden: true,
|
||||
|
||||
async onopen(response) {
|
||||
const contentType = response.headers.get('content-type') || ''
|
||||
|
||||
if (response.status === 401) {
|
||||
throw new SseHttpError('SSE unauthorized', response)
|
||||
}
|
||||
|
||||
if (!response.ok) {
|
||||
throw new SseHttpError(`SSE connection failed: ${response.status}`, response)
|
||||
}
|
||||
|
||||
if (!contentType.includes('text/event-stream')) {
|
||||
throw new SseHttpError(`Invalid SSE content-type: ${contentType || 'empty'}`, response)
|
||||
}
|
||||
|
||||
connecting = false
|
||||
isConnected.value = true
|
||||
retryCount = 0
|
||||
sseBus.emit('connected', undefined)
|
||||
console.log('[SSE] Connected')
|
||||
},
|
||||
|
||||
onmessage(message) {
|
||||
if (!message.event || !message.data) {
|
||||
return
|
||||
}
|
||||
|
||||
try {
|
||||
const data = JSON.parse(message.data)
|
||||
sseBus.emit(message.event as keyof SseEvents, data)
|
||||
} catch {
|
||||
console.warn('[SSE] Failed to parse event payload:', message.event, message.data)
|
||||
}
|
||||
},
|
||||
|
||||
onclose() {
|
||||
resetConnectionState()
|
||||
console.log('[SSE] Closed')
|
||||
},
|
||||
|
||||
onerror(error) {
|
||||
resetConnectionState()
|
||||
|
||||
if (isSseHttpError(error) && error.response.status === 401) {
|
||||
void refreshTokenAndReconnect()
|
||||
throw error
|
||||
}
|
||||
|
||||
retryCount += 1
|
||||
if (retryCount >= MAX_RETRY_COUNT) {
|
||||
sseBus.emit('error', {
|
||||
message: 'Realtime connection failed. Please refresh the page.',
|
||||
retryCount
|
||||
})
|
||||
throw error
|
||||
}
|
||||
|
||||
const delay = getRetryDelay()
|
||||
console.warn(`[SSE] Connection error, retry in ${delay / 1000}s (${retryCount}/${MAX_RETRY_COUNT})`)
|
||||
return delay
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
function abort(): void {
|
||||
if (abortController) {
|
||||
abortController.abort()
|
||||
abortController = null
|
||||
}
|
||||
|
||||
currentToken = null
|
||||
retryCount = 0
|
||||
resetConnectionState()
|
||||
}
|
||||
|
||||
function disconnect(): void {
|
||||
abort()
|
||||
console.log('[SSE] Disconnected')
|
||||
}
|
||||
|
||||
async function refreshTokenAndReconnect(): Promise<void> {
|
||||
if (isRefreshing) {
|
||||
return
|
||||
}
|
||||
|
||||
const accountStore = useAccountStore()
|
||||
if (!accountStore.refreshToken) {
|
||||
sseBus.emit('error', { message: 'Login expired. Please sign in again.', retryCount: 0 })
|
||||
return
|
||||
}
|
||||
|
||||
isRefreshing = true
|
||||
try {
|
||||
const newToken = await accountStore.refreshAccessToken()
|
||||
if (newToken) {
|
||||
retryCount = 0
|
||||
await connect()
|
||||
}
|
||||
} catch {
|
||||
sseBus.emit('error', {
|
||||
message: 'Realtime connection failed. Please refresh the page.',
|
||||
retryCount: 0
|
||||
})
|
||||
} finally {
|
||||
isRefreshing = false
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
isConnected,
|
||||
connect,
|
||||
disconnect
|
||||
connect: () => client.connect(),
|
||||
disconnect: () => {
|
||||
client.disconnect()
|
||||
isConnected.value = false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
168
src/common/sse/SseClient.ts
Normal file
168
src/common/sse/SseClient.ts
Normal file
@@ -0,0 +1,168 @@
|
||||
import { fetchEventSource } from '@microsoft/fetch-event-source'
|
||||
|
||||
/**
|
||||
* 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<void> {
|
||||
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()
|
||||
|
||||
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
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/** 断开连接（主动关闭，不重连） */
|
||||
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
|
||||
}
|
||||
@@ -89,9 +89,17 @@ watch(activeGameId, () => {
|
||||
fetchRoomList()
|
||||
})
|
||||
|
||||
/** 首次加载 */
|
||||
/**
|
||||
* 首次加载：HTTP 全量拉取（SSE 可能尚未连接，HTTP 是最可靠的基线）
|
||||
*
|
||||
* 后续增量由 SSE 事件（room_created/room_updated/room_removed）驱动，
|
||||
* SSE 重连时通过 connected 事件触发全量覆盖作为数据补偿。
|
||||
*/
|
||||
fetchRoomList()
|
||||
|
||||
/** 是否已完成首次加载（用于区分 SSE 首次 connected 和重连） */
|
||||
let initialLoadDone = false
|
||||
|
||||
// ===== SSE 实时事件订阅处理函数 =====
|
||||
/** 新房间创建 → 添加到列表头部 */
|
||||
function onRoomCreated(room: RoomItem) {
|
||||
@@ -108,17 +116,33 @@ function onRoomUpdated(room: RoomItem) {
|
||||
function onRoomRemoved({ roomId }: { roomId: string }) {
|
||||
rooms.value = rooms.value.filter((r) => r.id !== roomId)
|
||||
}
|
||||
/**
|
||||
* SSE 重连 → 全量拉取房间列表覆盖增量丢失
|
||||
*
|
||||
* 首次 connected 跳过（已由上面的 fetchRoomList() 处理），
|
||||
* 仅重连时触发全量覆盖，避免初始重复请求。
|
||||
* 重连期间的增量事件可能丢失，HTTP 全量拉取作为补偿。
|
||||
*/
|
||||
function onSseConnected() {
|
||||
if (!initialLoadDone) {
|
||||
initialLoadDone = true
|
||||
return
|
||||
}
|
||||
fetchRoomList()
|
||||
}
|
||||
|
||||
// ===== SSE 实时事件订阅 =====
|
||||
onMounted(() => {
|
||||
sseBus.on('room_created', onRoomCreated)
|
||||
sseBus.on('room_updated', onRoomUpdated)
|
||||
sseBus.on('room_removed', onRoomRemoved)
|
||||
sseBus.on('connected', onSseConnected)
|
||||
})
|
||||
onUnmounted(() => {
|
||||
sseBus.off('room_created', onRoomCreated)
|
||||
sseBus.off('room_updated', onRoomUpdated)
|
||||
sseBus.off('room_removed', onRoomRemoved)
|
||||
sseBus.off('connected', onSseConnected)
|
||||
})
|
||||
|
||||
/** 筛选后的房间列表 */
|
||||
|
||||
@@ -13,7 +13,13 @@ export type SseEvents = {
|
||||
}
|
||||
|
||||
/**
|
||||
* Global SSE event bus. The connection layer only dispatches events; feature
|
||||
* modules decide how to handle them.
|
||||
* 全局 SSE 事件总线（发布-订阅模式）
|
||||
*
|
||||
* 连接层只负责解析 SSE 消息并 emit 到总线，业务模块自行订阅处理。
|
||||
* 示例:
|
||||
* ```ts
|
||||
* import { sseBus } from '@/stores/sseBus'
|
||||
* sseBus.on('room_updated', (room) => { ... })
|
||||
* ```
|
||||
*/
|
||||
export const sseBus = mitt<SseEvents>()
|
||||
|
||||
Reference in New Issue
Block a user