Server-Sent EventsCopy Anchor Link

An SDK exports an event-stream operation with input, event contracts, and buffer limits. Applications configure authentication and optional reconnection on their own clients, then call the imported operation when they need a stream.

Open a stream, iterate it, then release the owned handle with await using. Manual close() and closed remain available; clients and plugins don’t dispose a returned stream for you.

Basic SetupCopy Anchor Link

import {
  createClient,
  defineEventStream,
  struct,
  withEndpoint,
} from "@defjs/core";

const client = createClient(withEndpoint("https://api.example.com"));
const notifications = defineEventStream({
  maxBufferSize: 64 * 1024,
  maxQueueSize: 100,
  path: "/notifications",
  events: {
    message: struct.object({ text: struct.string() }),
  },
});

async function main() {
  const [error, stream] = await notifications(client);
  if (error) {
    console.error(error.code);
    return;
  }
  await using ownedStream = stream;
  for await (const event of ownedStream) {
    if (event.event === "message") console.log(event.data.text);
  }
}

void main();

Define the streamCopy Anchor Link

defineEventStream(...) needs events, positive safe-integer maxBufferSize, positive safe-integer maxQueueSize, and a relative path. Method defaults to GET.

The request shape matches HTTP: builder returning path, query, headers, and body. GET / HEAD cannot return a body. A declared error struct types the startup fault and decodes a non-2xx handshake body by struct kind (string text, blob / file blob, arrayBuffer bytes, otherwise JSON). Defjs sends Accept: text/event-stream when you didn’t already set Accept. A declared input always requires an explicit builder. input only defines the parameters; the builder alone decides where each value goes. Only definitions without input may omit builder. A ReadableStream body follows the HTTP rules: on a runtime that cannot stream request bodies, such as Firefox, startup fails with ENV_UNSUPPORTED before anything is sent. In Chromium it needs HTTP/2 or later; over HTTP/1.1 startup fails with NET_UNREACHABLE. A stream body is sent once and never reconnected (see Reconnect).

One logical stream can span several physical Fetch attempts when you opt into reconnect. Without withSSEReconnect, transient network and stream-read failures end the logical stream. You still get one handle and one async iterator when the open succeeds.

Open and inspectCopy Anchor Link

The operation resolves only after status, content-type, and body checks pass:

import {
  createClient,
  defineEventStream,
  struct,
  withEndpoint,
} from "@defjs/core";

const client = createClient(withEndpoint("https://api.example.com"));
const notifications = defineEventStream({
  maxBufferSize: 64 * 1024,
  maxQueueSize: 100,
  path: "/notifications",
  events: { message: struct.string() },
});

async function main() {
  const [error, stream, startupOpen] = await notifications(client);
  if (error) {
    console.error(error.code, startupOpen?.response.status);
    return;
  }
  console.log(
    stream.open.response.status,
    startupOpen.response.status,
    stream.open.url,
  );
  stream.close("example-finished");
  await stream.closed;
}

void main();

Response must be successful, media type essence text/event-stream, and have a body. Non-2xx startup → HTTP_STATUS. When error is declared, that fault carries the decoded handshake body; a bad media type is RES_MEDIA_TYPE_INVALID and a body that cannot be read as the declared kind is RES_DECODE_FAILED. A response snapshot can still sit in the third tuple slot when validation fails after the response arrives.

A missing, relative, or invalid base endpoint or an illegal path fails with REQ_BUILD_FAILED during building, before any interceptor or transport runs, even if an interceptor would short-circuit. The resolved URL must keep the base origin and stay under its directory; C0 control characters (including tab, CR, and LF), backslashes, ?, #, and directory escapes in the path are rejected. Scheme-like and protocol-relative forms are rejected even with leading spaces; a plain inner space is allowed and sent as %20. An interceptor's final req.endpoint is checked against its final req.baseEndpoint with the same rule; deliberately replacing the base with another valid absolute URL is allowed. Deterministic URL errors are never retried.

startupOpen is the initial snapshot for that call; a new call never reuses a previous call's snapshot, even if it fails before a response arrives. stream.open is live and changes on later physical opens. Keep the tuple value when the first response matters. startupOpen is the returned connection's first successful handshake snapshot only for a native Defjs handle or a wrapper forwarding the same live open object; custom handles and wrappers overriding open retain their current public open even after reconnect, so shared closed cannot bypass URL sanitization or header filtering, and Defjs does not reconstruct first-handshake metadata from arbitrary transformations.

import type {
  EventStreamHandle,
  EventStreamOpenInfo,
  Fault,
} from "@defjs/core";

type StreamResult<T> =
  | [error: null, stream: EventStreamHandle<T>, open: EventStreamOpenInfo]
  | [error: Fault, stream: undefined, open: EventStreamOpenInfo | undefined];

const result: StreamResult<string> | undefined = undefined;
void result;

