suaveplan/packages/foundation/flow/streams/src/transforms/tap/tap.md

tap

(no submodules)

Purpose

The tap module exposes two complementary forms of the same side-effect probe: tapStream, the WHATWG TransformStream factory, and TapTransform, the plugin-OO wrapper used by BaseTransform-aware pipelines (StreamManager, TransformFactory, hook/event consumers). Both forms invoke a caller-supplied function for every chunk and forward the chunk downstream unchanged — the stream equivalent of a debug probe attached to logging, metrics, tracing, or test assertions.

Errors thrown by the side-effect are caught and re-wrapped as StreamTransformFailedError, erroring the stream and propagating upstream cancellation. For error-tolerant tapping (fire-and-forget) wrap the function in a try/catch before passing it in. The class form additionally honours options.signal before and after the side-effect via assertNotAborted, rejecting with StreamAbortedError when fired.

Features

• Identity passthrough — chunks are never modified
• Sync and async side-effect functions
• Errors in fn error the stream (StreamTransformFailedError)
• AbortSignal support (StreamAbortedError)
• Configurable highWaterMark
• OO form (TapTransform) plugs into BaseTransform-driven pipelines
• Streaming form (tapStream) is the single source of truth — the OO form delegates to it

Usage: Basic Example

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

const seen: number[] = [];
const result = await toArray(
    fromArray([1, 2, 3]).pipeThrough(tapStream((n) => seen.push(n)))
);
console.log(result); // [1, 2, 3]
console.log(seen); // [1, 2, 3]

import { TapTransform } from "@suaveplan/streams/transforms/tap/tap";

const log = new TapTransform<string>((s) => console.log("[tap]", s));
await log.process("hello"); // logs "[tap] hello", returns "hello"

Usage: Advanced Example

import { TapTransform, tapStream } from "@suaveplan/streams/transforms/tap/tap";
import { fromArray } from "@suaveplan/streams/readable/from-array/from-array";
import { mapStream } from "@suaveplan/streams/transforms/map/map";
import { toArray } from "@suaveplan/streams/sinks/to-array/to-array";
import { StreamManager } from "@suaveplan/streams/manager/manager/manager";

let chunkCount = 0;

const result = await toArray(
    fromArray([10, 20, 30])
        .pipeThrough(
            tapStream((n) => {
                chunkCount++;
                console.log("before:", n);
            })
        )
        .pipeThrough(mapStream((n) => n / 10))
        .pipeThrough(tapStream((n) => console.log("after:", n)))
);
// logs: before: 10, after: 1, before: 20, after: 2, before: 30, after: 3
console.log(chunkCount); // 3

let total = 0;
const meter = new TapTransform<number>(async (n) => {
    await Promise.resolve();
    total += n;
});
await new StreamManager().pipe(meter).execute(5); // process(5) returns 5, total === 5

API Reference

tapStream<T>(fn: (chunk: T) => void | Promise<void>, options?: TransformOptions): TransformStream<T, T>

Parameters:

• fn — side-effect function; return value is ignored. May be async.
• options.signal — AbortSignal; errors the stream on abort.
• options.highWaterMark — WHATWG queue depth (default: 1).

Returns: TransformStream<T, T>.

Throws (via error channel): StreamTransformFailedError when fn throws, StreamAbortedError when the signal fires.

class TapTransform<T = unknown>

constructor(fn: TapFn<T>)

fn is (chunk: T) => void | Promise<void>. Any value returned by the function is ignored — only the original chunk is forwarded.

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

Awaits fn(input) and returns input unchanged. Honours options.signal — rejects with StreamAbortedError (STREAM.ABORTED) when fired before or after the side-effect runs.

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

Returns a fresh WHATWG TransformStream produced by tapStream(fn). Errors thrown by fn are wrapped in StreamTransformFailedError (STREAM.TRANSFORM_FAILED) at the readable side.

type TapFn<T>

(chunk: T) => void | Promise<void> — sync or async side-effect callback.

Implementation Notes

fn is awaited before the chunk is enqueued, so an async tap delays throughput by the duration of the async operation. For fire-and-forget semantics (e.g. non-blocking analytics calls), pass a function that starts the async work but does not await it. The chunk passthrough (controller.enqueue(chunk)) executes only after fn resolves, ensuring ordering is preserved.

The OO class follows the canonical M2b BaseTransform-subclass pattern established by MapTransform:

1. The functional factory (tapStream) stays untouched — single source of truth for behaviour.
2. The OO subclass holds the configuration in a private readonly field and exposes name from the constructor.
3. process(input, options) runs the side-effect single-shot, calling assertNotAborted before and after the work, then returns the original input.
4. toTransformStream() returns the result of the functional factory directly — no parallel implementation, no drift surface.

Because tap is a stateless pass-through, there is no buffering, no batching, and no internal state to reset between invocations. Callers who need rate limiting on the side-effect should wrap fn themselves; this module deliberately stays unopinionated.