readSse
CORE · function · Runtime value
Decode an SSE response body into events.
Framing contract:
- A network chunk is not an event. Bytes are buffered until a blank line, because UTF-8 sequences, CRLF pairs and
data:lines may all straddle chunk boundaries. \r\n,\rand\nall terminate a line; a trailing\rat the end of a chunk is held back until the next chunk shows whether an\nfollows.- Comment lines (leading
:) are discarded; they are used as heartbeats. - The size limit applies per line and per event (1 MiB), not per network read, so many complete events coalesced into one chunk after backpressure remain valid.
- Cancellation is checked between lines, so aborting while the consumer is paused stops emitting the remaining buffered events of the same chunk.
Import
ts
import { readSse } from '@jangts/agent-core/sse';js
import { readSse } from '@jangts/agent-core/sse';Other public entry points:@jangts/agent-core, @jangts/agent-core/types, @jangts/agent-core/sse, @jangts/agent-core/support.
Definition
Call
ts
readSse(body: ReadableStream<Uint8Array>, options?: SseReadOptions): AsyncGenerator<SseEvent>Decode an SSE response body into events.
Framing contract:
- A network chunk is not an event. Bytes are buffered until a blank line, because UTF-8 sequences, CRLF pairs and
data:lines may all straddle chunk boundaries. \r\n,\rand\nall terminate a line; a trailing\rat the end of a chunk is held back until the next chunk shows whether an\nfollows.- Comment lines (leading
:) are discarded; they are used as heartbeats. - The size limit applies per line and per event (1 MiB), not per network read, so many complete events coalesced into one chunk after backpressure remain valid.
- Cancellation is checked between lines, so aborting while the consumer is paused stops emitting the remaining buffered events of the same chunk.
| Parameter | Type | Passing | Description |
|---|---|---|---|
body | ReadableStream<Uint8Array<ArrayBufferLike>> | Required | — |
options | SseReadOptions | Optional | — |
Fields of options
| Field | Type | Required | Description |
|---|---|---|---|
signal | AbortSignal | undefined | Optional | Cancels the pending read and any later read from the same stream. |
idleTimeoutMs | number | undefined | Optional | Maximum silence between chunks in milliseconds; omitted or <= 0 disables the timer. |
waitForCancel | boolean | undefined | Optional | Await reader.cancel() before returning. false lets browser adapters skip the wait. |
Returns: AsyncGenerator<SseEvent, any, any>.
Examples
- SSE, JSON frames, and bounded event streams: Reuse transport mechanics while preserving cancellation and backpressure; protocol parsing and business validation remain separate.
Parse one SSE frame locally
Beginner · Node.js
ts
import { Support } from '@jangts/agent-core';
const frame = Support.encodeSseEvent({ event: 'status', data: JSON.stringify({ phase: 'thinking' }) });
const body = new Response(frame).body!;
for await (const event of Support.readSse(body)) console.log(event.event, event.data);js
import { Support } from '@jangts/agent-core';
const frame = Support.encodeSseEvent({ event: 'status', data: JSON.stringify({ phase: 'thinking' }) });
const body = new Response(frame).body;
for await (const event of Support.readSse(body))
console.log(event.event, event.data);Producer waits for each event delivery
Configuration · Node.js
ts
import { Support } from '@jangts/agent-core';
const signal = new AbortController().signal;
const stream = Support.streamEvents<{ phase: string }>(async emit => {
await emit({ phase: 'preparing' });
await emit({ phase: 'complete' });
}, { signal });
for await (const event of stream) console.log(event.phase);js
import { Support } from '@jangts/agent-core';
const signal = new AbortController().signal;
const stream = Support.streamEvents(async (emit) => {
await emit({ phase: 'preparing' });
await emit({ phase: 'complete' });
}, { signal });
for await (const event of stream)
console.log(event.phase);Explicitly limit JSON frame count and surrounding content
Extension · Node.js
ts
import { Support } from '@jangts/agent-core';
const text = '{"kind":"reply","answer":"服务正常"}';
console.log(Support.scanJsonObjectFrames(text));
console.log(Support.parseJsonFrames(text, { maxFrames: 1, maxExtraClosings: 0 }));
const sent: string[] = [];
await Support.writeSseFrame({
writer: { write(frame) { sent.push(frame); return true; }, async waitForDrain() {} },
frame: Support.encodeSseEvent({ event: 'answer', data: text }),
}, { signal: new AbortController().signal });
console.log(sent);js
import { Support } from '@jangts/agent-core';
const text = '{"kind":"reply","answer":"服务正常"}';
console.log(Support.scanJsonObjectFrames(text));
console.log(Support.parseJsonFrames(text, { maxFrames: 1, maxExtraClosings: 0 }));
const sent = [];
await Support.writeSseFrame({
writer: { write(frame) { sent.push(frame); return true; }, async waitForDrain() { } },
frame: Support.encodeSseEvent({ event: 'answer', data: text }),
}, { signal: new AbortController().signal });
console.log(sent);