Utilities

History

ondrain

History

ondrain(drainable): Promise | null

Attributes

drainable:Object

An object implementing the drainable protocol.

Returns:Promise | null

Wait for a drainable writer to regain physical buffer capacity. Returns null if the object does not implement the drainable protocol, or a promise that fulfills with true when buffered data falls below the byte budget.

For writers using 'drop-oldest' or 'drop-newest', this waits for physical capacity even though writes do not block. This allows producers to avoid data loss by waiting before writing.

import { push, ondrain, text } from 'node:stream/iter';

const { writer, readable } = push({ budget: 16384 });
const chunk = new Uint8Array(8192);  // 8 KB
writer.writeSync(chunk);
writer.writeSync(chunk);  // 16 KB total -- buffer full

// Start consuming so the buffer can actually drain
const consuming = text(readable);

// Buffer is full -- wait for drain
const canWrite = await ondrain(writer);
if (canWrite) {
  await writer.write('c');
}
await writer.end();
await consuming;

JavaScript

const { push, ondrain, text } = require('node:stream/iter');

async function run() {
  const { writer, readable } = push({ budget: 16384 });
  const chunk = new Uint8Array(8192);  // 8 KB
  writer.writeSync(chunk);
  writer.writeSync(chunk);  // 16 KB total -- buffer full

  // Start consuming so the buffer can actually drain
  const consuming = text(readable);

  // Buffer is full -- wait for drain
  const canWrite = await ondrain(writer);
  if (canWrite) {
    await writer.write('c');
  }
  await writer.end();
  await consuming;
}

run().catch(console.error);

JavaScript

merge

History

merge(...sources, options?): AsyncIterable

Attributes

...sources:AsyncIterable | Iterable

whose chunks must be Uint8Array[]

options:Object

signal:AbortSignal

Returns:AsyncIterable

whose chunks fulfill with Uint8Array[]

Merge multiple async iterables by yielding batches in temporal order (whichever source produces data first). All sources are consumed concurrently, with at most one pending next() call per source.

If a source fails, the returned iterable rejects with its error. merge() calls return() on the other sources but does not wait for it to settle: an async generator source that is suspended in an await only runs its cleanup once that await completes.

import { from, merge, text } from 'node:stream/iter';

const merged = merge(from('hello '), from('world'));
console.log(await text(merged)); // Order depends on timing

JavaScript

const { from, merge, text } = require('node:stream/iter');

async function run() {
  const merged = merge(from('hello '), from('world'));
  console.log(await text(merged)); // Order depends on timing
}

run().catch(console.error);

JavaScript

tap

History

tap(callback): Function

Attributes

callback:Function

(chunks, options) => void Called with each batch and with null when the source ends. options.signal is the pipeline's AbortSignal.

Returns:Function

A stateless transform.

Create a pass-through transform that observes batches without modifying them. Useful for logging, metrics, or debugging.

import { from, pull, text, tap } from 'node:stream/iter';

const result = pull(
  from('hello'),
  tap((chunks) => {
    if (chunks !== null) console.log('Batch size:', chunks.length);
  }),
);
console.log(await text(result));

JavaScript

const { from, pull, text, tap } = require('node:stream/iter');

async function run() {
  const result = pull(
    from('hello'),
    tap((chunks) => {
      if (chunks !== null) console.log('Batch size:', chunks.length);
    }),
  );
  console.log(await text(result));
}

run().catch(console.error);

JavaScript

tap() intentionally does not prevent in-place modification of the chunks by the tapping callback; but return values are ignored.

tapSync

History

tapSync(callback): Function

Attributes

callback:Function

Returns:Function

Synchronous version of tap().