Decode eventsCopy Anchor Link

Wire event name → events[eventName]; else events.default. No matching Struct → event not delivered. Missing SSE event field → logical name message.

SSE data starts as text. A string struct keeps that text. Every other kind is JSON: parse the text, then decode with the struct. struct.object, struct.number, struct.boolean, struct.any, and struct.unknown are JSON, not trimmed text. A bare token such as ping does not satisfy struct.literal('ping'); the data has to be JSON. A server that sends the JSON string "abc" still satisfies struct.string(): the delivered value is that raw text, including the quotes, and is not JSON-decoded to abc.

Parsing follows the WHATWG SSE line and field rules: CR, LF, and CRLF end lines; a blank line dispatches a data-bearing event; multiple data: lines join with \n. Comments and unknown fields are ignored, and an unfinished event at EOF is not dispatched. Only one leading UTF-8 BOM at the start of each physical stream is removed. U+FEFF inside data or id values is preserved; a later BOM-prefixed field name such as data is unknown and ignored.

Emitted value: event, decoded data, optional non-empty id. With default, unknown event names are string in the inferred union.

Observe invalid eventsCopy Anchor Link

Invalid/undeclared events are dropped, not queued. withSSEOnInvalidEvent(...) can observe raw ID, name, text data, plus missing-struct or validation-failed and an optional cause.

import { createClient, withEndpoint, withSSEOnInvalidEvent } from "@defjs/core";

const client = createClient(
  withEndpoint("https://api.example.com"),
  withSSEOnInvalidEvent(({ reason, message, cause, signal }) => {
    if (signal.aborted) return;
    console.info("Dropped SSE event", {
      reason,
      event: message.event,
      hasCause: cause !== undefined,
    });
  }),
);

Observer runs at the transform boundary. Its failure is isolated unless the active attempt signal is aborted. Keep it short; don’t treat raw event data as trusted.

ReconnectCopy Anchor Link

Behavior change: SSE reconnect is opt-in. Without withSSEReconnect(...), network and stream-read failures are not retried (same as attempts: 0). Pass withSSEReconnect({ attempts: 3 }) (or another reviewed budget) to retry those failures. Normal EOF is not retried. Status/content-type validation, parser limits, message transform failures, queue overflow, and normal EOF stay terminal for the logical stream.

import { createClient, withEndpoint, withSSEReconnect } from "@defjs/core";

const client = createClient(
  withEndpoint("https://api.example.com"),
  withSSEReconnect({
    attempts: 3,
    delayMs: 1_000,
    factor: 2,
    maxDelayMs: 30_000,
    jitter: 0.5,
    shouldReconnect({ lastEventId, method }) {
      // A POST is replayed only when the server can resume it from Last-Event-ID.
      return method === "GET" || lastEventId !== "";
    },
  }),
);

attempts is the retry budget for each outage; it defaults to 3 when a reconnect object is set. A successful handshake resets the budget, callback attempt, and backoff exponent, so separate outages do not spend a lifetime budget. attempts: 0 disables retry. Omitting withSSEReconnect also disables retry. delayMs is the initial interval (default 1000); factor grows it (default 1, a constant delay); maxDelayMs is a hard cap on the final delay, after jitter and rounding (default: no configured cap). SSE jitter is a 0–1 multiplicative factor, same as WebSocket. The platform timer limit is 2_147_483_647 ms even without a configured cap. A zero base stays zero; overflow from otherwise valid numbers saturates at the caps. A digits-only stream retry: field updates the current base interval, including after recovery. Policy callback returning false / throwing / rejecting ends the logical stream. Repeated successful opens followed immediately by retryable stream failures can keep reconnecting at the current base delay (subject to jitter and caps), because each open resets the budget; use shouldReconnect to stop them. Normal EOF remains terminal.

Reconnect options are checked before interceptors or Fetch: attempts must be a non-negative safe integer; delayMs must be finite and non-negative; maxDelayMs must be a non-negative finite number or positive Infinity, which removes the configured cap but not the platform limit of 2_147_483_647 ms; factor must be finite and positive; jitter must be finite and within 0–1. Invalid values return REQ_OPTIONS_INVALID without opening a stream.

Latest parsed event ID becomes Last-Event-ID on a later attempt, encoded as UTF-8 bytes on the wire, including non-ASCII IDs. An empty id: clears the header; an ID containing NUL is ignored. Apart from that header, a later attempt resends the request built for the first one: the builder and interceptors run once per call, so method, URL, headers, and body stay the same. A header that must be fresh on every attempt, such as a short-lived token, belongs in the Fetch passed to withSSEHandle, which runs once per attempt. If a credential still expires, a non-2xx reconnect handshake after the stream has opened ends it with RES_DECODE_FAILED; inspect stream.open.response.status for the status. A non-2xx initial handshake instead returns a startup HTTP_STATUS fault when its declared error body decodes.

