Table of Contents

• Stream
  • Organization of this Document
  • Types of Streams
    • Object Mode
    • Buffering
  • API for Stream Consumers
    • Writable Streams
      • Class: stream.Writable
        • Event: 'close'
        • Event: 'drain'
        • Event: 'error'
        • Event: 'finish'
        • Event: 'pipe'
        • Event: 'unpipe'
        • writable.cork()
        • writable.end([chunk][, encoding][, callback])
        • writable.setDefaultEncoding(encoding)
        • writable.uncork()
        • writable.write(chunk[, encoding][, callback])
    • Readable Streams
      • Two Modes
      • Three States
      • Choose One
      • Class: stream.Readable
        • Event: 'close'
        • Event: 'data'
        • Event: 'end'
        • Event: 'error'
        • Event: 'readable'
        • readable.isPaused()
        • readable.pause()
        • readable.pipe(destination[, options])
        • readable.read([size])
        • readable.resume()
        • readable.setEncoding(encoding)
        • readable.unpipe([destination])
        • readable.unshift(chunk)
        • readable.wrap(stream)
    • Duplex and Transform Streams
      • Class: stream.Duplex
      • Class: stream.Transform
  • API for Stream Implementers
    • Simplified Construction
    • Implementing a Writable Stream
      • Constructor: new stream.Writable([options])
      • writable._write(chunk, encoding, callback)
      • writable._writev(chunks, callback)
      • Errors While Writing
      • An Example Writable Stream
    • Implementing a Readable Stream
      • new stream.Readable([options])
      • readable._read(size)
      • readable.push(chunk[, encoding])
      • Errors While Reading
      • An Example Counting Stream
    • Implementing a Duplex Stream
      • new stream.Duplex(options)
      • An Example Duplex Stream
      • Object Mode Duplex Streams
    • Implementing a Transform Stream
      • new stream.Transform([options])
      • Events: 'finish' and 'end'
      • transform._flush(callback)
      • transform._transform(chunk, encoding, callback)
      • Class: stream.PassThrough
  • Additional Notes
    • Compatibility with Older Node.js Versions
    • readable.read(0)
    • readable.push('')

Stream#

Stability: 2 - Stable

A stream is an abstract interface for working with streaming data in Node.js. The stream module provides a base API that makes it easy to build objects that implement the stream interface.

There are many stream objects provided by Node.js. For instance, a request to an HTTP server and process.stdout are both stream instances.

Streams can be readable, writable, or both. All streams are instances of EventEmitter.

The stream module can be accessed using:

const stream = require('stream');

While it is important for all Node.js users to understand how streams work, the stream module itself is most useful for developers that are creating new types of stream instances. Developer's who are primarily consuming stream objects will rarely (if ever) have need to use the stream module directly.

Organization of this Document#

This document is divided into two primary sections with a third section for additional notes. The first section explains the elements of the stream API that are required to use streams within an application. The second section explains the elements of the API that are required to implement new types of streams.

Types of Streams#

There are four fundamental stream types within Node.js:

• Readable - streams from which data can be read (for example fs.createReadStream()).
• Writable - streams to which data can be written (for example fs.createWriteStream()).
• Duplex - streams that are both Readable and Writable (for example net.Socket).
• Transform - Duplex streams that can modify or transform the data as it is written and read (for example zlib.createDeflate()).

Object Mode#

All streams created by Node.js APIs operate exclusively on strings and Buffer objects. It is possible, however, for stream implementations to work with other types of JavaScript values (with the exception of null, which serves a special purpose within streams). Such streams are considered to operate in "object mode".

Stream instances are switched into object mode using the objectMode option when the stream is created. Attempting to switch an existing stream into object mode is not safe.

Buffering#

Both Writable and Readable streams will store data in an internal buffer that can be retrieved using writable._writableState.getBuffer() or readable._readableState.buffer, respectively.

The amount of data potentially buffered depends on the highWaterMark option passed into the streams constructor. For normal streams, the highWaterMark option specifies a total number of bytes. For streams operating in object mode, the highWaterMark specifies a total number of objects.

