`node:stream`

Elide implements [Node.js streams][nodejs-stream] — five classes (Readable, Writable, Duplex, Transform, PassThrough) sharing the documented event/state model, plus the Promise-based composition utilities pipeline and finished. Each class extends EventEmitter through the prototype chain (so instanceof EventEmitter is true and the standard listener machinery works).

[nodejs-stream]: https://nodejs.org/api/stream.html

Module surface

declare class Readable extends EventEmitter {
  constructor(options?: {
    highWaterMark?: number;
    encoding?: string;
    objectMode?: boolean;
    emitClose?: boolean;
    autoDestroy?: boolean;
    read?(size: number): void;
  });

  readonly readable: boolean;
  readonly readableEnded: boolean;
  readonly readableFlowing: boolean | null;
  readonly readableHighWaterMark: number;
  readonly readableLength: number;
  readonly readableObjectMode: boolean;
  readonly readableEncoding: string | null;
  readonly destroyed: boolean;

  read(size?: number): unknown;
  setEncoding(encoding: string): this;
  pause(): void;
  resume(): this;
  isPaused(): boolean;
  unpipe(destination?: Writable): this;
  unshift(chunk: unknown, encoding?: string): void;
  push(chunk: unknown, encoding?: string): boolean;
  destroy(error?: unknown, callback?: (e: unknown) => void): this;
  pipe(destination: Writable, options?: object): Writable;

  toArray(options?: object): Promise<unknown[]>;
  forEach(fn: (chunk: unknown) => unknown, options?: object): Promise<void>;
  some(fn: (chunk: unknown) => unknown, options?: object): Promise<boolean>;
  every(fn: (chunk: unknown) => unknown, options?: object): Promise<boolean>;
  find(fn: (chunk: unknown) => unknown, options?: object): Promise<unknown>;
  reduce(fn: (acc: unknown, chunk: unknown) => unknown, initial?: unknown, options?: object): Promise<unknown>;
  map(fn: (chunk: unknown) => unknown, options?: object): Readable;
  filter(fn: (chunk: unknown) => unknown, options?: object): Readable;
  flatMap(fn: (chunk: unknown) => unknown, options?: object): Readable;
  drop(limit: number, options?: object): Readable;
  take(limit: number, options?: object): Readable;
  [Symbol.asyncIterator](): AsyncIterator<unknown>;
}

declare class Writable extends EventEmitter {
  constructor(options?: {
    highWaterMark?: number;
    decodeStrings?: boolean;
    defaultEncoding?: string;
    objectMode?: boolean;
    emitClose?: boolean;
    autoDestroy?: boolean;
    write?(chunk: unknown, encoding: string, callback: (e?: unknown) => void): void;
    final?(callback: (e?: unknown) => void): void;
  });

  readonly writable: boolean;
  readonly writableEnded: boolean;
  readonly writableFinished: boolean;
  readonly writableHighWaterMark: number;
  readonly writableLength: number;
  readonly writableObjectMode: boolean;
  readonly writableCorked: boolean;
  readonly destroyed: boolean;

  write(chunk: unknown, encoding?: string, callback?: (e?: unknown) => void): boolean;
  end(chunk?: unknown, encoding?: string, callback?: (e?: unknown) => void): void;
  cork(): void;
  uncork(): void;
  destroy(error?: unknown, callback?: (e: unknown) => void): this;
  setDefaultEncoding(encoding: string): this;
}

declare class Duplex extends Readable { /* writable side from Writable */ }
declare class Transform extends Duplex { /* uses _transform hook */ }
declare class PassThrough extends Transform {}

declare function pipeline(source: Readable, ...rest: (Writable | ((err?: unknown) => void))[]): Promise<void>;
declare function finished(stream: EventEmitter, callback?: (err?: unknown) => void): Promise<void>;
declare function compose(...streams: (Readable | Writable | Duplex | Transform)[]): Duplex;

State machine

