《TypeScript编程实战》10.2 SSE 与流式响应

event: progress
id: 42
retry: 3000
data: {"stage":"compile","percent":80}

data: 这是一条没有事件名的消息
data: 可以占多行，客户端会拼接成一个字符串

: 以冒号开头的是注释，常用来做保活

const encoder = new TextEncoder();

function sseFrame(event: string, data: unknown, id?: number): Uint8Array {
  const lines = [`event: ${event}`];
  if (id !== undefined) lines.push(`id: ${id}`);
  lines.push(`data: ${JSON.stringify(data)}`);
  return encoder.encode(lines.join('\n') + '\n\n'); // 结尾必须有空行
}

function streamResponse(): Response {
  const stream = new ReadableStream<Uint8Array>({
    start(controller) {
      const timer = setInterval(() => {
        controller.enqueue(sseFrame('tick', { ts: Date.now() }));
      }, 1000);
      // 客户端断开时清理，否则定时器会泄漏
      return () => clearInterval(timer);
    },
  });
  return new Response(stream, {
    headers: {
      'Content-Type': 'text/event-stream; charset=utf-8',
      'Cache-Control': 'no-cache, no-transform',
      'X-Accel-Buffering': 'no', // 让 Nginx 不要缓冲
    },
  });
}

import { Hono } from 'hono';
import { streamSSE } from 'hono/streaming';

const app = new Hono();

app.get('/events', (c) => {
  return streamSSE(c, async (stream) => {
    let id = 0;
    while (!stream.aborted) {
      await stream.writeSSE({
        event: 'progress',
        data: JSON.stringify({ stage: 'compile', percent: id * 10 }),
        id: String(id++),
      });
      await stream.sleep(1000);
    }
  });
});

const source = new EventSource('/events');

source.addEventListener('progress', (event) => {
  const data = JSON.parse(event.data); // event: MessageEvent<string>
  console.log(data.percent);
});

source.addEventListener('open', () => console.log('connected'));
source.addEventListener('error', () => console.log('reconnecting...'));

async function connect(signal: AbortSignal) {
  const res = await fetch('/events', {
    method: 'POST',
    headers: {
      'Content-Type': 'application/json',
      Authorization: `Bearer ${token}`,
      Accept: 'text/event-stream',
    },
    body: JSON.stringify({ channels: ['orders', 'trades'] }),
    signal,
  });
  if (!res.ok || !res.body) throw new Error(`SSE failed: ${res.status}`);

  const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
  let buffer = '';
  for (;;) {
    const { value, done } = await reader.read();
    if (done) break;
    buffer += value;
    // 事件以空行分隔，最后一段可能不完整，留在 buffer 里
    const chunks = buffer.split('\n\n');
    buffer = chunks.pop() ?? '';
    for (const chunk of chunks) dispatch(parseFrame(chunk));
  }
}

type SseFrame = { event: string; id?: string; data: string };

function parseFrame(raw: string): SseFrame | null {
  const frame: SseFrame = { event: 'message', data: '' };
  const dataLines: string[] = [];
  for (const line of raw.split('\n')) {
    if (line === '' || line.startsWith(':')) continue; // 空行与注释
    const idx = line.indexOf(':');
    const field = idx === -1 ? line : line.slice(0, idx);
    const value = idx === -1 ? '' : line.slice(idx + 1).replace(/^ /, '');
    if (field === 'data') dataLines.push(value);
    else if (field === 'event') frame.event = value;
    else if (field === 'id') frame.id = value;
  }
  if (dataLines.length === 0) return null; // 无 data 不派发事件
  frame.data = dataLines.join('\n');
  return frame;
}

import { z } from 'zod';

const schemas = {
  progress: z.object({ stage: z.string(), percent: z.number() }),
  done: z.object({ ok: z.boolean(), elapsedMs: z.number() }),
  error: z.object({ code: z.number(), message: z.string() }),
} as const;

type EventName = keyof typeof schemas;

// 每个事件名映射到它自己的 payload 类型
type Payload<K extends EventName> = z.infer<(typeof schemas)[K]>;

function on<K extends EventName>(name: K, handler: (data: Payload<K>) => void) {
  source.addEventListener(name, (e) => {
    const parsed = schemas[name].safeParse(JSON.parse((e as MessageEvent).data));
    if (parsed.success) handler(parsed.data as Payload<K>);
  });
}

on('progress', (d) => console.log(d.percent)); // d 被精确推导为 { stage; percent }
on('progress', (d) => console.log(d.elapsedMs)); // 编译错误：属性不存在