Data is buffered in Readable streams when the implementation calls stream.push(chunk). If the consumer of the Stream does not call stream.read(), the data will sit in the internal queue until it is consumed.

Once the total size of the internal read buffer reaches the threshold specified by highWaterMark, the stream will temporarily stop reading data from the underlying resource until the data currently buffered can be consumed (that is, the stream will stop calling the internal readable._read() method that is used to fill the read buffer).

Data is buffered in Writable streams when the writable.write(chunk) method is called repeatedly. While the total size of the internal write buffer is below the threshold set by highWaterMark, calls to writable.write() will return true. Once the size of the internal buffer reaches or exceeds the highWaterMark, false will be returned.

A key goal of the stream API, particularly the stream.pipe() method, is to limit the buffering of data to acceptable levels such that sources and destinations of differing speeds will not overwhelm the available memory.

Because Duplex and Transform streams are both Readable and Writable, each maintain two separate internal buffers used for reading and writing, allowing each side to operate independently of the other while maintaining an appropriate and efficient flow of data. For example, net.Socket instances are Duplex streams whose Readable side allows consumption of data received from the socket and whose Writable side allows writing data to the socket. Because data may be written to the socket at a faster or slower rate than data is received, it is important for each side to operate (and buffer) independently of the other.

API for Stream Consumers#

Almost all Node.js applications, no matter how simple, use streams in some manner. The following is an example of using streams in a Node.js application that implements an HTTP server:

const http = require('http');

const server = http.createServer( (req, res) => {
  // req is an http.IncomingMessage, which is a Readable Stream
  // res is an http.ServerResponse, which is a Writable Stream

  let body = '';
  // Get the data as utf8 strings.
  // If an encoding is not set, Buffer objects will be received.
  req.setEncoding('utf8');

  // Readable streams emit 'data' events once a listener is added
  req.on('data', (chunk) => {
    body += chunk;
  });

  // the end event indicates that the entire body has been received
  req.on('end', () => {
    try {
      const data = JSON.parse(body);
      // write back something interesting to the user:
      res.write(typeof data);
      res.end();
    } catch (er) {
      // uh oh!  bad json!
      res.statusCode = 400;
      return res.end(`error: ${er.message}`);
    }
  });
});

server.listen(1337);

// $ curl localhost:1337 -d '{}'
// object
// $ curl localhost:1337 -d '"foo"'
// string
// $ curl localhost:1337 -d 'not json'
// error: Unexpected token o

Writable streams (such as res in the example) expose methods such as write() and end() that are used to write data onto the stream.

Readable streams use the EventEmitter API for notifying application code when data is available to be read off the stream. That available data can be read from the stream in multiple ways.

Both Writable and Readable streams use the EventEmitter API in various ways to communicate the current state of the stream.

Duplex and Transform streams are both Writable and Readable.

Applications that are either writing data to or consuming data from a stream are not required to implement the stream interfaces directly and will generally have no reason to call require('stream').

Developers wishing to implement new types of streams should refer to the section API for Stream Implementers.

Writable Streams#

Writable streams are an abstraction for a destination to which data is written.

Examples of Writable streams include:

• HTTP requests, on the client
• HTTP responses, on the server
• fs write streams
• zlib streams
• crypto streams
• TCP sockets
• child process stdin
• process.stdout, process.stderr

Note: Some of these examples are actually Duplex streams that implement the Writable interface.

All Writable streams implement the interface defined by the stream.Writable class.

While specific instances of Writable streams may differ in various ways, all Writable streams follow the same fundamental usage pattern as illustrated in the example below:

const myStream = getWritableStreamSomehow();
myStream.write('some data');
myStream.write('some more data');
myStream.end('done writing data');

Class: stream.Writable#

Event: 'close'#

The 'close' event is emitted when the stream and any of its underlying resources (a file descriptor, for example) have been closed. The event indicates that no more events will be emitted, and no further computation will occur.

Not all Writable streams will emit the 'close' event.

Event: 'drain'#

If a call to stream.write(chunk) returns false, the 'drain' event will be emitted when it is appropriate to resume writing data to the stream.

