On this page

Pipelines

History
M

pipeTo

History
pipeTo(source, ...transforms?, writer, options?): Promise
Attributes
source:AsyncIterable | Iterable
The data source.
...transforms:Function | Object
Zero or more transforms to apply.
writer:Object
Destination with write(chunk) method.
options:Object
signal:AbortSignal
Abort the pipeline. Aborting fails the destination writer unless preventFail is true.
preventClose?:boolean
If true, do not call writer.end() when the source ends. Default: false.
preventFail?:boolean
If true, do not call writer.fail() on error. Default: false.
Returns:Promise
Fulfills with the total number of bytes written.

Pipe a source through transforms into a writer. If the writer has a writev(chunks) method, entire batches are passed in a single call (enabling scatter/gather I/O).

If the writer implements the optional *Sync methods (writeSync, writevSync, endSync), pipeTo() will attempt to use the synchronous methods first as a fast path, and fall back to the async versions only when the sync methods indicate they cannot complete (e.g., backpressure or waiting for the next tick). fail() is always called synchronously.

import { from, pipeTo } from 'node:stream/iter';
import { compressGzip } from 'node:zlib/iter';
import { open } from 'node:fs/promises';

const fh = await open('output.gz', 'w');
const totalBytes = await pipeTo(
  from('Hello, world!'),
  compressGzip(),
  fh.writer({ autoClose: true }),
);
JavaScript
const { from, pipeTo } = require('node:stream/iter');
const { compressGzip } = require('node:zlib/iter');
const { open } = require('node:fs/promises');

async function run() {
  const fh = await open('output.gz', 'w');
  const totalBytes = await pipeTo(
    from('Hello, world!'),
    compressGzip(),
    fh.writer({ autoClose: true }),
  );
}

run().catch(console.error);
JavaScript
M

pipeToSync

History
pipeToSync(source, ...transforms?, writer, options?): number
Attributes
source:Iterable
The sync data source.
...transforms:Function | Object
Zero or more sync transforms.
writer:Object
Destination with write(chunk) method.
options:Object
preventClose?:boolean
Default: false.
preventFail?:boolean
Default: false.
Returns:number
Total bytes written.

Synchronous version of pipeTo(). The source, all transforms, and the writer must be synchronous. Cannot accept async iterables or promises.

The writer must have the *Sync methods (writeSync, writevSync, endSync) and fail() for this to work.

M

pull

History
pull(source, ...transforms?, options?): AsyncIterable
Attributes
source:AsyncIterable | Iterable
The data source.
...transforms:Function | Object
Zero or more transforms to apply.
options:Object
signal:AbortSignal
Abort the pipeline.
Returns:AsyncIterable
whose chunks fulfill with Uint8Array[]

Create a lazy async pipeline. Source conversion and streamable protocol dispatch occur when pull() is called, but data is not read from source until the returned iterable is consumed. A signal that is already aborted is thrown synchronously after source conversion. Transforms are applied in order.

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

const asciiUpper = (chunks) => {
  if (chunks === null) return null;
  return chunks.map((c) => {
    for (let i = 0; i < c.length; i++) {
      c[i] -= (c[i] >= 97 && c[i] <= 122) * 32;
    }
    return c;
  });
};

const result = pull(from('hello'), asciiUpper);
console.log(await text(result)); // 'HELLO'
JavaScript
const { from, pull, text } = require('node:stream/iter');

const asciiUpper = (chunks) => {
  if (chunks === null) return null;
  return chunks.map((c) => {
    for (let i = 0; i < c.length; i++) {
      c[i] -= (c[i] >= 97 && c[i] <= 122) * 32;
    }
    return c;
  });
};

async function run() {
  const result = pull(from('hello'), asciiUpper);
  console.log(await text(result)); // 'HELLO'
}

run().catch(console.error);
JavaScript

Using an AbortSignal:

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

const ac = new AbortController();
const result = pull(source, transform, { signal: ac.signal });
ac.abort(); // Pipeline throws AbortError on next iteration
JavaScript
const { pull } = require('node:stream/iter');

const ac = new AbortController();
const result = pull(source, transform, { signal: ac.signal });
ac.abort(); // Pipeline throws AbortError on next iteration
JavaScript
M

pullSync

History
pullSync(source, ...transforms?): Iterable
Attributes
source:Iterable
The sync data source.
...transforms:Function | Object
Zero or more sync transforms.
Returns:Iterable
whose chunks return Uint8Array[]

Synchronous version of pull(). Source conversion and streamable protocol dispatch occur when pullSync() is called. All transforms must be synchronous.

PreviousSourcesNextPush streams