/** * 原生 EventSource 轻量封装 * 自动重连 & 任意事件监听 * 支持自定义请求头（通过 URL 参数传递 Authorization） */exportdefaultclassSSE{privateurl:string;privatees:EventSource|null;privateretry:number;privateheaders?:Record<string,string>;constructor(url:string,headers?:Record<string,string>){this.url=urlthis.es=nullthis.retry=3000// 重连间隔 msthis.headers=headersthis.connect()}/* 建立连接 */connect():void{// 如果有 headers，将 Authorization 添加到 URL 中（因为 EventSource 不支持自定义 headers）letfinalUrl=this.url;if(this.headers?.Authorization){consttoken=this.headers.Authorization.replace('Bearer ','');constseparator=this.url.includes('?')?'&':'?';finalUrl=`${this.url}${separator}authorization=${encodeURIComponent(token)}`;}this.es=newEventSource(finalUrl)this.es.addEventListener('open',()=>{console.log('[SSE] connected')})this.es.addEventListener('error',()=>{console.log('[SSE] disconnected, retrying...')this.es?.close()setTimeout(()=>this.connect(),this.retry)})}/** * 订阅任意后端事件 * @param {string} event 事件名（与后端 event:xxx 对应） * @param {Function} cb 回调 (data: any) => {} */subscribe<T=any>(event:string,cb:(data:T)=>void):void{this.es?.addEventListener(event,(e:MessageEvent)=>{try{constdata=JSON.parse(e.data)cb(data)}catch(err){console.error('[SSE] parse error',err)}})}/* 手动关闭连接 */close():void{this.es?.close()console.log('[SSE] closed by client')}}
import{defineStore}from'pinia';// 生成唯一 clientIdfunctiongenerateClientId():string{conststored=localStorage.getItem('sse_client_id');if(stored)returnstored;constnewId=`client_${Date.now()}_${Math.random().toString(36).substr(2,9)}`;localStorage.setItem('sse_client_id',newId);returnnewId;}exportconstuseNotificationStore=defineStore('notification',{state:()=>({sse:null,// ...}),actions:{asyncinitSSE(){if(this.sse)return;constclientId=generateClientId();consttoken=localStorage.getItem('token');if(!token){console.error('[NotificationStore] No token found, cannot init SSE');return;}// 建立 SSE 连接，传递 Authorization 请求头constsseUrl=`xxx/xxx/xx?clientId=${clientId}`;this.sse=newSSE(sseUrl,{Authorization:`${token}`});/* 任务状态事件 */this.sse.subscribe<{code:number;data:any;message:string}>('sse_task_status',(payload)=>{console.log('sse监听-----------------------',payload)if(payload.code!==200)return;// payload.data为实时数据// 取出数据放入store});},// 断开 SSE 连接disconnectSSE(){if(this.sse){this.sse.close();this.sse=null;console.log('[NotificationStore] SSE connection closed');}},},});
// 建立连接notificationStore.initSSE()// 断开连接notificationStore.disconnectSSE()
![]()
