并行开发前的清场:白名单、读流模块、流量格式化

任务书承诺这些在分发前已存在。/traffic /logs 补进读白名单(在流式清单
里却不在读白名单里,仪表盘和请求页一开工就 403),PUT /providers/rules
补进写白名单(规则页的立刻更新)。web/stream.ts 把分块流缝成完整的行,
分块边界和行边界无关是它存在的全部理由;shared/format.ts 是连接页和
仪表盘共用的流量数字格式化。
This commit is contained in:
2026-08-11 18:02:21 +08:00
parent d028e63a74
commit 6c8a1c4e0e
5 changed files with 142 additions and 0 deletions
+55
View File
@@ -0,0 +1,55 @@
import { describe, expect, it } from 'bun:test'
import { readLines } from './stream.ts'
function streamOf(chunks: string[]): ReadableStream<Uint8Array> {
const encoder = new TextEncoder()
return new ReadableStream({
start(controller) {
for (const chunk of chunks) controller.enqueue(encoder.encode(chunk))
controller.close()
},
})
}
async function collect(chunks: string[]): Promise<string[]> {
const lines: string[] = []
for await (const line of readLines(streamOf(chunks))) lines.push(line)
return lines
}
describe('按行读流', () => {
it('一块里有好几行就拆开', async () => {
expect(await collect(['a\nb\nc\n'])).toEqual(['a', 'b', 'c'])
})
it('半行留在缓冲里,等下一块补齐', async () => {
// 分块边界和行边界无关,这是这个模块存在的全部理由
expect(await collect(['{"up":1,"do', 'wn":2}\n{"up":3}\n'])).toEqual([
'{"up":1,"down":2}',
'{"up":3}',
])
})
it('流结束时缓冲里剩的半行也交出去', async () => {
expect(await collect(['a\nb'])).toEqual(['a', 'b'])
})
it('空行丢弃,\\r\\n 换行的 \\r 去掉', async () => {
expect(await collect(['a\r\n\n\nb\n'])).toEqual(['a', 'b'])
})
it('多字节字符跨块也不会读坏', async () => {
const bytes = new TextEncoder().encode('香港01\n')
const stream = new ReadableStream<Uint8Array>({
start(controller) {
// 故意从「港」字中间切开
controller.enqueue(bytes.slice(0, 4))
controller.enqueue(bytes.slice(4))
controller.close()
},
})
const lines: string[] = []
for await (const line of readLines(stream)) lines.push(line)
expect(lines).toEqual(['香港01'])
})
})
+33
View File
@@ -0,0 +1,33 @@
/**
* 把持续吐数据的 HTTP 响应体读成一行一行。
*
* 内核的 /traffic、/memory、/logs 在没有 WebSocket 升级时退化成分块流,一行一个
* JSON。分块的边界和行的边界无关:一次 read 可能拿到半行,也可能拿到好几行。
* 这里把它们缝成完整的行再交出去,空行丢弃。
*
* 停止读取的方式是中止 fetch(AbortController)或 break 出循环,两种都会走到
* finally 里把流取消掉,不留悬着的连接。
*/
export async function* readLines(body: ReadableStream<Uint8Array>): AsyncGenerator<string> {
const reader = body.getReader()
const decoder = new TextDecoder()
let buffer = ''
try {
for (;;) {
const { done, value } = await reader.read()
if (done) break
buffer += decoder.decode(value, { stream: true })
const parts = buffer.split('\n')
// 最后一段可能是半行,留在缓冲里等下一块补齐
buffer = parts.pop() ?? ''
for (const part of parts) {
const line = part.endsWith('\r') ? part.slice(0, -1) : part
if (line) yield line
}
}
const rest = buffer.trim()
if (rest) yield rest
} finally {
await reader.cancel().catch(() => {})
}
}