skills/docs/foundation/flow/streams/src/transforms/batch/batch.md

batch

(no submodules)

Purpose

The batch feature provides two coordinated APIs for grouping a stream of individual chunks into fixed-size arrays before passing them downstream: the functional factory batchStream and the plugin-OO wrapper class BatchTransform. Both accumulate chunks into arrays of a configured size and flush a partial trailing batch when the upstream closes. This is the stream equivalent of chunking an array with lodash.chunk — it groups a continuous flow of items into discrete pages for bulk operations like database bulk-inserts, API batch requests, or log aggregation.

The internal buffer is reset between batches, so there is no risk of the same chunk appearing in two batches. BatchTransform is a thin shell around batchStream: its streaming bridge delegates to the factory directly so semantics are identical between the two call paths. The class exists so consumers using the BaseTransform plugin contract (hooks, events, StreamManager, TransformFactory) can express a batching stage alongside other class-form transforms (MapTransform, FilterTransform, …) without mixing function-form and class-form APIs.

Features

• Fixed-size batching with partial-final-batch flush on stream close
• Synchronous processing — no async overhead per chunk
• AbortSignal support on both call paths
• Configurable highWaterMark (functional API)
• Throws RangeError immediately if size < 1 (both APIs)
• Plugin-OO BatchTransform integrates with BaseTransform hooks/events
• Single-shot BatchTransform.process(input) returns [input] — a one-element batch
• Type-parameterised on T so callers get T[] output inference

Usage: Basic Example

import { batchStream } from "@suaveplan/streams/transforms/batch/batch";
import { fromArray } from "@suaveplan/streams/readable/from-array/from-array";
import { toArray } from "@suaveplan/streams/sinks/to-array/to-array";

const result = await toArray(
  fromArray([1, 2, 3, 4, 5]).pipeThrough(batchStream(2))
);
console.log(result); // [[1, 2], [3, 4], [5]]

Usage: Advanced Example

import { BatchTransform } from "@suaveplan/streams/transforms/batch/batch";

const t = new BatchTransform<{ id: string }>({ size: 100 });

// Hook to add timing metadata before downstream consumers
t.hook("before-emit", (batch) => batch);

const ctrl = new AbortController();
setTimeout(() => ctrl.abort(new Error("shutdown")), 5_000);

await source
    .pipeThrough(t.toTransformStream({ signal: ctrl.signal }))
    .pipeTo(sinkWritable, { signal: ctrl.signal });

API Reference

batchStream<T>(size: number, options?: TransformOptions): TransformStream<T, T[]>

Parameters:

• size — number of chunks per batch; must be >= 1 (throws RangeError otherwise).
• options.signal — AbortSignal; errors the stream on abort.
• options.highWaterMark — WHATWG queue depth (default: 1).

Returns: TransformStream<T, T[]>.

Throws (immediate): RangeError if size < 1. Throws (via error channel): StreamAbortedError when the signal fires.

class BatchTransform<T = unknown>

constructor(config: BatchTransformConfig)

Throws RangeError if config.size < 1.

process<U>(input: U, options?: TransformOptions): Promise<U>

Returns [input] (a one-element batch) cast to U. Honors options.signal — rejects with StreamAbortedError (STREAM.ABORTED) when fired.

toTransformStream<A, B>(options?): TransformStream<A, B>

Returns a fresh WHATWG TransformStream produced by batchStream<T>(size, options). The readable side emits T[] arrays of length config.size, with a possibly-shorter final array when the upstream closes.

interface BatchTransformConfig

interface BatchTransformConfig {
    readonly size: number;
    readonly flushOnClose?: boolean; // reserved; currently always true
}

Implementation Notes

Batches are emitted as fresh arrays; the internal buffer is replaced on each flush rather than cleared so the emitted batch slice is not mutated by subsequent chunk accumulation. The flush callback handles the trailing partial batch. Signal checking happens in transform only; if the signal fires between chunks, the next pull from the source will trigger the abort check.

Batching is stateful — single-shot process is a degenerate case. A batch requires multiple inputs to fill, but process(input) is called once per pipeline execution and cannot accumulate across executions without leaking buffer state between unrelated callers. We chose the most useful single-shot interpretation: process(x) returns [x], treating each call as "batch this one item." Real batching happens on the streaming side via toTransformStream, where batchStream owns the buffer and emits arrays as they fill.

This trade-off is intentional. The alternative (instance-scoped buffer state used by process) would couple every caller to a hidden lifecycle they would have to flush manually, and would diverge from the BaseTransform contract (process is "operate on this input now"). Consumers that want true batching reach for toTransformStream; consumers that want per-item transformation under a uniform plugin surface still get a sensible process shape.

The flushOnClose flag on BatchTransformConfig is reserved for forward compatibility — batchStream currently always flushes its partial buffer on close, so the option has no behaviour to gate. It is documented now to avoid a breaking config change later. Type erasure on the streaming bridge follows the same pattern as MapTransform: callers supply method-level generics (t.toTransformStream<number, number[]>()) for strict typing; default generics fall through to unknown from the abstract base.