Consume an SSE stream

One async iterator per stream. Put close + await stream.closed in finally so early exits don’t leak the Fetch attempt.

See SSE for details.

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;
  }
  try {
    for await (const event of stream) {
      if (event.event === "message") {
        console.log(event.data.text);
        break;
      }
    }
  } finally {
    stream.close("consumer-finished");
    await stream.closed;
  }
}

void main();

api.example.com is a placeholder. With a server whose first event is message with the data { "text": "hello from the server" }, the snippet prints:

hello from the server

SSE does not retry network/read failures unless you enable withSSEReconnect(...). With a reconnect object, attempts defaults to 3 per outage; a successful handshake resets the budget and backoff exponent. attempts: 0 disables retry. Invalid numeric reconnect options fail before any request with REQ_OPTIONS_INVALID; see validation rules. Without shouldReconnect, only GET, HEAD, OPTIONS, PUT, and DELETE requests reconnect, and a ReadableStream request body never does.

Consume an SSE streamCopy Anchor Link

On this page

Consume an SSE stream