// Write the data to the supplied writable stream one million times.
// Be attentive to back-pressure.
function writeOneMillionTimes(writer, data, encoding, callback) {
  let i = 1000000;
  write();
  function write() {
    var ok = true;
    do {
      i--;
      if (i === 0) {
        // last time!
        writer.write(data, encoding, callback);
      } else {
        // see if we should continue, or wait
        // don't pass the callback, because we're not done yet.
        ok = writer.write(data, encoding);
      }
    } while (i > 0 && ok);
    if (i > 0) {
      // had to stop early!
      // write some more once it drains
      writer.once('drain', write);
    }
  }
}

Event: 'error'#

• <Error>

The 'error' event is emitted if an error occurred while writing or piping data. The listener callback is passed a single Error argument when called.

Note: The stream is not closed when the 'error' event is emitted.

Event: 'finish'#

The 'finish' event is emitted after the stream.end() method has been called, and all data has been flushed to the underlying system.

const writer = getWritableStreamSomehow();
for (var i = 0; i < 100; i ++) {
  writer.write('hello, #${i}!\n');
}
writer.end('This is the end\n');
writer.on('finish', () => {
  console.error('All writes are now complete.');
});

Event: 'pipe'#

• src <stream.Readable> source stream that is piping to this writable

The 'pipe' event is emitted when the stream.pipe() method is called on a readable stream, adding this writable to its set of destinations.

const writer = getWritableStreamSomehow();
const reader = getReadableStreamSomehow();
writer.on('pipe', (src) => {
  console.error('something is piping into the writer');
  assert.equal(src, reader);
});
reader.pipe(writer);

Event: 'unpipe'#

• src <Readable Stream> The source stream that unpiped this writable

The 'unpipe' event is emitted when the stream.unpipe() method is called on a Readable stream, removing this Writable from its set of destinations.

const writer = getWritableStreamSomehow();
const reader = getReadableStreamSomehow();
writer.on('unpipe', (src) => {
  console.error('Something has stopped piping into the writer.');
  assert.equal(src, reader);
});
reader.pipe(writer);
reader.unpipe(writer);

writable.cork()#

The writable.cork() method forces all written data to be buffered in memory. The buffered data will be flushed when either the stream.uncork() or stream.end() methods are called.

The primary intent of writable.cork() is to avoid a situation where writing many small chunks of data to a stream do not cause a backup in the internal buffer that would have an adverse impact on performance. In such situations, implementations that implement the writable._writev() method can perform buffered writes in a more optimized manner.

writable.end([chunk][, encoding][, callback])#

• chunk <String> | <Buffer> | <any> Optional data to write. For streams not operating in object mode, chunk must be a string or a Buffer. For object mode streams, chunk may be any JavaScript value other than null.
• encoding <String> The encoding, if chunk is a String
• callback <Function> Optional callback for when the stream is finished

Calling the writable.end() method signals that no more data will be written to the Writable. The optional chunk and encoding arguments allow one final additional chunk of data to be written immediately before closing the stream. If provided, the optional callback function is attached as a listener for the 'finish' event.

Calling the stream.write() method after calling stream.end() will raise an error.

// write 'hello, ' and then end with 'world!'
const file = fs.createWriteStream('example.txt');
file.write('hello, ');
file.end('world!');
// writing more now is not allowed!

writable.setDefaultEncoding(encoding)#

• encoding <String> The new default encoding
• Returns: this

The writable.setDefaultEncoding() method sets the default encoding for a Writable stream.

writable.uncork()#

The writable.uncork() method flushes all data buffered since stream.cork() was called.

When using writable.cork() and writable.uncork() to manage the buffering of writes to a stream, it is recommended that calls to writable.uncork() be deferred using process.nextTick(). Doing so allows batching of all writable.write() calls that occur within a given Node.js event loop phase.

stream.cork();
stream.write('some ');
stream.write('data ');
process.nextTick(() => stream.uncork());

If the writable.cork() method is called multiple times on a stream, the same number of calls to writable.uncork() must be called to flush the buffered data.

