Classic stream interop

History

These utility functions bridge between classic stream.Readable/stream.Writable streams and the stream/iter API.

Both fromReadable() and fromWritable() accept duck-typed objects -- they do not require the input to extend stream.Readable or stream.Writable directly. The minimum contract is described below for each function.

fromReadable

History

fromReadable(readable): AsyncIterable

Stability: 1Experimental

Attributes

readable:stream.Readable | Object

A classic Readable stream or any object with read(), on(), and off() methods.

Returns:AsyncIterable

whose chunks fulfill with Uint8Array[]

Converts a classic Readable stream (or duck-typed equivalent) into a stream/iter async iterable source that can be passed to from(), pull(), text(), etc.

If the object implements the toAsyncStreamable protocol (as stream.Readable does), that protocol is used. Otherwise, the function duck-types on read(), on(), and off() (EventEmitter) and wraps the stream with a batched async iterator.

The result is cached per instance -- calling fromReadable() twice with the same stream returns the same iterable.

For object-mode or encoded Readable streams, chunks are automatically normalized to Uint8Array.

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

const readable = new Readable({
  read() { this.push('hello world'); this.push(null); },
});

const result = await text(fromReadable(readable));
console.log(result); // 'hello world'

JavaScript

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

const readable = new Readable({
  read() { this.push('hello world'); this.push(null); },
});

async function run() {
  const result = await text(fromReadable(readable));
  console.log(result); // 'hello world'
}
run();

JavaScript

fromWritable

History

fromWritable(writable, options?): Object

Stability: 1Experimental

Attributes

writable:stream.Writable | Object

A classic Writable stream or any object with write() and on() methods.

options:Object

backpressure?:string

Backpressure policy. Default: 'strict'.

'strict':

writes are rejected when the buffer is full. Catches callers that ignore backpressure.

'unbounded':

writes wait for drain when the buffer is full. Recommended for use with pipeTo().

'drop-newest':

writes are silently discarded when the buffer is full.

'drop-oldest':

not supported. Throws ERR_INVALID_ARG_VALUE.

Returns:Object

A stream/iter Writer adapter.

Creates a stream/iter Writer adapter from a classic Writable stream (or duck-typed equivalent). The adapter can be passed to pipeTo() as a destination.

Since all writes on a classic Writable are fundamentally asynchronous, the synchronous Writer methods (writeSync, writevSync, endSync) always return false or -1, deferring to the async path. The per-write options.signal parameter from the Writer interface is also ignored.

The result is cached per instance and backpressure policy -- calling fromWritable() twice with the same stream and backpressure option returns the same Writer.

For duck-typed streams that do not expose writableHighWaterMark, writableLength, or similar properties, sensible defaults are used. Object-mode writables (if detectable) are rejected since the Writer interface is bytes-only.

import { Writable } from 'node:stream';
import { from, fromWritable, pipeTo } from 'node:stream/iter';

const writable = new Writable({
  write(chunk, encoding, cb) { console.log(chunk.toString()); cb(); },
});

await pipeTo(from('hello world'),
             fromWritable(writable, { backpressure: 'unbounded' }));

JavaScript

const { Writable } = require('node:stream');
const { from, fromWritable, pipeTo } = require('node:stream/iter');

async function run() {
  const writable = new Writable({
    write(chunk, encoding, cb) { console.log(chunk.toString()); cb(); },
  });

  await pipeTo(from('hello world'),
               fromWritable(writable, { backpressure: 'unbounded' }));
}
run();

JavaScript

toReadable

History

toReadable(source, options?): stream.Readable

Stability: 1Experimental

Attributes

source:AsyncIterable

whose chunks must fulfill with Uint8Array[] the return value of pull() or from().

options:Object

highWaterMark?:number

The internal buffer size in bytes before backpressure is applied. Default: 65536 (64 KB).

signal:AbortSignal

An optional signal to abort the readable.

Returns:stream.Readable

Creates a byte-mode stream.Readable from the source (the native batch format used by the stream/iter API). Each Uint8Array in a yielded batch is pushed as a separate chunk into the Readable.

import { createWriteStream } from 'node:fs';
import { from, pull, toReadable } from 'node:stream/iter';
import { compressGzip } from 'node:zlib/iter';

const source = pull(from('hello world'), compressGzip());
const readable = toReadable(source);

readable.pipe(createWriteStream('output.gz'));

JavaScript

const { createWriteStream } = require('node:fs');
const { from, pull, toReadable } = require('node:stream/iter');
const { compressGzip } = require('node:zlib/iter');

const source = pull(from('hello world'), compressGzip());
const readable = toReadable(source);

readable.pipe(createWriteStream('output.gz'));

JavaScript

toReadableSync

History

toReadableSync(source, options?): stream.Readable

Stability: 1Experimental

Attributes

source:Iterable

whose chunks must return Uint8Array[], such as the return value of pullSync() or fromSync().

options:Object

highWaterMark?:number

The internal buffer size in bytes before backpressure is applied. Default: 65536 (64 KB).

Returns:stream.Readable

Creates a byte-mode stream.Readable from the source. The _read() method pulls from the iterator synchronously, so data is available immediately via readable.read().

import { fromSync, toReadableSync } from 'node:stream/iter';

const source = fromSync('hello world');
const readable = toReadableSync(source);

console.log(readable.read().toString()); // 'hello world'

JavaScript

const { fromSync, toReadableSync } = require('node:stream/iter');

const source = fromSync('hello world');
const readable = toReadableSync(source);

console.log(readable.read().toString()); // 'hello world'

JavaScript

toWritable

History

toWritable(writer): stream.Writable

Stability: 1Experimental

Attributes

writer:Object

A stream/iter Writer. Only the write() method is required; end(), fail(), writeSync(), writevSync(), endSync(), and writev() are optional.

Returns:stream.Writable

Creates a classic stream.Writable backed by a stream/iter Writer.

Each _write() / _writev() call attempts the Writer's synchronous method first (writeSync / writevSync), falling back to the async method if the sync path returns false. Similarly, _final() tries endSync() before end(). When the sync path succeeds, the callback is deferred via queueMicrotask to preserve the async resolution contract.

The Writable's highWaterMark is set to Number.MAX_SAFE_INTEGER to effectively disable its internal buffering, allowing the underlying Writer to manage backpressure directly.

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

const { writer, readable } = push();
const writable = toWritable(writer);

writable.write('hello');
writable.end();

JavaScript

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

const { writer, readable } = push();
const writable = toWritable(writer);

writable.write('hello');
writable.end();

JavaScript