skills/docs/foundation/flow/streams/src/transforms/coalesce-bytes/coalesce-bytes.md

coalesce-bytes

(no submodules)

Purpose

coalesceBytes and CoalesceBytesTransform accumulate small inbound Uint8Array chunks into larger frames before emitting them downstream. They exist to bridge byte producers that hand out tiny fragments (network sockets, serial ports, decoders) with consumers that prefer fewer, larger writes (framing protocols, hash functions, syscalls), without forcing every producer to implement its own buffering and flush policy. Coalescing is driven by two independent triggers: a size threshold (minBytes) and a latency cap (maxLatencyMs) measured from the moment the first chunk in the current buffer arrived. On stream close any remainder is always emitted, so no bytes are silently dropped.

Features

• Size-driven flush when accumulated bytes reach minBytes.
• Latency-driven flush when maxLatencyMs ms have elapsed since the first buffered chunk (clock-based, evaluated on the next chunk arrival).
• Injectable Clock from @suaveplan/types for fully deterministic tests.
• Final-frame emission on stream close — no bytes are ever dropped.
• AbortSignal integration: aborted writes reject with StreamAbortedError.
• Plugin-OO wrapper CoalesceBytesTransform for use with BaseTransform, hooks, listeners, and the StreamManager pipeline.
• Strict input validation: minBytes <= 0 and maxLatencyMs < 0 both throw RangeError at construction time.

Usage: Basic Example

import { coalesceBytes } from "@suaveplan/streams/transforms/coalesce-bytes/coalesce-bytes";

const coalesce = coalesceBytes(1024);
const result = source.pipeThrough(coalesce);

Usage: Advanced Example

import { CoalesceBytesTransform } from "@suaveplan/streams/transforms/coalesce-bytes/coalesce-bytes";

let now = 0;
const t = new CoalesceBytesTransform({
    minBytes: 4096,
    maxLatencyMs: 50,
    clock: () => now, // deterministic in tests
});

await source.pipeTo(
    new WritableStream({
        write(chunk) {
            // ...
        },
    }),
    {
        // pipe through the OO wrapper
        preventClose: false,
    },
);

API Reference

• coalesceBytes(minBytes: number, options?: CoalesceBytesOptions): TransformStream<Uint8Array, Uint8Array> — factory.
• CoalesceBytesOptions:
  • maxLatencyMs?: number (default Infinity)
  • clock?: Clock (default () => Date.now())
  • signal?: AbortSignal
• class CoalesceBytesTransform extends BaseTransform
  • constructor(config: CoalesceBytesTransformConfig)
  • process<T>(input: T): Promise<T> — single-shot no-op pass-through.
  • toTransformStream<A, B>(): TransformStream<A, B> — returns a fresh coalescing stream wired to the constructor config.
• Throws RangeError for minBytes <= 0 or maxLatencyMs < 0.
• Throws StreamAbortedError on aborted writes.

Implementation Notes

The latency check is evaluated lazily on every chunk arrival rather than through a real timer. This keeps the implementation deterministic under an injected clock — tests advance time and write a probe chunk to flush. In production, byte producers naturally tick frequently enough that this is the desired behaviour. Buffered chunks are stored as a list of Uint8Array references; only the final emission allocates a contiguous output buffer (new Uint8Array(bufferedSize) + set() per chunk), so the steady-state allocation is one buffer per emitted frame. The single-shot process() is a pass-through because coalescing is fundamentally a streaming operation that needs multiple inputs — real coalescing happens via toTransformStream().