Reconnect resends a request only when that is safe:

• GET, HEAD, OPTIONS, PUT, and DELETE are idempotent (RFC 9110 §9.2.2) and reconnect within the budget.
• Any other method, such as POST or PATCH, reconnects only when shouldReconnect returns true. A failed connection cannot tell whether the server already acted: a POST that starts an LLM generation would run, and be billed, again, and its new events would follow the old ones in the same stream. Passing shouldReconnect replaces this default for every method, so check method in its context when one client serves both kinds of endpoint.
• A ReadableStream body is consumed by the first attempt. Any failure, before or after the stream opens, ends the stream with the original error; shouldReconnect is not called and no delay is spent. Use a string, bytes, Blob, form, or JSON body when the request must reconnect.

Buffer and queue limitsCopy Anchor Link

Both must be positive safe integers. Overflow is fatal — no silent discard of older events.

Fatal stream also clears buffered events, cancels the active body, rejects the iterator, and resolves stream.closed with kind: 'error'.

Close and disposeCopy Anchor Link

EventStreamHandle: one live opening snapshot, one terminal promise, one close, one async iterator, and one standard async disposer.

import type {
  EventStreamCloseInfo,
  EventStreamHandle,
  EventStreamOpenInfo,
} from "@defjs/core";

interface StreamApi<T> extends AsyncIterable<T>, AsyncDisposable {
  readonly open: EventStreamOpenInfo;
  readonly closed: Promise<EventStreamCloseInfo>;
  close(reason?: unknown): void;
  [Symbol.asyncDispose](): PromiseLike<void>;
}

const handle = null as unknown as EventStreamHandle<string>;
const api: StreamApi<string> = handle;
void api;

Terminal kinds (closed.kind): eof, aborted, or error. An error result also carries closed.code, an EventStreamFaultCode: CAP_BUFFER_EXCEEDED, CAP_QUEUE_OVERFLOW, EXT_HOOK_FAILED, EXT_OBSERVER_FAILED, NET_TIMEOUT, NET_UNREACHABLE, RES_DECODE_FAILED, or RES_MEDIA_TYPE_INVALID.

close(reason) aborts the active attempt, closes the queue, settles as aborted, even when the reason is a TimeoutError. The signal passed to the operation stays attached after the stream is open and covers the whole connection. Aborting it normally ends the iterator and reports aborted. If the abort reason is a TimeoutError or ERR_TIMEOUT (for example AbortSignal.timeout(...)), the iterator throws ERR_TIMEOUT and closed reports kind: 'error' with code: 'NET_TIMEOUT'. There is no startup-only deadline on the client. An application that wants one creates its own AbortController, aborts it with new DOMException("...", "TimeoutError") from a timer, and clears the timer after the await returns:

import {
  createClient,
  defineEventStream,
  struct,
  withEndpoint,
} from "@defjs/core";

const client = createClient(withEndpoint("https://api.example.com"));
const notifications = defineEventStream({
  maxBufferSize: 64 * 1024,
  maxQueueSize: 100,
  path: "/notifications",
  events: { message: struct.string() },
});

async function openDuringStartup(abortAfterMs: number) {
  const controller = new AbortController();
  const timer = setTimeout(() => {
    controller.abort(new DOMException("startup deadline", "TimeoutError"));
  }, abortAfterMs);
  const [error, stream] = await notifications(
    client({ signal: controller.signal }),
  );
  clearTimeout(timer);
  if (error) {
    console.error(error.code);
    return;
  }
  stream.close("opened");
  await stream.closed;
}

void openDuringStartup;

Loop break / return / throw invokes iterator return and closes with iterator-return. The code that starts the operation owns closure.

await using calls the same owned close path and waits until Defjs reading/reconnect work has stopped and the active reader lock has been released. It is intentionally bounded around a provider-controlled ReadableStream.cancel() promise that never settles, so it does not guarantee that such a provider promise completed. Explicit stream.close(reason) plus await stream.closed is still valid when you need a reason or the terminal result.

Custom structural EventStreamHandle implementations must now provide [Symbol.asyncDispose](): PromiseLike<void> and connect it to the same lifecycle. That is a compile-time breaking change for implementers; consumers that only receive a Defjs handle do not need to invoke anything new at runtime.

The supported minimum lib set is ES2022, ESNext.Disposable, DOM, and DOM.Iterable. Repository declaration tooling uses TypeScript 7, but there is no automated packed-consumer or minimum-lib verification gate, and older compilers are not promised. This set supplies types, not runtime Web API polyfills. A Client is not AsyncDisposable; manage each request with its AbortSignal.

Keep credentials, event data, event IDs, causes, and stream URLs out of routine logs. withCredentials(true) affects Fetch cookies for SSE; it does not configure WebSocket auth.

Related recipesCopy Anchor Link

• Consume an SSE stream
• Cancel an HTTP call