BatchStream

import { BatchStream } from "jsr:@ende/streams";

class

class BatchStream<T> extends TransformStream<T, T[]>

A TransformStream that groups input chunks into fixed-size batches. Emits a T[] once size input chunks have been collected and flushes any final non-empty partial batch when the input closes.

Input order is preserved.

For grouping an in-hand iterable instead of a stream, see chunk from @ende/collections. For resizing Uint8Array chunks at the byte level, see FixedChunkStream.

Example: Batch records for bulk upload

import { BatchStream } from "@ende/streams/batch-stream";
import { assertEquals } from "@ende/assert";

const records = ReadableStream.from([1, 2, 3, 4, 5])
  .pipeThrough(new BatchStream(2));

assertEquals(await Array.fromAsync(records), [[1, 2], [3, 4], [5]]);

Example: Type inference inside a pipeline

import { BatchStream } from "@ende/streams/batch-stream";
import { assertEquals } from "@ende/assert";

interface User {
  id: string;
}

const users: ReadableStream<User> = ReadableStream.from([
  { id: "a" },
  { id: "b" },
  { id: "c" },
]);

const batches: ReadableStream<User[]> = users.pipeThrough(
  new BatchStream(2),
);

assertEquals(await Array.fromAsync(batches), [
  [{ id: "a" }, { id: "b" }],
  [{ id: "c" }],
]);

Constructors

`constructor`

new BatchStream(size: number)

Constructs a new instance.

size: number: The number of input chunks per emitted batch. Must be a positive integer.

Source