# 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. **Diagram: A pull-driven stream** Participants: Async generator, DaloyJS helper, Client 1. **Client -> DaloyJS helper** (request) - Pull the next chunk - ReadableStream pull(), one per consumer read 2. **DaloyJS helper -> Async generator** (request) - Advance the iterator - iterator.next() called exactly once per pull 3. **DaloyJS helper -> Client** (response) - Encode and send one frame - SSE data: ... or one NDJSON line + \n 4. **DaloyJS helper -> Client** (async) - Optional keep-alive comment while idle - : keep-alive every keepAliveMs 5. **Client -> DaloyJS helper** (note) - Disconnect or abort - request.signal fires, iterator.return() runs finally The consumer drives the pace. A slow client pulls slowly, so the generator is only advanced when there is demand, then iterator.return() releases resources on disconnect. The helpers live in the main barrel and in the `/streaming` subpath: ```ts 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). ```ts // 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 ``` ```ts 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: ```ts 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](/docs/ai-sdk) in one line: ```ts 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. ```ts 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`: ```ts 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](/docs/security/compression). ## 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. --- Source: https://daloyjs.dev/docs/streaming