# Node.js Streams

> Learn how Node.js streams process data in chunks, handle backpressure, connect with pipeline, and create readable, writable, and transform streams.

Author: [Flavio Copes](https://flaviocopes.com/about/) | Published: 2018-08-08 | Updated: 2026-07-18 | Topics: [Node.js](https://flaviocopes.com/tags/node/) | Canonical: https://flaviocopes.com/nodejs-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()`:

```js
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:

```js
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:

```js
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:

```js
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:

```js
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:

```js
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:

```js
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:

```js
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:

```js
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:

```js
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](https://nodejs.org/api/stream.html), the promise-based [`pipeline()`](https://nodejs.org/api/stream.html#streampromisespipeline), and [implementing stream classes](https://nodejs.org/api/stream.html#api-for-stream-implementers).