Node.js Streams

Node.js streams process data over time instead of loading everything into memory first.

Files, HTTP messages, compression, standard input, and child-process output all use streams.

Streams are useful for two reasons:

• processing can start before all the data arrives
• memory use can stay bounded for large data

The important detail is backpressure. When a destination is slower than a source, streams signal the source to slow down.

The four stream types

Node.js has four main stream types:

• Readable: produces data
• Writable: consumes data
• Duplex: both readable and writable
• Transform: a duplex stream whose output is derived from its input

fs.createReadStream() returns a readable stream. fs.createWriteStream() returns a writable stream.

A TCP socket is duplex. A gzip compressor is a transform stream.

Copy a large file with pipeline

The simplest safe way to connect streams is pipeline():

const { createReadStream, createWriteStream } = require('node:fs')
const { pipeline } = require('node:stream/promises')

async function copyFile() {
  await pipeline(
    createReadStream('video.mp4'),
    createWriteStream('video-copy.mp4'),
  )
}

copyFile().catch(console.error)

pipeline() forwards data, handles backpressure, propagates errors, and cleans up the connected streams.

It resolves when the pipeline finishes. It rejects when a stream fails.

The older source.pipe(destination) API is still useful, but errors are not automatically forwarded through a complete chain. Prefer pipeline() when you need completion and error handling.

Compress data through a transform

Add a transform stream between the source and destination:

const { createReadStream, createWriteStream } = require('node:fs')
const { pipeline } = require('node:stream/promises')
const { createGzip } = require('node:zlib')

async function compressFile() {
  await pipeline(
    createReadStream('archive.tar'),
    createGzip(),
    createWriteStream('archive.tar.gz'),
  )
}

compressFile().catch(console.error)

Each stream handles one job. The pipeline connects them.

Stream a file over HTTP

An HTTP response is a writable stream:

const { createReadStream } = require('node:fs')
const { createServer } = require('node:http')

const server = createServer((request, response) => {
  const file = createReadStream('guide.pdf')

  file.once('open', () => {
    response.writeHead(200, {
      'content-type': 'application/pdf',
    })

    file.pipe(response)
  })

  file.on('error', (error) => {
    if (!response.headersSent) {
      response.writeHead(404, {
        'content-type': 'text/plain',
      })
      response.end('File not found')
    } else {
      response.destroy(error)
    }

    console.error(error)
  })
})

server.listen(3000)

The response can begin as soon as the first file chunk is available.

The example waits for the file to open before sending a 200 response. Once response bytes have been sent, you cannot replace them with a clean error response.

Be careful when using pipeline() directly with an HTTP response. On an error it can destroy the socket before your code sends an error body.

Consume a readable stream

Readable streams are asynchronous iterables.

Use for await...of to process one chunk at a time:

const { createReadStream } = require('node:fs')

async function countBytes() {
  const stream = createReadStream('guide.pdf')
  let total = 0

  for await (const chunk of stream) {
    total += chunk.length
  }

  return total
}

countBytes().then(console.log)

The loop waits for the next chunk and rejects if the stream emits an error.

Do not mix consumption styles on one readable stream unless you understand its flowing state. Calling pipe(), listening for data, using read(), and async iteration are different ways to consume it.

Create a readable stream

Use Readable.from() for an iterable or async iterable:

const { Readable } = require('node:stream')

const stream = Readable.from([
  'Hello ',
  'from ',
  'a stream\n',
])

stream.pipe(process.stdout)

For a custom source, implement the read() method:

const { Readable } = require('node:stream')

let current = 1

const numbers = new Readable({
  read() {
    if (current > 3) {
      this.push(null)
      return
    }

    this.push(String(current))
    current += 1
  },
})

push(null) signals the end of the readable stream.

Do not call push() forever while ignoring its return value. Custom high-volume sources must respect the stream’s demand and backpressure rules.

Create a writable stream

Implement write() in the constructor options:

const { Writable } = require('node:stream')

const logger = new Writable({
  write(chunk, encoding, callback) {
    console.log(chunk.toString())
    callback()
  },
})

logger.write('First message')
logger.end('Last message')

Call the callback when the chunk has been processed. Pass an error to the callback when writing fails.

end() can include one final chunk. It tells the stream that no more data will be written.

The writable emits finish after all data has been flushed. close means the underlying resource has closed. Those events do not mean the same thing.

Respect writable backpressure

writable.write() returns a boolean.

When it returns false, wait for drain before writing more:

const { once } = require('node:events')

async function writeChunks(writable, chunks) {
  for (const chunk of chunks) {
    if (!writable.write(chunk)) {
      await once(writable, 'drain')
    }
  }

  writable.end()
  await once(writable, 'finish')
}

Ignoring false lets data accumulate in memory and can eventually exhaust it.

pipe() and pipeline() manage this flow for you.

Create a transform stream

A transform receives chunks and produces new chunks:

const { Transform } = require('node:stream')

const uppercase = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, chunk.toString().toUpperCase())
  },
})

process.stdin
  .pipe(uppercase)
  .pipe(process.stdout)

Call the callback with an error as the first argument, or output as the second argument.

For a multi-stage production flow, use pipeline() instead of chaining bare pipe() calls.

Byte mode and object mode

Streams normally work with strings, buffers, and typed arrays.

Use object mode when each chunk is a JavaScript value:

const { Readable } = require('node:stream')

const users = Readable.from([
  { name: 'Flavio' },
  { name: 'Roger' },
])

Readable.from() uses object mode by default.

Object-mode buffer limits count objects. Byte-mode limits count bytes.

See the official Node.js documentation for streams, the promise-based pipeline(), and implementing stream classes.

Related posts about node: