Streaming data

// pull source: called whenever the queue wants more
function counter(limit: number): ReadableStream<number> {
  let n = 0;
  return new ReadableStream<number>({
    pull(controller) {
      if (n >= limit) {
        controller.close();
        return;
      }
      controller.enqueue(n);
      n += 1;
    },
    cancel(reason) {
      // consumer gave up: release resources here
    },
  });
}
 
// push source: start() wires up events once
function ticks(ms: number): ReadableStream<number> {
  let id: ReturnType<typeof setInterval> | undefined;
  return new ReadableStream<number>({
    start(controller) {
      const tick = () => controller.enqueue(Date.now());
      id = setInterval(tick, ms);
    },
    cancel() {
      clearInterval(id);
    },
  });
}

async function readAll(stream: ReadableStream<string>) {
  const reader = stream.getReader();
  const parts: string[] = [];
  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      parts.push(value);
    }
  } finally {
    reader.releaseLock();
  }
  return parts.join("");
}

const received: string[] = [];
const sink = new WritableStream<string>({
  write(chunk) {
    received.push(chunk); // may return a promise
  },
  close() {},
  abort(reason) {},
});
 
const writer = sink.getWriter();
await writer.ready; // wait for backpressure to clear
await writer.write("a");
await writer.close();
 
const upper = new TransformStream<string, string>({
  transform(chunk, controller) {
    controller.enqueue(chunk.toUpperCase());
  },
  flush(controller) {
    controller.enqueue("\n"); // after the last chunk
  },
});

declare const source: ReadableStream<string>;
declare const upper: TransformStream<string, string>;
declare const sink: WritableStream<string>;
 
const ac = new AbortController();
await source
  .pipeThrough(upper) // returns upper.readable
  .pipeTo(sink, { signal: ac.signal }); // resolves when done
 
const [forLog, forUser] = source.tee(); // two readers

// count chunks (default: highWaterMark 1)
const byCount = new CountQueuingStrategy({
  highWaterMark: 16,
});
 
// count bytes: buffer up to 64 KiB before pausing the source
const byBytes = new ByteLengthQueuingStrategy({
  highWaterMark: 64 * 1024,
});
 
declare const source: UnderlyingDefaultSource<Uint8Array>;
const stream = new ReadableStream(source, byBytes);
 
// new TransformStream(transformer, writable, readable)
const buffered = new TransformStream({}, byCount, byCount);

async function fetchGzippedText(url: string) {
  const res = await fetch(url);
  if (!res.ok || !res.body) throw new Error(`${res.status}`);
  const text = res.body
    .pipeThrough(new DecompressionStream("gzip"))
    .pipeThrough(new TextDecoderStream()); // string chunks
  // not new Response(text): Response bodies must be bytes
  return (await Array.fromAsync(text)).join("");
}
 
async function gzip(text: string): Promise<Blob> {
  const compressed = new Blob([text])
    .stream()
    .pipeThrough(new CompressionStream("gzip"));
  return new Response(compressed).blob();
}

declare const stream: ReadableStream<string>;
 
for await (const chunk of stream) {
  if (chunk === "STOP") break; // cancels the stream
}
 
// keep the stream alive after breaking out
declare const other: ReadableStream<string>;
for await (
  const c of other.values({ preventCancel: true })
) {
  break;
}
 
// collect everything
const all = await Array.fromAsync(other);

function fromIterable<T>(
  iterable: AsyncIterable<T> | Iterable<T>,
): ReadableStream<T> {
  const it =
    Symbol.asyncIterator in iterable
      ? iterable[Symbol.asyncIterator]()
      : iterable[Symbol.iterator]();
  return new ReadableStream<T>({
    async pull(controller) {
      const { value, done } = await it.next();
      if (done) controller.close();
      else controller.enqueue(value);
    },
    async cancel(reason) {
      await it.return?.(reason);
    },
  });
}
 
async function* pages(url: string) {
  for (let page = 1; page <= 3; page++) {
    const res = await fetch(`${url}?page=${page}`);
    yield (await res.json()) as unknown;
  }
}
 
const results = fromIterable(pages("/api/items"));

import {
  createReadStream,
  createWriteStream,
} from "node:fs";
import { Readable, Transform } from "node:stream";
import { pipeline } from "node:stream/promises";
 
const upper = new Transform({
  transform(chunk: Buffer, _encoding, callback) {
    callback(null, chunk.toString("utf8").toUpperCase());
  },
});
 
await pipeline(
  createReadStream("in.txt"),
  upper,
  createWriteStream("out.txt"),
);
 
// async generators work as pipeline stages
await pipeline(
  createReadStream("in.txt", { encoding: "utf8" }),
  async function* (source: AsyncIterable<string>) {
    for await (const text of source) yield text.trim();
  },
  createWriteStream("trimmed.txt"),
);
 
// object mode: chunks are any JS value
const rows = Readable.from([{ id: 1 }, { id: 2 }]);
for await (const row of rows) {
  // row: any; annotate or validate
}

