94 lines
3.2 KiB
TypeScript
94 lines
3.2 KiB
TypeScript
/**
|
|
* Streaming response primitives.
|
|
*
|
|
* `streamResponse` turns a (sync or async) iterable of strings/bytes into a
|
|
* streaming `Response` — the basis for streaming SSR (send the shell, then flush
|
|
* page chunks as they render) and any progressively-generated output. `sse`
|
|
* builds a Server-Sent Events stream from an async iterable of events.
|
|
*
|
|
* API routes and pages can already return a `Response` with a `ReadableStream`
|
|
* body and the framework streams it unbuffered; these helpers just make the
|
|
* common cases ergonomic.
|
|
*/
|
|
|
|
export interface StreamResponseInit {
|
|
status?: number;
|
|
headers?: HeadersInit;
|
|
/** Content-Type; default "text/html; charset=utf-8". */
|
|
contentType?: string;
|
|
}
|
|
|
|
type Chunk = string | Uint8Array;
|
|
type ChunkSource = Iterable<Chunk> | AsyncIterable<Chunk>;
|
|
|
|
/** Build a streaming Response from an (async) iterable of chunks. */
|
|
export function streamResponse(source: ChunkSource, init: StreamResponseInit = {}): Response {
|
|
const encoder = new TextEncoder();
|
|
const iterator = getIterator(source);
|
|
|
|
const stream = new ReadableStream<Uint8Array>({
|
|
async pull(controller) {
|
|
try {
|
|
const { done, value } = await iterator.next();
|
|
if (done) {
|
|
controller.close();
|
|
return;
|
|
}
|
|
controller.enqueue(typeof value === "string" ? encoder.encode(value) : value);
|
|
} catch (err) {
|
|
controller.error(err);
|
|
}
|
|
},
|
|
async cancel() {
|
|
await iterator.return?.(undefined);
|
|
},
|
|
});
|
|
|
|
const headers = new Headers(init.headers);
|
|
if (!headers.has("content-type")) {
|
|
headers.set("content-type", init.contentType ?? "text/html; charset=utf-8");
|
|
}
|
|
// Tell the server's compressor (and proxies) not to buffer/transform a stream.
|
|
if (!headers.has("cache-control")) headers.set("cache-control", "no-transform");
|
|
return new Response(stream, { status: init.status ?? 200, headers });
|
|
}
|
|
|
|
export interface ServerSentEvent {
|
|
data: string;
|
|
event?: string;
|
|
id?: string;
|
|
/** Client reconnection hint in milliseconds. */
|
|
retry?: number;
|
|
}
|
|
|
|
/** Build a Server-Sent Events (text/event-stream) Response from events. */
|
|
export function sse(source: Iterable<ServerSentEvent> | AsyncIterable<ServerSentEvent>): Response {
|
|
const iterator = getIterator(source);
|
|
async function* frames(): AsyncGenerator<string> {
|
|
for (;;) {
|
|
const { done, value } = await iterator.next();
|
|
if (done) return;
|
|
yield formatEvent(value);
|
|
}
|
|
}
|
|
return streamResponse(frames(), {
|
|
contentType: "text/event-stream",
|
|
headers: { "cache-control": "no-cache, no-transform", connection: "keep-alive" },
|
|
});
|
|
}
|
|
|
|
function formatEvent(e: ServerSentEvent): string {
|
|
let out = "";
|
|
if (e.event) out += `event: ${e.event}\n`;
|
|
if (e.id) out += `id: ${e.id}\n`;
|
|
if (e.retry !== undefined) out += `retry: ${Math.floor(e.retry)}\n`;
|
|
for (const line of e.data.split("\n")) out += `data: ${line}\n`;
|
|
return out + "\n";
|
|
}
|
|
|
|
function getIterator<T>(source: Iterable<T> | AsyncIterable<T>): AsyncIterator<T> | Iterator<T> {
|
|
const asAsync = (source as AsyncIterable<T>)[Symbol.asyncIterator];
|
|
if (typeof asAsync === "function") return asAsync.call(source);
|
|
return (source as Iterable<T>)[Symbol.iterator]();
|
|
}
|