Readable Streams

A Readable stream is a source you pull data out of — a file being read, an incoming HTTP request, process.stdin, the output of a child process. The tricky part is that a readable has two modes of operation with very different behavior, and an internal buffer governed by highWaterMark. This page covers how to consume readables correctly and how their flow control actually works.

The modern way: async iteration

import { createReadStream } from 'node:fs'

const stream = createReadStream('big.log', { encoding: 'utf8' })

let lines = 0
for await (const chunk of stream) {
  lines += chunk.split('\n').length - 1
}
console.log('approx lines:', lines)

`for await` is the recommended consumer

Async iteration pulls one chunk at a time, pauses the stream while your loop body runs (automatic backpressure), surfaces errors through normal `try/catch`, and destroys the stream when the loop ends or throws. It's the safest default — prefer it over manual event wiring unless you need pause/resume or multiple consumers.

Two modes: paused and flowing

Every readable starts paused and switches to flowing depending on how you consume it. This single fact explains most stream confusion:

Mode	How you enter it	How data arrives
Paused (default)	Initial state; call `.pause()`	You pull with `.read()`
Flowing	Attach a `data` listener, `.pipe()`, or `.resume()`	Data is pushed at you via `data` events

Attaching a `data` listener starts the flow immediately — and you can lose data

The moment you call `stream.on('data', …)`, the stream switches to flowing mode and starts emitting. If you attach the listener *late* (e.g. after an `await` on the next tick), chunks emitted before then are gone. Similarly, registering a `data` handler and never reading drains the source as fast as possible with **no backpressure**. Async iteration and `pipe` avoid both traps.

The classic event API

import { createReadStream } from 'node:fs'

const stream = createReadStream('file.txt')
const chunks = []

stream.on('data',  (chunk) => chunks.push(chunk))     // flowing mode
stream.on('end',   () => {
  const all = Buffer.concat(chunks)
  console.log('total bytes:', all.length)
})
stream.on('error', (err) => console.error(err))

Event	Fires when
`data`	A chunk is available (flowing mode)
`end`	No more data — all chunks consumed
`error`	Something failed; stream may not emit `end`
`close`	Stream and its resources are closed
`readable`	Data is available to `.read()` (paused mode)

Always handle `error` — an unhandled stream error crashes the process

A readable that emits `error` with no listener throws, taking down your process. Every stream you consume by hand needs an `error` handler. (`pipeline` and `for await` handle this for you — another reason to prefer them.) Also note `end` does **not** fire on error, so don't put cleanup only in `end`.

Paused mode: pulling with read()

In paused mode you pull explicitly. Listen for readable, then drain with read() until it returns null (buffer empty for now):

const stream = createReadStream('file.txt')

stream.on('readable', () => {
  let chunk
  while ((chunk = stream.read()) !== null) {
    console.log(`read ${chunk.length} bytes`)
  }
})
stream.on('end', () => console.log('done'))

`read(n)` lets you pull a specific byte count

Passing a size — `stream.read(16)` — returns exactly that many bytes (or `null` if fewer are buffered), which is handy for parsing fixed-width binary formats. Without an argument, `read()` returns whatever is currently buffered. `read()` returning `null` means "nothing right now," *not* end-of-stream — wait for the next `readable` event.

highWaterMark: the internal buffer

A readable buffers data internally up to its highWaterMark (default 64 KB for byte streams, 16 in object mode). When the buffer is full, the stream stops pulling from the underlying source until you consume some — this is backpressure working at the source.

import { createReadStream } from 'node:fs'

// Smaller buffer → more, smaller chunks; larger → fewer, bigger chunks
const stream = createReadStream('data.bin', { highWaterMark: 16 * 1024 })  // 16 KB

`highWaterMark` is a threshold, not a hard chunk size

It sets *when* the stream stops reading ahead, and it influences chunk sizes, but it's not a guarantee that every chunk is exactly that size — the last chunk is usually smaller, and the source may hand over less. Tune it to trade memory (smaller) against syscall overhead (larger); the 64 KB default is sensible for most file work.

Object mode

By default readables emit Buffer/string chunks. In object mode each chunk is an arbitrary JavaScript value — perfect for pipelines of parsed records (database rows, JSON objects) rather than bytes:

import { Readable } from 'node:stream'

// Emit objects, not bytes:
const users = Readable.from([
  { id: 1, name: 'Ada' },
  { id: 2, name: 'Linus' },
])

for await (const user of users) {
  console.log(user.name)   // 'Ada', then 'Linus'
}

`Readable.from(iterable)` is the easy constructor

`Readable.from` turns any iterable or async iterable (array, generator, async generator) into a readable stream — automatically in object mode for non-Buffer values. It's the simplest way to feed existing data into a stream pipeline without implementing `_read` by hand.

Consuming the whole thing

import { createReadStream } from 'node:fs'

// Collect to one buffer/string (only for data that fits in memory!):
async function readAll(stream) {
  const chunks = []
  for await (const chunk of stream) chunks.push(chunk)
  return Buffer.concat(chunks)
}

const buf = await readAll(createReadStream('config.json'))
console.log(buf.toString('utf8'))

Buffering a whole stream defeats its purpose

Collecting every chunk into an array re-introduces the unbounded-memory problem streams exist to avoid. Only do it for data you *know* is small. For large inputs, transform and forward chunks as they arrive (via `pipeline`) rather than accumulating them.

Quick reference

Goal	Approach
Consume safely (default)	`for await (const chunk of stream)`
Pipe to a writable	`pipeline(readable, writable)`
Fine-grained pull	paused mode + `read()`
From an array/iterable	`Readable.from(iterable)`
Tune buffering	`{ highWaterMark: bytes }`

Now the other end — sinks you push data into: [Writable Streams](/nodejs/writable-streams).