import { createReadStream } from "node:fs";
import { Readable } from "node:stream";
import type {
  ReadableStream as NodeRS,
} from "node:stream/web";
 
declare const res: Response;
 
// web to Node: any async iterable works with Readable.from
const nodeA = Readable.from(res.body ?? []);
// or fromWeb; the cast bridges DOM vs node:stream/web types
const nodeB = Readable.fromWeb(
  res.body as unknown as NodeRS<Uint8Array>,
);
 
// Node to web, e.g. to return from a route handler
const web = Readable.toWeb(createReadStream("big.bin"));
const response = new Response(
  web as unknown as ReadableStream,
);

async function download(
  url: string,
  onProgress: (received: number, total?: number) => void,
): Promise<Blob> {
  const res = await fetch(url);
  if (!res.ok || !res.body) throw new Error(`${res.status}`);
  const length = Number(res.headers.get("content-length"));
  const total = length > 0 ? length : undefined;
 
  const chunks: Uint8Array<ArrayBuffer>[] = [];
  let received = 0;
  for await (const chunk of res.body) {
    chunks.push(chunk);
    received += chunk.byteLength;
    onProgress(received, total);
  }
  return new Blob(chunks);
}

const encoder = new TextEncoder();
const sleep = (ms: number) =>
  new Promise((resolve) => setTimeout(resolve, ms));
 
export function GET(req: Request): Response {
  const stream = new ReadableStream<Uint8Array>({
    async start(controller) {
      for (const word of ["streams", "are", "neat"]) {
        if (req.signal.aborted) break; // client left
        controller.enqueue(encoder.encode(`${word}\n`));
        await sleep(300);
      }
      controller.close();
    },
  });
  return new Response(stream, {
    headers: {
      "Content-Type": "text/plain; charset=utf-8",
      "X-Accel-Buffering": "no", // nginx: do not buffer
    },
  });
}

declare function makeStream(): ReadableStream<Uint8Array>;
 
const server = Bun.serve({
  port: 3000,
  idleTimeout: 0, // seconds; default 10 closes quiet streams
  routes: {
    "/stream": () => new Response(makeStream()),
  },
  fetch() {
    return new Response("Not found", { status: 404 });
  },
});

retry: 5000
: comment lines start with a colon (use as heartbeat)
 
id: 41
event: price
data: {"sym":"ACME","px":12.5}
 
id: 42
data: first line
data: second line
 

type ServerEvents = {
  price: { sym: string; px: number };
  notice: { text: string };
};
 
function on<K extends keyof ServerEvents>(
  source: EventSource,
  type: K,
  handler: (data: ServerEvents[K]) => void,
): () => void {
  const listener = (e: MessageEvent<string>) => {
    // validate with Zod in real code; the wire is untrusted
    handler(JSON.parse(e.data) as ServerEvents[K]);
  };
  source.addEventListener(type, listener);
  return () => source.removeEventListener(type, listener);
}
 
const source = new EventSource("/api/prices", {
  withCredentials: false, // true sends cookies cross-origin
});
const off = on(source, "price", ({ sym, px }) => {});
source.onerror = () => {
  // readyState CONNECTING: retrying; CLOSED: gave up
};
// later: off(); source.close();

const encoder = new TextEncoder();
 
function sseMessage(
  data: unknown,
  opts: { event?: string; id?: string } = {},
): Uint8Array {
  const lines = [
    opts.event && `event: ${opts.event}`,
    opts.id && `id: ${opts.id}`,
    `data: ${JSON.stringify(data)}`,
  ].filter(Boolean);
  return encoder.encode(`${lines.join("\n")}\n\n`);
}
 
export function GET(req: Request): Response {
  const lastId = Number(
    req.headers.get("last-event-id") ?? 0,
  );
  let timer: ReturnType<typeof setInterval> | undefined;
 
  const stream = new ReadableStream<Uint8Array>({
    start(controller) {
      let id = lastId; // resume after the last one seen
      timer = setInterval(() => {
        id += 1;
        const msg = sseMessage(
          { sym: "ACME", px: 10 + Math.random() },
          { event: "price", id: String(id) },
        );
        controller.enqueue(msg);
      }, 1000);
    },
    cancel() {
      clearInterval(timer); // client disconnected
    },
  });
 
  return new Response(stream, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      "X-Accel-Buffering": "no",
    },
  });
}

function splitLines(): TransformStream<string, string> {
  let buffer = "";
  return new TransformStream<string, string>({
    transform(chunk, controller) {
      buffer += chunk;
      const lines = buffer.split(/\r?\n/);
      buffer = lines.pop() ?? ""; // keep the partial line
      for (const line of lines) controller.enqueue(line);
    },
    flush(controller) {
      if (buffer) controller.enqueue(buffer);
    },
  });
}

import { z } from "zod";
 
declare function splitLines(): TransformStream<
  string,
  string
>;
 
function parseNdjson<S extends z.ZodType>(
  schema: S,
): TransformStream<string, z.infer<S>> {
  return new TransformStream<string, z.infer<S>>({
    transform(line, controller) {
      if (line.trim() === "") return; // skip blank lines
      const result = schema.safeParse(JSON.parse(line));
      if (result.success) controller.enqueue(result.data);
      else controller.error(result.error);
    },
  });
}
 
