im0o
0eb8d05ed8
流式接口不能走 api.ts 的 request()（它会把整个响应读完），页面自己 fetch 加 stream.ts 按行读。解析和追加上限放在 web/logs.ts 里，因为 「脏行跳过」「超上限丢最旧的」是点名要测的行为，埋在组件里测不到。 收到的行先进缓冲、每 200ms 批量刷进列表；暂停就是不刷但照收， 缓冲也设上限，恢复时补上。级别是建连时定的，切换即重连。
107 lines
3.2 KiB
TypeScript
107 lines
3.2 KiB
TypeScript
import { afterEach, describe, expect, it, mock } from 'bun:test'
|
|
import { LOG_CAP, parseLogLine, pushCapped, streamLogs } from './logs.ts'
|
|
import { LogStreamError } from './logs.ts'
|
|
|
|
describe('解析日志行', () => {
|
|
it('形状对的行解出级别和内容', () => {
|
|
expect(parseLogLine('{"type":"info","payload":"[TCP] dial example.com"}')).toEqual({
|
|
type: 'info',
|
|
payload: '[TCP] dial example.com',
|
|
})
|
|
})
|
|
|
|
it('不是 JSON 的脏行返回 null 而不是抛出去', () => {
|
|
expect(parseLogLine('not json at all')).toBeNull()
|
|
expect(parseLogLine('{"type":"info","payl')).toBeNull()
|
|
})
|
|
|
|
it('是 JSON 但形状不对的也跳过', () => {
|
|
expect(parseLogLine('null')).toBeNull()
|
|
expect(parseLogLine('[1,2]')).toBeNull()
|
|
expect(parseLogLine('{"type":"info"}')).toBeNull()
|
|
expect(parseLogLine('{"type":"info","payload":42}')).toBeNull()
|
|
})
|
|
})
|
|
|
|
describe('成批追加上限', () => {
|
|
it('超出上限丢最旧的，留下最新的', () => {
|
|
const list = [1, 2, 3]
|
|
pushCapped(list, [4, 5, 6], 4)
|
|
expect(list).toEqual([3, 4, 5, 6])
|
|
})
|
|
|
|
it('一批比上限还大时只留下限条', () => {
|
|
const list: number[] = []
|
|
pushCapped(list, [1, 2, 3, 4, 5], 3)
|
|
expect(list).toEqual([3, 4, 5])
|
|
})
|
|
})
|
|
|
|
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()
|
|
},
|
|
})
|
|
}
|
|
|
|
const realFetch = globalThis.fetch
|
|
|
|
// bun 的测试环境没有 localStorage，而 streamLogs 要从里面取令牌
|
|
if (typeof globalThis.localStorage === 'undefined') {
|
|
;(globalThis as Record<string, unknown>).localStorage = {
|
|
getItem: () => '',
|
|
setItem: () => {},
|
|
removeItem: () => {},
|
|
}
|
|
}
|
|
|
|
afterEach(() => {
|
|
globalThis.fetch = realFetch
|
|
})
|
|
|
|
describe('读日志流', () => {
|
|
it('脏行被跳过，不会中断后面的行', async () => {
|
|
globalThis.fetch = mock(() =>
|
|
Promise.resolve(
|
|
new Response(
|
|
streamOf([
|
|
'{"type":"info","payload":"第一条"}\n',
|
|
'这不是 JSON\n',
|
|
'{"type":"warning","payload":"第',
|
|
'二条"}\n',
|
|
]),
|
|
),
|
|
),
|
|
) as unknown as typeof fetch
|
|
|
|
const lines: string[] = []
|
|
for await (const line of streamLogs('info', new AbortController().signal)) {
|
|
lines.push(`${line.type}: ${line.payload}`)
|
|
}
|
|
expect(lines).toEqual(['info: 第一条', 'warning: 第二条'])
|
|
})
|
|
|
|
it('接口报错时用服务端给的 message 抛出', async () => {
|
|
globalThis.fetch = mock(() =>
|
|
Promise.resolve(
|
|
new Response(JSON.stringify({ message: '内核未连接' }), { status: 502 }),
|
|
),
|
|
) as unknown as typeof fetch
|
|
|
|
const run = async () => {
|
|
for await (const _ of streamLogs('info', new AbortController().signal)) {
|
|
// 不该有产出
|
|
}
|
|
}
|
|
await expect(run()).rejects.toThrow(new LogStreamError('内核未连接'))
|
|
})
|
|
|
|
it('LOG_CAP 是个有限值，无限流不会把内存吃光', () => {
|
|
expect(LOG_CAP).toBeGreaterThan(0)
|
|
expect(LOG_CAP).toBeLessThanOrEqual(10000)
|
|
})
|
|
})
|