curl -v https://httpbin.org/get

> GET /get HTTP/1.1            ← 请求行：方法 路径 协议版本
> Host: httpbin.org            ← 请求头，一行一个
> Accept: */*
>                              ← 空行：请求头结束
< HTTP/1.1 200 OK              ← 状态行
< Content-Type: application/json
< Content-Length: 256          ← 响应体有多少字节
<                              ← 空行：响应头结束
{ ...响应体... }

HTTP/1.1 200 OK
Content-Type: text/event-stream
Transfer-Encoding: chunked

b                       ← 十六进制 b = 11：下面这块 11 字节
data: 你\n\n            （"data: " 6 字节 + 中文 3 字节 + 两个换行）
b
data: 好\n\n
0                       ← 长度 0：结束

curl -N https://api.openai.com/v1/chat/completions \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model":"gpt-4o-mini","stream":true,"messages":[{"role":"user","content":"数到3"}]}'

data: {"id":"chatcmpl-1","choices":[{"delta":{"role":"assistant","content":""}}]}

data: {"id":"chatcmpl-1","choices":[{"delta":{"content":"1"}}]}

data: {"id":"chatcmpl-1","choices":[{"delta":{"content":"，2"}}]}

data: [DONE]

event: token          ← 事件类型，可选，不写默认是 "message"
id: 42                ← 事件 id，可选，用于断线续传
retry: 3000           ← 告诉浏览器：断线后等 3000 毫秒再重连，可选
data: 第一行          ← 数据，可以有多行
data: 第二行          ←   多行 data 会用 "\n" 拼起来
                      ← 空行：这条事件结束，派发出去
: ping                ← 以冒号开头的是注释，解析器直接丢弃（常用作心跳）

// ❌ 错误：假设每次 read 正好是完整的一行或几行
for await (const chunk of body) {
  for (const line of decoder.decode(chunk).split("\n")) {
    if (line.startsWith("data: ")) handle(line.slice(6));
  }
}

export class SSEParser {
  private buf = "";              // 还没凑成完整一行的残余文本
  private data: string[] = [];   // 当前事件攒的 data 行
  private eventType = "";
  private lastId: string | undefined;
  private pendingCR = false;     // 上一块以 \r 结尾

  feed(chunk: string): SSEEvent[] {
    if (chunk === "") return [];
    if (this.pendingCR && chunk.startsWith("\n")) chunk = chunk.slice(1); // ①
    this.pendingCR = false;
    this.buf += chunk;

    const out: SSEEvent[] = [];
    for (;;) {
      const m = /\r\n|\n|\r/.exec(this.buf);       // ② 找第一个行尾
      if (!m) break;                                 //    没有完整行：等下一块
      if (m[0] === "\r" && m.index === this.buf.length - 1) this.pendingCR = true;
      const line = this.buf.slice(0, m.index);
      this.buf = this.buf.slice(m.index + m[0].length); // ③ 剩下的留在缓冲区
      const ev = this.line(line);
      if (ev) out.push(ev);
    }
    return out;
  }

  private line(line: string): SSEEvent | undefined {
    if (line === "") return this.dispatch();         // ④ 空行：派发事件
    if (line.startsWith(":")) return undefined;      // 注释 / 心跳

    const i = line.indexOf(":");
    const field = i === -1 ? line : line.slice(0, i);
    let value = i === -1 ? "" : line.slice(i + 1);
    if (value.startsWith(" ")) value = value.slice(1); // 只去一个空格

    switch (field) {
      case "data":  this.data.push(value); break;
      case "event": this.eventType = value; break;
      case "id":    this.lastId = value; break;
      // retry 略，完整版见练习
    }
    return undefined;
  }

  private dispatch(): SSEEvent | undefined {
    if (this.data.length === 0) { this.eventType = ""; return undefined; }
    const ev = { event: this.eventType || "message", data: this.data.join("\n"), id: this.lastId };
    this.data = [];
    this.eventType = "";   // 注意：id 不清空，规范规定 lastEventId 会一直沿用
    return ev;
  }
}

const decoder = new TextDecoder();
for (;;) {
  const { value, done } = await reader.read();
  if (done) break;
  parser.feed(decoder.decode(value, { stream: true })); // stream: true 是关键
}
parser.feed(decoder.decode()); // 最后冲刷一次，吐出解码器里剩的字节

const es = new EventSource("/events");
es.addEventListener("token", (e) => console.log(e.data));

发起 GET 请求
   ↓
持续接收 text/event-stream
   ↓
解析 event / data / id
   ↓
连接断开 → 自动重连（带上 Last-Event-ID，见 9 节）

POST /api/chat
Authorization: Bearer xxx
Content-Type: application/json
Accept: text/event-stream

{"model": "xxx", "messages": [{"role": "user", "content": "帮我解释一下 Transformer"}]}

                    SSE（text/event-stream）
                            │
              ┌─────────────┴─────────────┐
              │                           │
         EventSource                    fetch
              │                           │
         只能 GET                    GET / POST / …
       不能自定义请求头               任意请求头
        不能带请求体                  可以带请求体
       自动重连、自动解析            自己解析、自己重连
              │                           │
    适合：通知、进度推送         适合：LLM Chat / Agent / 模型 API

fetch(POST) → response.body（ReadableStream）→ 自己解析 SSE → token / tool / done

export async function* streamChat(url: string, body: unknown, opts: { signal?: AbortSignal } = {}) {
  const res = await fetch(url, {
    method: "POST",
    headers: { "Content-Type": "application/json", Accept: "text/event-stream" },
    body: JSON.stringify(body),
    signal: opts.signal,                 // 0.2 课的 AbortController：用户点停止就 abort
  });
  if (!res.ok) throw new Error(`HTTP ${res.status}`);   // ① 先看状态码
  if (!res.headers.get("content-type")?.startsWith("text/event-stream"))
    throw new Error("not SSE");                          // ② 再看类型
  for await (const ev of parseSSE(res.body!, opts.signal)) {
    if (ev.event === "token") yield { type: "token", text: JSON.parse(ev.data).text };
    else if (ev.event === "done") { yield { type: "done" }; return; }
  }
  throw new Error("stream ended without done");          // ③
}

HTTP 流正常结束  ≠  LLM 任务正常完成

① HTTP 层
   ├── 2xx                 → 继续
   └── 401 / 429 / 5xx     → 失败（读 JSON 错误体，429 按 Retry-After 退避）

② SSE 协议层
   ├── text/event-stream   → 继续
   └── 其他 Content-Type   → 失败

③ 业务协议层
   ├── token               → 持续输出
   ├── tool                → 工具调用事件
   ├── done                → 成功结束
   └── 流 EOF 但没有 done  → 异常 / 未完成

func chatHandler(w http.ResponseWriter, r *http.Request) {
    flusher, ok := w.(http.Flusher) // ① 断言：这个 ResponseWriter 支不支持 Flush
    if !ok {
        http.Error(w, "streaming unsupported", http.StatusInternalServerError)
        return
    }

    h := w.Header()                 // ② 响应头必须在第一次写 body 之前设好
    h.Set("Content-Type", "text/event-stream; charset=utf-8")
    h.Set("Cache-Control", "no-cache, no-transform")
    h.Set("X-Accel-Buffering", "no")
    w.WriteHeader(http.StatusOK)
    flusher.Flush()                 // 先把头发出去，客户端马上知道“连上了”

    ctx := r.Context()              // ③ 客户端断开时，这个 ctx 会被取消
    tokens := callLLM(ctx)          // 0.1 课写过的：返回 <-chan string，ctx 取消就停

    heartbeat := time.NewTicker(15 * time.Second)
    defer heartbeat.Stop()

    for {
        select {
        case <-ctx.Done():          // ④ 客户端走了：直接返回，callLLM 也因同一个 ctx 停下
            return
        case <-heartbeat.C:
            fmt.Fprint(w, ": ping\n\n") // ⑤ 注释行当心跳
            flusher.Flush()
        case tok, ok := <-tokens:
            if !ok {                // channel 关了：模型输出完毕
                fmt.Fprint(w, "event: done\ndata: [DONE]\n\n")
                flusher.Flush()
                return
            }
            b, _ := json.Marshal(map[string]string{"text": tok})
            fmt.Fprintf(w, "event: token\ndata: %s\n\n", b)
            flusher.Flush()         // ⑥ 每写一条就 Flush，否则会被缓冲
        }
    }
}

res.writeHead(200, {
  "Content-Type": "text/event-stream; charset=utf-8",
  "Cache-Control": "no-cache, no-transform",
  "X-Accel-Buffering": "no",
});

const ac = new AbortController();
res.on("close", () => {                 // ① 连接关闭（不管谁关的）
  if (!res.writableFinished) ac.abort(); //    还没正常写完就关了 = 客户端断开
});

const hb = setInterval(() => res.write(": ping\n\n"), 15_000);
try {
  for (let i = start; i < tokens.length; i++) {
    await sleep(interval, ac.signal);    // ② 所有等待都挂在 signal 上
    res.write(encodeSSE({ event: "token", id: String(i), data: JSON.stringify({ text: tokens[i] }) }));
  }
  res.write(encodeSSE({ event: "done", data: "[DONE]" }));
  res.end();
} finally {
  clearInterval(hb);                     // ③ 无论怎么结束，定时器都要清
}

client ── Last-Event-ID: 123 ──▶ server ── 从 124 开始补发 ──▶ 124, 125, 126 …
                                    ↑
                         前提：事件按 id 存在某处

const lastId = Number(req.headers["last-event-id"] ?? -1);
const start = Number.isInteger(lastId) && lastId >= 0 ? lastId + 1 : 0;

模型调用 ──写入──▶ 消息缓冲（Redis Stream / 内存环形缓冲，按会话 id 存）
                         │
前端连接 ◀──按 id 读取───┘   断线重连时从 Last-Event-ID 之后读

location /api/chat {
    proxy_pass http://backend;
    proxy_http_version 1.1;          # 长连接需要 1.1
    proxy_set_header Connection "";
    proxy_buffering off;             # 关缓冲
    proxy_read_timeout 300s;         # 模型慢的时候别被 60 秒默认值掐掉
    gzip off;
}

GET /ws HTTP/1.1
Host: example.com
Upgrade: websocket                         ← 我想换协议
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Version: 13

HTTP/1.1 101 Switching Protocols           ← 好，换了
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=

import "github.com/coder/websocket"
import "github.com/coder/websocket/wsjson"

type ClientMsg struct {
    Type string `json:"type"` // "user_message" | "interrupt"
    Text string `json:"text,omitempty"`
}

func wsHandler(w http.ResponseWriter, r *http.Request) {
    c, err := websocket.Accept(w, r, &websocket.AcceptOptions{
        OriginPatterns: []string{"app.example.com"}, // ① 只允许自己的前端连
    })
    if err != nil {
        return
    }
    defer c.CloseNow()

    ctx, cancel := context.WithCancel(r.Context())
    defer cancel()

    var (
        mu        sync.Mutex
        cancelGen context.CancelFunc = func() {}    // 当前这次生成的取消函数
    )

    for {
        var msg ClientMsg
        if err := wsjson.Read(ctx, c, &msg); err != nil { // ② 读循环：阻塞等客户端消息
            return                                        //    连接断开就退出
        }
        switch msg.Type {
        case "interrupt":
            mu.Lock()
            cancelGen()                                   // ③ 打断正在进行的生成
            mu.Unlock()
        case "user_message":
            mu.Lock()
            cancelGen()                                   // 新消息到来，先停掉旧的
            genCtx, genCancel := context.WithCancel(ctx)
            cancelGen = genCancel
            mu.Unlock()
            go generate(genCtx, c, msg.Text)              // ④ 生成放在另一个 goroutine
        }
    }
}

func generate(ctx context.Context, c *websocket.Conn, prompt string) {
    for tok := range callLLM(ctx, prompt) {
        if err := wsjson.Write(ctx, c, map[string]string{"type": "token", "text": tok}); err != nil {
            return
        }
    }
}

const ws = new WebSocket("wss://api.example.com/ws"); // 线上一定用 wss://（TLS）

ws.addEventListener("open", () => {
  ws.send(JSON.stringify({ type: "user_message", text: "你好" }));
});

ws.addEventListener("message", (e) => {
  const msg = JSON.parse(e.data as string);
  if (msg.type === "token") render(msg.text);
});

ws.addEventListener("close", (e) => {
  console.log("closed", e.code, e.reason); // 1000 正常，1006 异常断开（没收到 close 帧）
  // ⚠️ WebSocket 不会自动重连，要自己做：指数退避 + 重新订阅会话
});

stopButton.onclick = () => ws.send(JSON.stringify({ type: "interrupt" }));

cd projects/03-streaming
npm install
npm run typecheck && npm test   # 13 个用例全部通过
npm run dev                     # 看终端里逐字打印

it("任意位置切块，结果都与整块一致", () => {
  const raw = 'event: token\r\nid: 1\r\ndata: {"text":"你好"}\r\n\r\n: ping\n\ndata: x\ndata: y\n\n';
  const whole = parseAll([raw]);
  for (let i = 0; i <= raw.length; i++) {
    for (let j = i; j <= raw.length; j++) {
      expect(parseAll([raw.slice(0, i), raw.slice(i, j), raw.slice(j)])).toEqual(whole);
    }
  }
});

const body = new ReadableStream<Uint8Array>({
  start(c) {
    for (const b of bytes) c.enqueue(new Uint8Array([b])); // 每次只给 1 个字节
    c.close();
  },
});

const ac = new AbortController();
for await (const ev of streamChat(url, {}, { signal: ac.signal })) {
  if (ev.type === "token") {
    got.push(ev.text);
    if (got.length === 2) ac.abort();   // 收到 2 个 token 就取消
  }
}
// 断言：服务端的 stats.aborted 变成 1，completed 仍为 0