Flowing vs paused. readableFlowing starts as null; the first 'data' listener (or pipe()/resume()) flips it to true, and pause() flips it to false. Adding a 'data' listener auto-resumes the stream and primes the pump (calls _read once via a microtask so the listener is in place by the time chunks arrive).
Buffering. push(chunk) enqueues into a FIFO; push(null) marks end-of-stream. In flowing mode, each push immediately emits 'data' and forwards to any piped destinations; in paused mode, callers pull chunks via read().
End-of-stream. Once push(null) lands and the buffer is empty, 'end' fires (once). With autoDestroy: true (default), the stream is then destroyed and 'close' fires too.
Backpressure. Writable.write returns false when the buffered length passes the high-water mark; the next time the buffer drains, 'drain' fires.

Subclass hooks

User code provides the actual data-source / data-sink logic by overriding the documented hooks on the prototype. Construction options accept the same hooks as a shorthand:

Option	Installs as	When invoked
`read(size)`	`_read`	Pull-mode: when `read()` is called and the buffer is empty. Auto-flow primes once on first `'data'` listener.
`write(chunk, encoding, cb)`	`_write`	Drain-mode: each queued chunk in turn. The callback signals write-acknowledgement.
`final(cb)`	`_final`	Right before `'finish'` fires, if defined; lets the stream perform async cleanup.
`transform(chunk, encoding, cb)`	`_transform`	Transform: each input chunk; call `cb(null, output)` to push.
`flush(cb)`	`_flush`	Transform: at end of input, before `'end'`, to push trailing data.

`pipeline(...streams, callback?)`

Chains streams via pipe and returns a Promise that resolves when the destination finishes (or rejects on the first 'error' from any participant). When a trailing callback is supplied it fires with (err?) on settle.

`finished(stream, callback?)`

Returns a Promise that resolves the next time the stream emits 'finish', 'end', or 'close', or rejects on 'error'. The optional callback fires with the same outcome.

Async iteration

Readable exposes the Node v20+ async iteration surface: Symbol.asyncIterator plus Promise-returning consumers (toArray, forEach, some, every, find, reduce) and stream-returning transformers (map, filter, flatMap, drop, take). User callbacks may return a Promise — the source is paused while the result is awaited (back-pressure under async fn) and resumed on fulfilment, so chunks are processed one at a time. Errors in the user fn — synchronous or via rejection — destroy the source and propagate as a Promise rejection (consumers) or 'error' event (transformers). stream.compose(...streams) chains streams via the existing pipe machinery and returns a Duplex-shaped wrapper: writes flow into the head, reads come from the tail. Errors on any participant destroy the wrapper with the captured reason.

Constructing and adapting streams

Readable.from(iterable[, options]) — build a Readable from a synchronous or asynchronous iterable.
Readable.wrap(oldStream) — bridge an old-style (data / end event) stream into a Readable; the data, end, and error events are wired through.
All five constructors (Readable, Writable, Duplex, Transform, PassThrough) are callable without new — Readable(opts) equals new Readable(opts), with instanceof and subclassing preserved.
Each constructor accepts a signal (AbortSignal) option; aborting it destroys the stream.
_readableState / _writableState expose live read/write state views for libraries that inspect them.

Sub-module specifiers

Specifier	Exports
`node:stream/promises`	Promise-based `pipeline` and `finished`.
`node:stream/iter` (Elide extension)	Async-iterator helpers: `text`, `bytes`, `collect`, `pull`, `from`, `fromSync`, `bytesSync`, `textSync`.

Implementation notes / gaps

Subclass _write calls and 'finish' emission are scheduled via the promise-job queue (matching Node's setImmediate-style scheduling), so callers can attach 'finish' / 'drain' / 'error' listeners after end() and still observe the event.

Quick example

import { pipeline, Readable, Transform, Writable } from "node:stream";

const source = new Readable({
  read() { this.push("a"); this.push("b"); this.push(null); },
});

const upper = new Transform({
  transform(chunk, _enc, cb) { cb(null, String(chunk).toUpperCase()); },
});

const collected = [];
const sink = new Writable({
  write(chunk, _enc, cb) { collected.push(String(chunk)); cb(); },
});

await pipeline(source, upper, sink);
console.log(collected.join(""));   // "AB"