Skip to content
中文

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, \r and \n all terminate a line; a trailing \r at the end of a chunk is held back until the next chunk shows whether an \n follows.
  • 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, \r and \n all terminate a line; a trailing \r at the end of a chunk is held back until the next chunk shows whether an \n follows.
  • 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.
ParameterTypePassingDescription
bodyReadableStream<Uint8Array<ArrayBufferLike>>Required—
optionsSseReadOptionsOptional—
Fields of options
FieldTypeRequiredDescription
signalAbortSignal | undefinedOptionalCancels the pending read and any later read from the same stream.
idleTimeoutMsnumber | undefinedOptionalMaximum silence between chunks in milliseconds; omitted or &lt;= 0 disables the timer.
waitForCancelboolean | undefinedOptionalAwait reader.cancel() before returning. false lets browser adapters skip the wait.

Returns: AsyncGenerator<SseEvent, any, any>.

Examples ​

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);

API reference, usage examples and developer guides.