SSE 类型化事件流订阅

示例说明

SSE 是服务端到客户端单向流的轻量标准：浏览器原生 EventSource 自带断线重连与事件分发，比 WebSocket 更适合进度推送这类只读场景。addEventListener 按事件名注册——服务端的 event: progress 行决定路由目标，配合 TypeScript 的处理器映射接口，每种事件的载荷类型都在编译期钉死。

技术要点

• 事件名路由配合类型化处理器映射
• done 事件显式 close 结束流
• 返回清理函数统一拆除监听与连接

代码解读

subscribeJobStream 为三种事件注册监听：progress 持续更新进度、done 取到结果后关闭连接、alert 透传告警。onerror 里通过 readyState 区分自动重连与服务器正常关闭。返回的 teardown 函数把监听与连接一并清理，防止组件卸载后泄漏。

使用建议

JSON.parse 前先校验载荷形态更稳妥。跨域 SSE 需要服务端 CORS 配置。文本事件就够用时别上二进制协议。

代码示例

type ProgressEvent = { jobId: string; percent: number };
type DoneEvent = { jobId: string; resultUrl: string };
type AlertEvent = { level: 'warn' | 'error'; text: string };

interface StreamHandlers {
  progress: (event: ProgressEvent) => void;
  done: (event: DoneEvent) => void;
  alert: (event: AlertEvent) => void;
}

function subscribeJobStream(jobId: string, handlers: StreamHandlers) {
  const source = new EventSource('/api/jobs/' + jobId + '/stream');

  const entries: [keyof StreamHandlers, (e: MessageEvent) => void][] = [
    ['progress', (e) => handlers.progress(JSON.parse(e.data) as ProgressEvent)],
    ['done', (e) => {
      handlers.done(JSON.parse(e.data) as DoneEvent);
      source.close();
    }],
    ['alert', (e) => handlers.alert(JSON.parse(e.data) as AlertEvent)],
  ];

  for (const [type, listener] of entries) {
    source.addEventListener(type, listener as EventListener);
  }

  source.onerror = () => {
    // EventSource 自动重连；done 后的断开无需处理
    if (source.readyState === EventSource.CLOSED) {
      console.log('stream closed by server');
    }
  };

  return () => {
    for (const [type, listener] of entries) {
      source.removeEventListener(type, listener as EventListener);
    }
    source.close();
  };
}

const teardown = subscribeJobStream('job-7', {
  progress: (e) => console.log('progress', e.percent),
  done: (e) => console.log('result at', e.resultUrl),
  alert: (e) => console.warn(e.level, e.text),
});
teardown();