@portable/sse

Functions

`Sse.connect`

Opens an event stream with fetch and returns its events, reconnecting with Last-Event-ID when the response ends or fails, as EventSource does. Unlike EventSource, the request can have any method, headers and body, and the events are a ReadableStream that can be piped and cancelled. Cancelling it aborts the request in flight.

Each attempt sends a clone of request, so a request body is sent again on every reconnect, and a fetch wrapper can change the headers of each attempt. A 204 response closes the stream. Any other status than 200, or a content type other than text/event-stream, fails it with ResponseError.

function connect(
  request: Request | string | URL,
  options: connect.Options = {},
): ReadableStream<Event>;

Option	Type	Default	Description
`attempts`	`number`	`Infinity`	How many attempts in a row, counting the first, may fail before `connect` stops reconnecting. The stream then errors with the last error, or closes if the last response ended normally. An accepted response resets the count.
`delay`	`any`	`{ initial: 1000, max: 30_000 }`	Wait before each reconnect, until the server sends a `retry` field.
`fetch`	`((request: Request) => Promise<Response>)`	`globalThis.fetch`	Sends each request. Pass a wrapper to change a request before it is sent.
`idleTimeout`	`number`	`Infinity`	How long to wait, in ms, for the response headers and then for each chunk of the body. If no data arrives in that time, `connect` aborts the request and handles the attempt as a failed connection with a `TimeoutError` `DOMException`, so by default it reconnects. Keep-alive comment lines from the server also restart the wait.
`maxLength`	`number`	`Infinity`	Longest line, and longest `data` of one event, in UTF-16 code units.
`reconnectIf`	`((error: unknown, retry: { attempt: number; }) => boolean \| PromiseLike<boolean>)`	`(error) => !(error instanceof Sse.ResponseError \|\| error instanceof Sse.MaxLengthError)`	Called after a connection ends while `attempts` isn't used up, with the error and `{ attempt }`, where `attempt` is the number of reconnects since the last accepted response. `error` is `undefined` when the response ended normally. Return `false` to stop reconnecting. The stream then closes if `error` is `undefined`, and errors with `error` otherwise.
`signal`	`AbortSignal`		Stops the stream, the request in flight and any wait, and fails the stream with `signal.reason`.

import * as Sse from '@portable/sse';
import * as Streams from '@portable/streams';
 
const request = new Request('https://example.com/events', {
  method: 'POST',
  body: JSON.stringify({ topic: 'prices' }),
});
const events = Sse.connect(request, {
  // Each reconnect gets a fresh token.
  fetch: async (request) => {
    request.headers.set('authorization', `Bearer ${await getToken()}`);
    return fetch(request);
  },
  signal: AbortSignal.timeout(60_000),
});
for await (const event of Streams.toAsyncIterable(events)) {
  console.log(event.data);
}

`Sse.decode`

Reads every event out of a complete text/event-stream body. An event without its closing blank line is dropped, as EventSource drops it when a stream ends.

function decode(
  input: Uint8Array | string,
  options: decode.Options = {},
): Event[];

Option	Type	Default	Description
`maxLength`	`number`	`Infinity`	Longest line, and longest `data` of one event, in UTF-16 code units.

import * as Sse from '@portable/sse';
 
Sse.decode('event: greeting\ndata: hello\n\n');
// [{ event: 'greeting', data: 'hello', id: undefined, retry: undefined }]

`Sse.encode`

Writes one message in text/event-stream format, ending with the blank line that dispatches it. A message with only a comment has no blank line.

function encode(message: Message): string;

import * as Sse from '@portable/sse';
 
Sse.encode({ event: 'update', data: 'line 1\nline 2', id: '7' });
// 'event: update\ndata: line 1\ndata: line 2\nid: 7\n\n'

Classes

`Sse.DecoderStream`

A TransformStream that reads a text/event-stream body into events, following the WHATWG HTML standard's event stream rules. Byte chunks are decoded as UTF-8.

class DecoderStream extends TransformStream<
  Uint8Array | string,
  Event
> {
  constructor(options: DecoderStream.Options = {});
}

Option	Type	Default	Description
`maxLength`	`number`	`Infinity`	Longest line, and longest `data` of one event, in UTF-16 code units. A longer one fails the stream with `MaxLengthError`.
`signal`	`AbortSignal`		Fails the stream with `signal.reason` and cancels its source when aborted.

import * as Sse from '@portable/sse';
 
const response = await fetch('https://example.com/events', {
  method: 'POST',
  body,
});
const events = response.body!.pipeThrough(new Sse.DecoderStream());

`Sse.EncoderStream`

A TransformStream that writes messages as text/event-stream bytes, for the body of a Response.

class EncoderStream extends TransformStream<
  Message,
  Uint8Array<ArrayBuffer>
> {
  constructor(options: EncoderStream.Options = {});
}

Option	Type	Default	Description
`signal`	`AbortSignal`		Fails the stream with `signal.reason` and cancels its source when aborted.

import * as Sse from '@portable/sse';
 
const body = messages.pipeThrough(new Sse.EncoderStream());
return new Response(body, {
  headers: { 'content-type': 'text/event-stream' },
});

Errors

`Sse.InvalidInputError`

Thrown by encode and EncoderStream for a field that can't be written.

class InvalidInputError extends Errors.InvalidInputError {
  readonly kind: string;
}

import * as Sse from '@portable/sse';
 
try {
  Sse.encode({ event: 'a\nb', data: 'x' });
} catch (error) {
  if (error instanceof Sse.InvalidInputError) {
    console.log(error.message);
  }
}

`Sse.MaxLengthError`

Thrown when a line or an event's data grows past maxLength.

class MaxLengthError extends Errors.MaxLengthError {
  constructor(options: Errors.MaxLengthError.Options);
  readonly kind: string;
}

import * as Sse from '@portable/sse';
 
try {
  Sse.decode(`data: ${'x'.repeat(2000)}\n\n`, { maxLength: 1000 });
} catch (error) {
  if (error instanceof Sse.MaxLengthError) {
    console.log(`longer than ${error.maxLength}`);
  }
}

`Sse.ResponseError`

Thrown by connect for a response that isn't an event stream.

class ResponseError extends Errors.ResponseError {
  readonly kind: string;
}

import * as Sse from '@portable/sse';
 
const reader = Sse.connect('https://example.com/events').getReader();
try {
  await reader.read();
} catch (error) {
  if (error instanceof Sse.ResponseError) {
    console.log(error.response.status);
  }
}

Types

`Sse.Event`

An event read from a stream, as EventSource would dispatch it.

type Event = {
  event: string;
  data: string;
  id?: string | undefined;
  retry?: number | undefined;
};

`Sse.Message`

A message to write to an event stream. Every field is optional, and an Sse.Event fits.

type Message = {
  event?: string | undefined;
  data?: string | undefined;
  id?: string | undefined;
  retry?: number | undefined;
  comment?: string | undefined;
};