sse.js
3.03 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
/**
* SSE 帧切分与迭代 —— 自 base.js 提取的纯函数模块（职责单一硬约束）。
*
* 零副作用、零依赖：从 Response.body（ReadableStream）按 SSE 事件分隔（空行）分帧，
* 抽 `data:` 负载、`[DONE]` 终止、parseFrame 解析 JSON。base.js 一行 import 接入。
* 中止时抛出 `signal.reason`（承接 base.js 的每 attempt 组合信号），使内部超时的
* TimeoutError reason 能透传到 map-error（区别于外部 abort 的 AbortError）。
*/
/**
* 从 Response.body 按行解析 SSE，yield 每个 `data:` 帧的 JSON（或字符串）。
* 兼容 `event: x\ndata: {...}` 多行帧；遇 `[DONE]` 停止。
* @param {Response} res
* @param {AbortSignal} [signal]
* @returns {AsyncGenerator<any>}
*/
export async function* iterateSse(res, signal) {
const body = res.body
if (!body) return
const decoder = new TextDecoder()
let buf = ''
const reader = body.getReader()
try {
for (;;) {
if (signal?.aborted) throw abortReason(signal)
const { value, done } = await reader.read()
if (done) break
buf += decoder.decode(value, { stream: true })
// 按 SSE 事件分隔（空行）切块；块内取 data: 行拼接。
let sepIdx
while ((sepIdx = indexOfEventSep(buf)) !== -1) {
const rawEvent = buf.slice(0, sepIdx)
buf = buf.slice(sepIdx).replace(/^(\r?\n){1,2}/, '')
const data = extractData(rawEvent)
if (data === null) continue
if (data === '[DONE]') return
yield parseFrame(data)
}
}
// 收尾：缓冲里可能残留最后一个无尾随空行的事件。
const tail = extractData(buf)
if (tail !== null && tail !== '[DONE]') yield parseFrame(tail)
} finally {
reader.releaseLock?.()
}
}
/**
* 中止原因：优先透传 signal.reason（内部超时的 TimeoutError / 外部的 AbortError），
* 缺省造 AbortError。
* @param {AbortSignal} signal
* @returns {unknown}
*/
function abortReason(signal) {
return signal.reason ?? Object.assign(new Error('aborted'), { name: 'AbortError' })
}
/**
* 找事件分隔（`\n\n` 或 `\r\n\r\n`）的起点；无则 -1。
* @param {string} buf
* @returns {number}
*/
function indexOfEventSep(buf) {
const a = buf.indexOf('\n\n')
const b = buf.indexOf('\r\n\r\n')
if (a === -1) return b
if (b === -1) return a
return Math.min(a, b)
}
/**
* 从一个 SSE 事件块里抽出 data 负载（拼接多行 data:）；无 data 行返回 null。
* @param {string} rawEvent
* @returns {string | null}
*/
function extractData(rawEvent) {
const lines = rawEvent.split(/\r?\n/)
/** @type {string[]} */
const dataLines = []
for (const line of lines) {
if (line.startsWith('data:')) dataLines.push(line.slice(5).trimStart())
}
if (dataLines.length === 0) return null
return dataLines.join('\n')
}
/**
* data 负载若是 JSON 则解析，否则原样返回字符串。
* @param {string} data
* @returns {any}
*/
function parseFrame(data) {
try {
return JSON.parse(data)
} catch {
return data
}
}