const LogEntry = z.object({
  level: z.enum(["info", "warn", "error"]),
  msg: z.string(),
});
 
const res = await fetch("/api/logs.ndjson");
if (!res.body) throw new Error("No body");
const entries = res.body
  .pipeThrough(new TextDecoderStream())
  .pipeThrough(splitLines())
  .pipeThrough(parseNdjson(LogEntry));
 
for await (const entry of entries) {
  if (entry.level === "error") console.error(entry.msg);
}

function map<I, O>(
  fn: (chunk: I) => O | Promise<O>,
): TransformStream<I, O> {
  return new TransformStream<I, O>({
    async transform(chunk, controller) {
      controller.enqueue(await fn(chunk));
    },
  });
}
 
function filter<T, S extends T>(
  pred: (chunk: T) => chunk is S,
): TransformStream<T, S> {
  return new TransformStream<T, S>({
    transform(chunk, controller) {
      if (pred(chunk)) controller.enqueue(chunk);
    },
  });
}
 
async function first<T>(
  stream: ReadableStream<T>,
): Promise<T | undefined> {
  const reader: ReadableStreamDefaultReader<T> =
    stream.getReader();
  try {
    const result = await reader.read();
    return result.done ? undefined : result.value;
  } finally {
    await reader.cancel(); // done with the rest
  }
}
 
declare const numbers: ReadableStream<number | null>;
const doubled = numbers
  .pipeThrough(filter((n): n is number => n !== null))
  .pipeThrough(map((n) => n * 2)); // ReadableStream<number>

async function streamInto(
  el: HTMLElement,
  url: string,
  body: unknown,
  signal?: AbortSignal,
): Promise<void> {
  const res = await fetch(url, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify(body),
    signal, // "Stop" button: controller.abort()
  });
  if (!res.ok || !res.body) throw new Error(`${res.status}`);
  const reader = res.body
    .pipeThrough(new TextDecoderStream())
    .getReader();
  el.textContent = "";
  for (;;) {
    const { done, value } = await reader.read();
    if (done) break;
    el.append(value); // a text node: never parsed as HTML
  }
}

type Row = { id: number; email: string };
declare function queryRows(): AsyncIterable<Row>; // cursor
 
const encoder = new TextEncoder();
 
export function GET(): Response {
  const rows = queryRows()[Symbol.asyncIterator]();
  const body = new ReadableStream<Uint8Array>({
    async pull(controller) { // only when the client reads
      const { done, value } = await rows.next();
      if (done) controller.close();
      else controller.enqueue(
        encoder.encode(`${JSON.stringify(value)}\n`),
      );
    },
    async cancel() {
      await rows.return?.(); // client left: close the cursor
    },
  });
  return new Response(body, {
    headers: { "Content-Type": "application/x-ndjson" },
  });
}

type SseEvent = { event: string; data: string };
 
async function* sse(
  url: string,
  init: RequestInit, // e.g. headers: { Authorization }
): AsyncGenerator<SseEvent> {
  const res = await fetch(url, init);
  if (!res.ok || !res.body) throw new Error(`${res.status}`);
  const lines = res.body
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(splitLines());
  let event = "message";
  let buf: string[] = [];
  for await (const line of lines) {
    if (line === "") { // blank line ends an event
      if (buf.length) yield { event, data: buf.join("\n") };
      [event, buf] = ["message", []];
      continue;
    }
    const [field, ...rest] = line.split(":");
    const value = rest.join(":").replace(/^ /, "");
    if (field === "data") buf = [...buf, value];
    else if (field === "event") event = value;
  }
}

function batch<T>(size: number): TransformStream<T, T[]> {
  let buffer: T[] = [];
  return new TransformStream<T, T[]>({
    transform(item, controller) {
      buffer.push(item);
      if (buffer.length >= size) {
        controller.enqueue(buffer);
        buffer = [];
      }
    },
    flush(controller) {
      if (buffer.length > 0) controller.enqueue(buffer);
    },
  });
}
 
type Row = { id: number };
declare const rows: ReadableStream<Row>;
declare function insertMany(rows: Row[]): Promise<void>;
 
// one INSERT per 500 rows; backpressure waits for the DB
await rows.pipeThrough(batch<Row>(500)).pipeTo(
  new WritableStream({ write: (rs) => insertMany(rs) }),
);

import { createHash } from "node:crypto";
import { createReadStream } from "node:fs";
 
async function sha256File(path: string): Promise<string> {
  const hash = createHash("sha256");
  for await (const chunk of createReadStream(path)) {
    hash.update(chunk as Buffer); // constant memory
  }
  return hash.digest("hex");
}
 
// Bun: native hasher over a web stream
async function sha256Bun(path: string): Promise<string> {
  const hasher = new Bun.CryptoHasher("sha256");
  for await (const chunk of Bun.file(path).stream()) {
    hasher.update(chunk);
  }
  return hasher.digest("hex");
}