suaveplan/packages/foundation/flow/streams/src/transforms/chunk-bytes/chunk-bytes.md

chunk-bytes

(no submodules)

Purpose

chunkBytes re-frames an arbitrary Uint8Array byte stream into fixed-size frames of exactly frameSize bytes. Upstream producers rarely emit byte chunks that align to a downstream protocol's frame boundary — file readers, sockets, and decompression streams all hand out variable-length buffers. This transform absorbs that variability so consumers can rely on stable frame geometry, which is the precondition for length-prefixed framing, fixed-block ciphers, and any sliding-window decoder.

Features

• Fixed-size output frames with caller-supplied frameSize.
• Three tail policies for the residual: emit (default), pad, drop.
• Zero-copy frame emission via Uint8Array.subarray for full frames.
• Synchronous RangeError when frameSize <= 0.
• Plugin-OO wrapper (ChunkBytesTransform) for use with BaseTransform-aware pipelines, with a process() that returns the full concatenated output.
• AbortSignal honoured by process() — rejects with StreamAbortedError.

Usage: Basic Example

import { chunkBytes } from "@suaveplan/streams/transforms/chunk-bytes/chunk-bytes";
import { fromBytes } from "@suaveplan/streams/readable/from-bytes/from-bytes";
import { toArray } from "@suaveplan/streams/sinks/to-array/to-array";

const input = new Uint8Array(16);
for (let i = 0; i < 16; i++) input[i] = i + 1;

const frames = await toArray(
    fromBytes(input, 7).pipeThrough(chunkBytes(4))
);
// frames.length === 4, every frame is exactly 4 bytes

Usage: Advanced Example

Pad the trailing partial frame with zeroes (useful when feeding a fixed-block cipher or a fixed-frame radio modem):

import { ChunkBytesTransform } from "@suaveplan/streams/transforms/chunk-bytes/chunk-bytes";

const t = new ChunkBytesTransform(8, { tail: "pad" });

const padded = await t.process(new Uint8Array([1, 2, 3, 4, 5]));
// padded === Uint8Array([1, 2, 3, 4, 5, 0, 0, 0])

// Or use it in a pipeline alongside any other transform:
readable.pipeThrough(t.toTransformStream()).pipeTo(downstream);

t.dispose();

Drop the trailing partial frame when downstream cannot handle short reads:

chunkBytes(1024, { tail: "drop" });

API Reference

chunkBytes(frameSize, options?)

• frameSize: number — exact byte count per output frame; must be > 0.
• options.tail?: 'emit' | 'pad' | 'drop' — residual policy. Defaults to 'emit'.
• Returns: TransformStream<Uint8Array, Uint8Array>.
• Throws: RangeError if frameSize <= 0. The stream errors with TypeError on non-Uint8Array chunks.

class ChunkBytesTransform extends BaseTransform

• new ChunkBytesTransform(frameSize, options?) — same parameters; throws RangeError synchronously.
• process(input: Uint8Array, options?): Promise<Uint8Array> — pushes input through an internal chunkBytes stream and returns the concatenated frames as a single Uint8Array. Honours options.signal.
• toTransformStream<A, B>() — returns chunkBytes(frameSize, options).
• dispose() — clears hooks/listeners (calls super.dispose()).

Implementation Notes

The transform keeps a carry Uint8Array of bytes received but not yet consumed by a complete frame. On each upstream chunk it concatenates carry + chunk, emits as many frameSize-byte subarrays as fit, and copies the residual into a fresh allocation so the upstream chunk can be released. The flush callback resolves the residual against the configured tail policy.

Performance: full frames are emitted as subarray views (no copy). Only the residual carry is copied, bounded by frameSize - 1 bytes per upstream chunk, so steady-state allocations are O(frameSize) regardless of upstream chunk size. The 'pad' policy allocates exactly one frameSize buffer at end-of-stream.

Known limitations: the transform is byte-oriented; it does not understand record boundaries beyond fixed length. For length-prefixed or delimiter-separated framing, layer this on top of a higher-level decoder.