Streaming

Streaming

@chttp/stream adds Server-Sent Events and chunked text responses to @chttp/router. It is authored in the scriptc subset, so it compiles into the same native binary as your app.

import { createRouter } from "@chttp/router";
import { streamSSE } from "@chttp/stream";

const app = createRouter();

app.get("/clock", (c) =>
  streamSSE(c, (s) => {
    const t = setInterval(() => s.send(new Date().toISOString(), "tick"), 1000);
    s.onAbort(() => clearInterval(t));
  }),
);

How it works

streamSSE(c, fn) sends the status line and headers, flushes them, marks the response as sent (c.sent = true), then returns fn(stream). Marking it sent matters: the router answers 500 when a handler settles without writing anything, and a stream callback usually returns before the first byte. Because the helper returns whatever fn returns, an async callback is awaited by the router like any async handler.

The response is 200 text/event-stream with cache-control: no-cache, no-transform and x-accel-buffering: no. Headers set with c.header(...) before the call are kept.

SSEStream

Multi-line data is split on LF, CRLF or CR into one data: line each, as the SSE spec requires. CR and LF are stripped from event and id, so a value cannot inject extra fields. Every write returns true while the stream is open and false once it is closed.

onAbort is wired to the response close event. It fires when the client leaves and also after your own close(), so it is the one place to clear timers and drop subscriptions. Registering after the connection closed runs the callback immediately.

streamText

streamText(c, fn) is the same shape for text/plain chunked output. The stream has write(s), writeln(s), close() and onAbort(fn).

app.get("/log", (c) =>
  streamText(c, (s) => {
    s.writeln("starting");
    s.writeln("done");
    s.close();
  }),
);

Slow consumers

scriptc 0.3.1 cannot observe backpressure: the result of write(), the drain event and writableLength are all refused at compile time. A client that stops reading would otherwise make the server buffer without bound.

As a stand-in, every stream has a pending-bytes cap. If more than maxPending bytes (default 1 MiB) are written inside one windowMs window (default 1000 ms), the stream is closed and the write returns false. Both are plain fields, so tune them per stream:

streamSSE(c, (s) => {
  s.maxPending = 64 * 1024;
  s.windowMs = 1000;
});

This is a rate window, not true queue depth. Once scriptc models drain, the cap will track real pending bytes.

Compiling and testing

examples/sse-clock is a complete app. Build it with scriptc build src/main.ts -o app, run it, then watch events arrive:

curl -N http://localhost:3000/clock

Stop curl and the interval is cleared by the onAbort handler. Always register that handler when your callback starts a timer.

Next: Errors.