51 lines
1.5 KiB
TypeScript
51 lines
1.5 KiB
TypeScript
/** CommandCode streams NDJSON; tolerate SSE-style `data:` prefixes and `[DONE]` sentinels. */
|
|
import { debugWhen, isDebugEnabled } from "./log.js";
|
|
|
|
export async function* iterateEvents(
|
|
body: ReadableStream<Uint8Array>,
|
|
dbg: boolean = isDebugEnabled(),
|
|
): AsyncGenerator<Record<string, any>> {
|
|
const reader = body.getReader();
|
|
const decoder = new TextDecoder();
|
|
let buffer = "";
|
|
|
|
try {
|
|
while (true) {
|
|
const { done, value } = await reader.read();
|
|
if (done) break;
|
|
buffer += decoder.decode(value, { stream: true });
|
|
|
|
let newline: number;
|
|
while ((newline = buffer.indexOf("\n")) >= 0) {
|
|
const raw = buffer.slice(0, newline);
|
|
buffer = buffer.slice(newline + 1);
|
|
const event = parseLine(raw, dbg);
|
|
if (event) yield event;
|
|
}
|
|
}
|
|
|
|
// flush any trailing line without a newline
|
|
buffer += decoder.decode();
|
|
if (buffer.trim()) {
|
|
const event = parseLine(buffer, dbg);
|
|
if (event) yield event;
|
|
}
|
|
} finally {
|
|
reader.releaseLock();
|
|
}
|
|
}
|
|
|
|
function parseLine(raw: string, dbg: boolean): Record<string, any> | null {
|
|
let line = raw.trim();
|
|
if (!line || line.startsWith(":") || line.startsWith("event:")) return null;
|
|
if (line.startsWith("data:")) line = line.slice(5).trim();
|
|
if (!line || line === "[DONE]") return null;
|
|
try {
|
|
const parsed = JSON.parse(line);
|
|
return parsed && typeof parsed === "object" ? parsed : null;
|
|
} catch {
|
|
debugWhen(dbg, "events", "parse-skip", `line=${line.slice(0, 512)}`);
|
|
return null;
|
|
}
|
|
}
|