Streaming responses

DaloyJS ships first-class helpers for two streaming formats that are common in HTTP APIs: Server-Sent Events (SSE) and newline-delimited JSON (NDJSON). Both helpers wrap an AsyncIterable in a backpressure-safe ReadableStream: the underlying iterator is only advanced when the consumer pulls the next chunk, so a slow client cannot cause unbounded memory growth.

They also honor an optional AbortSignal and call iterator.return() when the client disconnects, so any caller-owned resources (DB cursors, upstream fetches, message-queue subscriptions) get released cleanly.

The helpers live in the main barrel and in the /streaming subpath:

import {
  sseStream,
  sseResponse,
  ndjsonStream,
  ndjsonResponse,
} from "@daloyjs/core";

// Or, if you want a tree-shake-friendly subpath:
import { sseStream } from "@daloyjs/core/streaming";

Server-Sent Events (SSE)

Yield either a string (sent as data: ...) or an SSEMessage object with any combination of event, id, retry, comment, and data. Multi-line data and comment values are split into one data: (or :) line per source line, treating CRLF, a lone LF, and a lone CR as line breaks, exactly like the browser's EventSource parser does. A lone \r can therefore never smuggle a forged field into the frame (since 1.3.7).

• event: any run of CR/LF is replaced with a single space.
• id: any run of CR, LF, or NUL is replaced with a single space. A NUL would otherwise make EventSource silently ignore the id and defeat Last-Event-ID resumption (since 1.3.7).

// Untrusted input is safe to forward: the lone "\r" becomes a line break
// inside the data field, not a new "event:" field.
yield { event: "chat", id: "a\0b", data: "hello\revent: admin" };
// event: chat
// id: a b
// data: hello
// data: event: admin

import { sseStream } from "@daloyjs/core";

app.get(
  "/events",
  {
    operationId: "events",
    responses: { 200: { description: "SSE stream" } },
  },
  ({ request }) => ({
    status: 200 as const,
    headers: { "content-type": "text/event-stream" },
    body: sseStream(
      async function* () {
        for (let i = 0; i < 5; i++) {
          yield { event: "tick", id: String(i), data: { now: Date.now() } };
          await new Promise((r) => setTimeout(r, 1_000));
        }
      },
      { signal: request.signal, keepAliveMs: 15_000 }
    ),
  }),
);

Use sseResponse(...) when you want a fully-formed Response with the standard SSE headers (text/event-stream, cache-control: no-cache, no-transform, connection: keep-alive, and x-accel-buffering: no) already set:

import { sseResponse } from "@daloyjs/core";

const res = sseResponse(async function* () {
  yield { event: "ping", data: "hi" };
});

A handler may also return a raw Response directly, instead of the { status, body, headers } shape. Because it bypasses response-schema validation, the route must explicitly set acknowledgeNoResponseBodySchema: true. It is still finalized like any other response: the request id, secureHeaders(), CORS, your onSend hooks, and fingerprint stripping all still apply. This is what lets you forward a stream from a library such as the Vercel AI SDK in one line:

app.get(
  "/ping",
  {
    operationId: "ping",
    acknowledgeNoResponseBodySchema: true,
    responses: { 200: { description: "SSE stream" } },
  },
  // Return the Response as-is. Useful for AI SDK streams or a
  // forwarded upstream fetch() response.
  () => sseResponse(async function* () {
    yield { event: "ping", data: "hi" };
  }),
);

Keep-alive comments

Pass keepAliveMs to send a : keep-alive comment frame at a fixed interval. This prevents idle proxies from closing the connection while no events are flowing.

Newline-delimited JSON (NDJSON)

Yield any JSON-serializable value. Each value is encoded with JSON.stringify and terminated with a single \n. Strings are emitted as JSON strings, and values that cannot be represented as JSON throw instead of emitting invalid NDJSON.

import { ndjsonStream } from "@daloyjs/core";

app.get(
  "/exports/users.ndjson",
  {
    operationId: "exportUsers",
    responses: { 200: { description: "NDJSON dump" } },
  },
  ({ request }) => ({
    status: 200 as const,
    headers: { "content-type": "application/x-ndjson" },
    body: ndjsonStream(
      (async function* () {
        for await (const user of db.users.cursor()) {
          yield user;
        }
      })(),
      { signal: request.signal }
    ),
  }),
);

ndjsonResponse(...) builds the same stream with application/x-ndjson headers pre-set.

Backpressure & cancellation

Both helpers use the pull() entry point of ReadableStream: they call iterator.next() exactly once per pull. The runtime decides when to pull: a slow client on a Node socket pulls slowly, a fast Cloudflare consumer pulls quickly. You never need to write throttling code.

When the request is aborted (client disconnects, request timeout fires, explicit AbortController.abort()), the stream is closed and iterator.return() is invoked so a generator's finally block runs and any underlying cursor/socket is released.

On the Node adapter, a client that hangs up mid-stream now cancels the response body stream itself (since 1.3.7). Node's pipe() only unpipes its source when the socket closes early, so the adapter destroys the source explicitly. That cancel reaches sseStream() / ndjsonStream(), which call iterator.return(), so your generator's finally block and the keep-alive timer are cleaned up even when you did not pass a signal:

app.get(
  "/feed",
  {
    operationId: "feed",
    acknowledgeNoResponseBodySchema: true,
    responses: { 200: { description: "SSE stream" } },
  },
  () =>
    sseResponse(async function* () {
      const sub = await queue.subscribe("feed");
      try {
        for await (const msg of sub) yield { data: msg };
      } finally {
        // Runs when the client disconnects.
        await sub.close();
      }
    }),
);

Compression and etag()

Both compression() and etag() would have to buffer a body before they can act on it, which would withhold every event from the client. Neither touches a stream (since 1.3.7):

• compression() skips text/event-stream, NDJSON / JSON Lines / application/json-seq, and any response with Cache-Control: no-transform. sseResponse() sets no-transform, so it is always sent uncompressed.
• etag() skips text/event-stream and application/x-ndjson, plus any body with no Content-Length or one above its maxBytes cap. Those responses are sent untagged.

You can keep both middlewares registered globally; streaming routes pass through unchanged. See Compression middleware.

Cross-runtime compatibility

The helpers only depend on web-standard ReadableStream and TextEncoder, so the same handler works identically on Node, Bun, Deno, and Cloudflare Workers. The DaloyJS response serializer recognizes a ReadableStream body when you set an explicit non-JSON content-type and forwards it to the runtime without buffering.

OpenAPI

OpenAPI 3.1 has no rich schema for streamed event payloads. Document streaming routes with a free-form 200 response ({ description }) and describe the event shape in prose, or attach an example string showing one or two frames.