Streams

Owners: @nodejs/streams.

Purpose

Implement the Node-flavored stream contract — Readable, Writable, Duplex, Transform, PassThrough — plus the WHATWG WebStreams (ReadableStream, WritableStream, TransformStream) and their interop. Streams are the primary backpressure-aware I/O primitive used by fs, net, tls, http, http2, zlib, and the worker MessagePort.

Directory layout

lib/
  stream.js                       // public Stream module
  stream/                         // promise-based helpers
    consumers.js                  // stream.consumers.{text,json,buffer,arrayBuffer,blob}
    promises.js                   // pipeline / finished as promises
    web.js                        // WebStreams re-exports
  internal/streams/
    readable.js                   // Readable
    writable.js                   // Writable
    duplex.js                     // Duplex (composes R+W)
    transform.js                  // Transform (R+W with _transform)
    passthrough.js                // PassThrough
    duplexpair.js                 // Duplex pair (testing helper)
    duplexify.js                  // wrap thing-with-write-and-end into Duplex
    pipeline.js                   // stream.pipeline + finished
    end-of-stream.js              // finished()
    destroy.js                    // shared destroy() machinery
    compose.js                    // stream.compose (in/out chain)
    operators.js                  // map / filter / reduce / take / drop / toArray
    add-abort-signal.js           // AbortSignal hookup
    legacy.js                     // pre-streams2 compat
    state.js                      // shared bitfield state
    utils.js                      // safe pipe helpers
    from.js                       // Readable.from / fromWeb
    iter/                         // async iterator integration
    fast-utf8-stream.js           // optimized utf8 decode stream used by HTTP
    lazy_transform.js
  internal/webstreams/            // WHATWG WebStreams implementation
src/
  stream_base.{cc,h}              // C++ StreamBase
  stream_base-inl.h
  stream_pipe.{cc,h}              // zero-copy uv_link/stream_pipe between StreamBase
  stream_wrap.{cc,h}              // libuv handle ⇄ StreamBase glue
  js_stream.{cc,h}                // bridge JS-implemented streams to StreamBase

Key abstractions

Type / file	Role
Readable (lib/internal/streams/readable.js)	Pull-based readable side: buffer, push/pull, paused/flowing modes, async iter.
Writable (lib/internal/streams/writable.js)	Push-based writable side: write/cork/uncork, finish, drain.
Duplex (lib/internal/streams/duplex.js)	Combines R + W on the same instance.
Transform (lib/internal/streams/transform.js)	Duplex with _transform/_flush; backpressure-coupled.
pipeline (lib/internal/streams/pipeline.js)	Chain streams with proper teardown on error.
eos (lib/internal/streams/end-of-stream.js)	One-shot completion observer used internally and exposed as stream.finished.
destroy() (lib/internal/streams/destroy.js)	Shared destroy/finalize state-machine.
StreamBase (src/stream_base.h)	C++ side of the contract: DoWrite, ReadStart/Stop, OnStreamRead.
StreamPipe (src/stream_pipe.cc)	Server-side splice between two StreamBase instances; backs HTTP→TLS for example.
JSStream (src/js_stream.cc)	Lets a JS-implemented stream act as a StreamBase for native consumers.

How it works (Readable/Writable lifecycle)

stateDiagram-v2
    [*] --> Constructed
    Constructed --> Reading: read() / data listener / for-await
    Reading --> Buffered: push() data
    Buffered --> Reading: consumer reads
    Reading --> Ended: push(null)
    Ended --> Closed: destroy()/close
    Reading --> Errored: destroy(err)
    Errored --> Closed
    Closed --> [*]

Key invariants enforced by lib/internal/streams/state.js and the destroy.js finalizer:

• destroy() is idempotent and propagates through pipeline/compose.
• close always fires after error and finish/end.
• Backpressure: writable.write() returns false when the internal buffer crosses highWaterMark; consumers must wait for 'drain'.
• Async iterators and for await are first-class: Readable[Symbol.asyncIterator] is the canonical consumption API in modern code.

pipeline and finished

stream.pipeline(a, b, c, cb) is the safe way to chain streams. It guarantees:

• Errors propagate to the callback exactly once.
• All streams are destroyed on error.
• The promise variant (require('stream/promises').pipeline) returns a Promise.

The implementation in lib/internal/streams/pipeline.js accepts:

• Streams (Readable/Writable/Duplex).
• Async iterables (treated as Readable.from).
• Functions (async function* gen(source)) — turned into Transform streams.
• Web streams (handed to/from the WHATWG version via lib/internal/streams/from.js).

compose and operators

stream.compose(a, b, c) creates a new Duplex whose read side is c.readable and whose write side is a.writable. The lib/internal/streams/operators.js file provides functional helpers (map, filter, take, drop, flatMap, toArray, reduce, forEach) on async iterables.

WebStreams interop

lib/internal/webstreams/ is a self-contained WHATWG WebStreams implementation. Conversion happens in two places:

• Readable.fromWeb(rs) / Readable.toWeb(readable) — lib/internal/streams/readable.js.
• Writable.fromWeb / Writable.toWeb — lib/internal/streams/writable.js.
• Duplex.fromWeb / Duplex.toWeb.

The conversions handle queue size, byte-stream vs. default streams, abort signal pairing, and lock semantics.

C++ StreamBase pipeline

graph LR
    JS["socket.write(data)"] --> WriteWrap["WriteWrap (ReqWrap)"]
    WriteWrap --> SB["StreamBase::Writev"]
    SB --> uvw["uv_write"]
    uvw --> uv[libuv]
    uv --> done["uv_write callback"]
    done --> SB2["StreamBase::OnAfterWrite"]
    SB2 --> JS2["MakeCallback -> JS 'drain'"]

A peer optimisation: when source and sink are both StreamBases and have compatible types, src/stream_pipe.cc splices them so data does not have to land in JS. HTTP req.pipe(res) over TCP can take this path.

Integration points

• fs: fs.createReadStream/fs.createWriteStream produce ReadStream / WriteStream over libuv uv_fs_* requests. See systems › file-system.
• net/tls: net.Socket is a Duplex over TCPWrap/PipeWrap; tls.TLSSocket wraps a Socket with an additional TLSWrap.
• http/http2: client/server interfaces are Duplex streams; the HTTP/2 implementation hosts its own Http2Stream Duplex.
• zlib: each compressor/decompressor is a Transform over libuv worker threads.
• worker_threads.MessagePort: implements a Duplex-like interface for object streams when using worker.postMessage.

Entry points for modification

• Bug in backpressure / 'drain' scheduling? Look at lib/internal/streams/writable.js clearBuffer and onfinish.
• Bug in iteration? Readable[Symbol.asyncIterator] and lib/internal/streams/iter/ are the relevant places.
• Bug in pipeline? lib/internal/streams/pipeline.js. The matching tests are test/parallel/test-stream-pipeline-*.js and test/parallel/test-stream-*.js.
• C++ side of stream wiring? src/stream_base.cc and the corresponding *_wrap.cc (e.g., tcp_wrap.cc).