eventstream

js/net/http/eventstream.ts

fino:net/http/eventstream — transport-agnostic SSE parser and formatter.

Provides the wire-level primitives for Server-Sent Events independently of any specific HTTP transport. EventSource layers its GET-only reconnecting client on top; AI model providers use parseEventStream directly for POST-body SSE streams.

import { parseEventStream } from 'fino:net/http/eventstream';
import { HttpClient } from 'fino:net/http/client';

const client = new HttpClient();
const res = await client.request(url, {
  method: 'POST',
  body: JSON.stringify({ stream: true }),
  headers: { 'content-type': 'application/json', accept: 'text/event-stream' },
});
for await (const event of parseEventStream(res.body)) {
  console.log(event.type, event.data);
}

Interfaces

interface SseEvent {

Parsed server-sent event yielded by EventSourceReader.

Events without data: fields are not yielded. id and retry are null when the event did not include those fields.

for await (const event of new EventSourceReader(body)) console.log(event.type, event.data);

Properties

type: string

Event type; defaults to "message" when the stream omits event:.

data: string

Event payload with multiple data: lines joined by newline.

id: string | null

Event ID from id:, or null when absent.

retry: number | null

Retry interval from retry:, or null when absent or invalid.

interface SseEventOptions {

Options accepted by EventSourceWriter.write() / .event().

Only data is required; all other fields are optional.

await writer.write({ event: 'update', data: 'payload', id: '1' });

Properties

data: string

Event payload. Multi-line strings are split into one data: line each.

event?: string

Event type written as an event: field; the client's message handler fires when omitted.

id?: string

Event ID written as an id: field; the client echoes it back as Last-Event-ID on reconnect.

retry?: number

Reconnection interval in milliseconds, floored to an integer and written as a retry: field.

Classes

class EventSourceReader extends Reader<SseEvent> {

Parses an SSE byte stream into discrete SseEvent objects.

Extends Reader<SseEvent> so it inherits for await, close(), closed, and [Symbol.asyncDispose](). Accepts any async iterable of byte chunks.

const reader = new EventSourceReader(response.body);
for await (const event of reader) {
  console.log(event.type, event.data);
}
console.log(reader.lastEventId);

Constructors

constructor(source: AsyncIterable<Uint8Array | ArrayBuffer>)

Create an SSE parser over a byte stream.

The parser is single-use; it consumes the source iterator as it yields events.

const reader = new EventSourceReader(response.body);

Getters

get lastEventId(): string

The last event ID seen in the stream. Reflects the id: field of the most recently yielded event. Persists across all events in the stream.

console.log(reader.lastEventId);

Methods

async read(): Promise<SseEvent | null>

Pull the next parsed event from the stream, or null once the source is exhausted.

This is the pull-based counterpart to for await; the inherited async iterator calls it under the hood. Events with no data: field are skipped rather than yielded, so each resolved value always carries a payload. The parser reads and buffers as many source chunks as needed to complete one event before resolving.

const reader = new EventSourceReader(response.body);
let event;
while ((event = await reader.read()) !== null) {
  console.log(event.type, event.data);
}

class EventSourceWriter extends Writer<SseEventOptions> {

Formats and writes SSE events to a BytesWriter.

Extends Writer<SseEventOptions> so it inherits pipe(), close(), closed, and [Symbol.asyncDispose]().

The underlying BytesWriter is not closed when this writer is closed. Callers are responsible for flushing and closing the byte sink.

const esw = new EventSourceWriter(writer);
await esw.write({ data: 'hello' });
await esw.write({ event: 'update', data: 'line1\nline2', id: '42' });
await esw.comment('keep-alive');
await esw.retry(5000);

Constructors

constructor(writer: Writer<Uint8Array> | BytesWriter)

Create an SSE writer around any byte-accepting writer, such as a BytesWriter or the writer end of a Channel<Uint8Array>.

const events = new EventSourceWriter(writer);

Methods

async write(opts: SseEventOptions): Promise<void>

Write an SSE event.

Multi-line data strings are split into separate data: lines. retry is floored to an integer. Field values are not escaped.

await events.write({ event: 'update', data: 'line 1\nline 2', id: '42' });
event(opts: SseEventOptions): Promise<void>

Alias for write(). Provided for compatibility and readability in server-side SSE handlers.

await events.event({ event: 'update', data: 'payload' });
async comment(text: string = ''): Promise<void>

Write a comment line. Useful for keep-alive heartbeats.

Multi-line comments are emitted as multiple comment lines.

await events.comment('heartbeat');
async retry(ms: number): Promise<void>

Write a standalone retry: field to update the client's reconnection interval without dispatching an event.

await events.retry(5000);

Functions

function parseEventStream(source: AsyncIterable<Uint8Array | ArrayBuffer>): EventSourceReader

Parse an async byte stream as SSE events.

Returns an EventSourceReader which is both AsyncIterable<SseEvent> and a Reader<SseEvent> with close(), closed, lastEventId, and [Symbol.asyncDispose]().

import { parseEventStream } from 'fino:net/http/eventstream';

for await (const event of parseEventStream(res.body)) {
  console.log(event.type, event.data);
}