fix: 暂存
This commit is contained in:
@@ -1,186 +1,179 @@
|
||||
import { ref } from 'vue'
|
||||
import { fetchEventSource } from '@microsoft/fetch-event-source'
|
||||
import { sseBus } from '@/stores/sseBus'
|
||||
import type { SseEvents } from '@/stores/sseBus'
|
||||
import { useAccountStore } from '@/stores/modules/account'
|
||||
|
||||
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
|
||||
let isRefreshing = false
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
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
|
||||
}
|
||||
|
||||
/**
|
||||
* SSE 连接管理 composable（基于 @microsoft/fetch-event-source）
|
||||
* Global SSE connection manager.
|
||||
*
|
||||
* 设计原则：
|
||||
* - 全局唯一：同一时间只有一条活跃的 SSE 连接
|
||||
* - token 驱动：仅 accessToken 变更时才断开重连
|
||||
* - 防重复：已连接（相同 token）时 connect() 直接跳过
|
||||
* - 所有业务变更通过 sseBus 事件总线推送，各组件自行订阅
|
||||
*
|
||||
* 使用示例：
|
||||
* ```ts
|
||||
* const { isConnected, connect, disconnect } = useSse()
|
||||
* watch(() => accountStore.accessToken, (v) => v ? connect() : disconnect())
|
||||
* ```
|
||||
* Module-level state guarantees a single active connection across all callers.
|
||||
*/
|
||||
export function useSse() {
|
||||
/** SSE 连接是否已建立 */
|
||||
const isConnected = ref(false)
|
||||
/** 用于取消 inflight 连接的控制器 */
|
||||
let abortController: AbortController | null = null
|
||||
/** 当前连接使用的 token（用于判断是否需要重连） */
|
||||
let currentToken: string | null = null
|
||||
/** 是否正在建立连接（防并发） */
|
||||
let connecting = false
|
||||
/** 当前连接的重试计数 */
|
||||
let retryCount = 0
|
||||
/** 是否正在刷新 token（防并发） */
|
||||
let isRefreshing = false
|
||||
|
||||
/**
|
||||
* 建立 SSE 连接
|
||||
*
|
||||
* 通过 fetch-event-source 发起，设置 saToken 请求头鉴权。
|
||||
* 连接期间持续接收服务端推送事件，直到 abort 或服务端关闭。
|
||||
*
|
||||
* 防重复：如果已用相同 token 连接或正在连接中，直接跳过。
|
||||
*/
|
||||
async function connect(): Promise<void> {
|
||||
const accountStore = useAccountStore()
|
||||
const token = accountStore.accessToken
|
||||
|
||||
if (!token) {
|
||||
console.warn('[SSE] 无 accessToken，跳过连接')
|
||||
console.warn('[SSE] Missing accessToken, skip connection')
|
||||
return
|
||||
}
|
||||
|
||||
// 相同 token 已连接或正在连接中，跳过
|
||||
if (currentToken === token && (isConnected.value || connecting)) {
|
||||
return
|
||||
}
|
||||
|
||||
// 取消旧连接
|
||||
abort()
|
||||
|
||||
connecting = true
|
||||
currentToken = token
|
||||
abortController = new AbortController()
|
||||
|
||||
// Vite 代理: /api/sse/subscribe → 后端 /sse/subscribe
|
||||
fetchEventSource('/api/sse/subscribe', {
|
||||
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)
|
||||
}
|
||||
|
||||
/** 连接建立 */
|
||||
async onopen() {
|
||||
connecting = false
|
||||
isConnected.value = true
|
||||
retryCount = 0
|
||||
console.log('[SSE] 连接已建立')
|
||||
sseBus.emit('connected', undefined)
|
||||
console.log('[SSE] Connected')
|
||||
},
|
||||
|
||||
/** 收到服务端事件 */
|
||||
onmessage(msg) {
|
||||
// 心跳和 connected 确认消息可能 data 为空
|
||||
if (!msg.data) return
|
||||
onmessage(message) {
|
||||
if (!message.event || !message.data) {
|
||||
return
|
||||
}
|
||||
|
||||
try {
|
||||
const data = JSON.parse(msg.data)
|
||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||
sseBus.emit(msg.event as any, data)
|
||||
const data = JSON.parse(message.data)
|
||||
sseBus.emit(message.event as keyof SseEvents, data)
|
||||
} catch {
|
||||
console.warn('[SSE] 事件数据解析失败:', msg.event, msg.data)
|
||||
console.warn('[SSE] Failed to parse event payload:', message.event, message.data)
|
||||
}
|
||||
},
|
||||
|
||||
/** 连接关闭 */
|
||||
onclose() {
|
||||
connecting = false
|
||||
isConnected.value = false
|
||||
console.log('[SSE] 连接已关闭')
|
||||
resetConnectionState()
|
||||
console.log('[SSE] Closed')
|
||||
},
|
||||
|
||||
/**
|
||||
* 连接错误处理（必须同步，签名: (err) => number | void）
|
||||
*
|
||||
* 401 → fire-and-forget 刷新 token，阻止当前连接重连，
|
||||
* 刷新成功后由 Vue watcher 触发 connect() 自动重连
|
||||
* 其他错误 → 返回重试延迟（ms），库自动排队重连
|
||||
*/
|
||||
onerror(err) {
|
||||
connecting = false
|
||||
isConnected.value = false
|
||||
onerror(error) {
|
||||
resetConnectionState()
|
||||
|
||||
// 鉴权失败 → 刷新 token（异步 fire-and-forget，onerror 本身保持同步）
|
||||
if (err.response?.status === 401) {
|
||||
if (isRefreshing) {
|
||||
// 已有刷新在进行中，直接放弃本次连接
|
||||
throw err
|
||||
}
|
||||
|
||||
isRefreshing = true
|
||||
const store = useAccountStore()
|
||||
|
||||
if (!store.refreshToken) {
|
||||
sseBus.emit('error', { message: '登录已过期，请重新登录', retryCount: 0 })
|
||||
throw err
|
||||
}
|
||||
|
||||
// 异步刷新，不阻塞 onerror 返回
|
||||
store
|
||||
.refreshAccessToken()
|
||||
.then((newToken) => {
|
||||
if (newToken) {
|
||||
console.log('[SSE] token 刷新成功，重新建立连接')
|
||||
retryCount = 0
|
||||
// connect() 由 App.vue 的 watch 触发（accessToken 变更）
|
||||
}
|
||||
})
|
||||
.catch(() => {
|
||||
console.warn('[SSE] token 刷新失败')
|
||||
sseBus.emit('error', { message: '实时连接异常，请刷新页面', retryCount: 0 })
|
||||
})
|
||||
.finally(() => {
|
||||
isRefreshing = false
|
||||
})
|
||||
|
||||
throw err // 不自动重连
|
||||
if (isSseHttpError(error) && error.response.status === 401) {
|
||||
void refreshTokenAndReconnect()
|
||||
throw error
|
||||
}
|
||||
|
||||
// 其他错误：指数退避重连
|
||||
retryCount++
|
||||
if (retryCount >= 5) {
|
||||
retryCount += 1
|
||||
if (retryCount >= MAX_RETRY_COUNT) {
|
||||
sseBus.emit('error', {
|
||||
message: '实时连接异常，请刷新页面',
|
||||
message: 'Realtime connection failed. Please refresh the page.',
|
||||
retryCount
|
||||
})
|
||||
throw err // 超过阈值，停止重连
|
||||
throw error
|
||||
}
|
||||
|
||||
// 返回重试间隔（ms），指数退避：3s, 6s, 12s, 24s，最大 30s
|
||||
const delay = Math.min(3000 * Math.pow(2, retryCount - 1), 30000)
|
||||
console.warn(`[SSE] 连接错误，${delay / 1000}s 后重试 (${retryCount}/5)`)
|
||||
const delay = getRetryDelay()
|
||||
console.warn(`[SSE] Connection error, retry in ${delay / 1000}s (${retryCount}/${MAX_RETRY_COUNT})`)
|
||||
return delay
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* 取消当前 SSE 连接
|
||||
*/
|
||||
function abort(): void {
|
||||
if (abortController) {
|
||||
abortController.abort()
|
||||
abortController = null
|
||||
}
|
||||
connecting = false
|
||||
isConnected.value = false
|
||||
|
||||
currentToken = null
|
||||
retryCount = 0
|
||||
resetConnectionState()
|
||||
}
|
||||
|
||||
/**
|
||||
* 断开 SSE 连接
|
||||
*
|
||||
* 在用户登出时调用。
|
||||
*/
|
||||
function disconnect(): void {
|
||||
abort()
|
||||
console.log('[SSE] 连接已断开')
|
||||
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 {
|
||||
|
||||
@@ -1,40 +1,19 @@
|
||||
// webgame-frontend/src/stores/sseBus.ts
|
||||
import mitt from 'mitt'
|
||||
import type { RoomItem } from '@/api/modules/game'
|
||||
|
||||
/**
|
||||
* SSE 推送事件类型映射
|
||||
*
|
||||
* 每个事件类型对应一个数据类型，类型安全。
|
||||
* connected / error 由 useSse 连接管理发出，其余由服务端 SSE 事件映射而来。
|
||||
*/
|
||||
type SseEvents = {
|
||||
/** SSE 连接成功 */
|
||||
connected: void
|
||||
/** SSE 连接异常 */
|
||||
export type SseEvents = {
|
||||
connected: undefined
|
||||
error: { message: string; retryCount: number }
|
||||
/** 新房间创建 */
|
||||
room_created: RoomItem
|
||||
/** 房间状态更新 */
|
||||
room_updated: RoomItem
|
||||
/** 房间移除 */
|
||||
room_removed: { roomId: string }
|
||||
/** 被踢出房间 */
|
||||
kicked: { roomId: string }
|
||||
/** 所在房间游戏开始 */
|
||||
game_started: { roomId: string }
|
||||
/** 余额变动 */
|
||||
balance_chg: { balance: number; delta: number }
|
||||
}
|
||||
|
||||
/**
|
||||
* 全局 SSE 事件总线（发布-订阅模式）
|
||||
*
|
||||
* 示例:
|
||||
* ```ts
|
||||
* import { sseBus } from '@/stores/sseBus'
|
||||
* const off = sseBus.on('room_updated', (room) => { ... })
|
||||
* onUnmounted(off)
|
||||
* ```
|
||||
* Global SSE event bus. The connection layer only dispatches events; feature
|
||||
* modules decide how to handle them.
|
||||
*/
|
||||
export const sseBus = mitt<SseEvents>()
|
||||
|
||||
Reference in New Issue
Block a user