stream.cork();
stream.write('some ');
stream.cork();
stream.write('data ');
process.nextTick(() => {
  stream.uncork();
  // The data will not be flushed until uncork() is called a second time.
  stream.uncork();
});

writable.write(chunk[, encoding][, callback])#

• chunk <String> | <Buffer> The data to write
• encoding <String> The encoding, if chunk is a String
• callback <Function> Callback for when this chunk of data is flushed
• Returns: <Boolean> false if the stream wishes for the calling code to wait for the 'drain' event to be emitted before continuing to write additional data; otherwise true.

The writable.write() method writes some data to the stream, and calls the supplied callback once the data has been fully handled. If an error occurs, the callback may or may not be called with the error as its first argument. To reliably detect write errors, add a listener for the 'error' event.

The return value indicates whether the written chunk was buffered internally and the buffer has exceeded the highWaterMark configured when the stream was created. If false is returned, further attempts to write data to the stream should be paused until the 'drain' event is emitted.

A Writable stream in object mode will always ignore the encoding argument.

Readable Streams#

Readable streams are an abstraction for a source from which data is consumed.

Examples of Readable streams include:

• HTTP responses, on the client
• HTTP requests, on the server
• fs read streams
• zlib streams
• crypto streams
• TCP sockets
• child process stdout and stderr
• process.stdin

All Readable streams implement the interface defined by the stream.Readable class.

Two Modes#

Readable streams effectively operate in one of two modes: flowing and paused.

When in flowing mode, data is read from the underlying system automatically and provided to an application as quickly as possible using events via the EventEmitter interface.

In paused mode, the stream.read() method must be called explicitly to read chunks of data from the stream.

All Readable streams begin in paused mode but can be switched to flowing mode in one of the following ways:

• Adding a 'data' event handler.
• Calling the stream.resume() method.
• Calling the stream.pipe() method to send the data to a Writable.

The Readable can switch back to paused mode using one of the following:

• If there are no pipe destinations, by calling the stream.pause() method.
• If there are pipe destinations, by removing any 'data' event handlers, and removing all pipe destinations by calling the stream.unpipe() method.

The important concept to remember is that a Readable will not generate data until a mechanism for either consuming or ignoring that data is provided. If the consuming mechanism is disabled or taken away, the Readable will attempt to stop generating the data.

Note: For backwards compatibility reasons, removing 'data' event handlers will not automatically pause the stream. Also, if there are piped destinations, then calling stream.pause() will not guarantee that the stream will remain paused once those destinations drain and ask for more data.

Note: If a Readable is switched into flowing mode and there are no consumers available handle the data, that data will be lost. This can occur, for instance, when the readable.resume() method is called without a listener attached to the 'data' event, or when a 'data' event handler is removed from the stream.

Three States#

The "two modes" of operation for a Readable stream are a simplified abstraction for the more complicated internal state management that is happening within the Readable stream implementation.

Specifically, at any given point in time, every Readable is in one of three possible states:

• readable._readableState.flowing = null
• readable._readableState.flowing = false
• readable._readableState.flowing = true

When readable._readableState.flowing is null, no mechanism for consuming the streams data is provided so the stream will not generate its data.

Attaching a listener for the 'data' event, calling the readable.pipe() method, or calling the readable.resume() method will switch readable._readableState.flowing to true, causing the Readable to begin actively emitting events as data is generated.

Calling readable.pause(), readable.unpipe(), or receiving "back pressure" will cause the readable._readableState.flowing to be set as false, temporarily halting the flowing of events but not halting the generation of data.

While readable._readableState.flowing is false, data may be accumulating within the streams internal buffer.

Choose One#

The Readable stream API evolved across multiple Node.js versions and provides multiple methods of consuming stream data. In general, developers should choose one of the methods of consuming data and should never use multiple methods to consume data from a single stream.

Use of the readable.pipe() method is recommended for most users as it has been implemented to provide the easiest way of consuming stream data. Developers that require more fine-grained control over the transfer and generation of data can use the EventEmitter and readable.pause()/readable.resume() APIs.

Class: stream.Readable#

Event: 'close'