On this page

Utilities

History
M

ondrain

History
ondrain(drainable): Promise | null
Attributes
drainable:Object
An object implementing the drainable protocol.
Returns:Promise | null

Wait for a drainable writer's backpressure to clear. Returns null if the object does not implement the drainable protocol, or a promise that fulfills with true when the writer can accept more data.

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
M

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.

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
M

tap

History
tap(callback): Function
Attributes
callback:Function
(chunks) => void Called with each batch.
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) => 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) => 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.

M

tapSync

History
tapSync(callback): Function
Attributes
callback:Function
Returns:Function

Synchronous version of tap().

PreviousConsumersNextMulti-consumer