Readable Streams

Difficulty: Intermediate

Streams are one of the most powerful concepts in Node.js. A stream is an abstract interface for working with data that arrives (or is sent) piece by piece, rather than loading everything into memory at once. This is critical for performance when dealing with large files, network responses, or any data source that produces data over time. Readable streams represent a source of data that you can consume.

The most common way to create a readable stream is fs.createReadStream(path, options). Instead of loading an entire file into memory like readFile, createReadStream reads the file in chunks (the default chunk size is 64KB, controlled by the highWaterMark option). Each chunk triggers a 'data' event. When all data has been read, the 'end' event fires. If an error occurs (e.g., file not found), the 'error' event fires. This chunk-based approach means you can process a 10GB file without using more than 64KB of memory.

Readable streams operate in two modes: flowing and paused. In flowing mode, data is read from the source automatically and provided as fast as possible via 'data' events. You enter flowing mode by attaching a 'data' listener, calling stream.resume(), or piping to a writable stream. In paused mode (the default), you must explicitly call stream.read() to pull chunks of data. You can switch between modes: calling stream.pause() switches from flowing to paused, and attaching a 'data' listener or calling stream.resume() switches back to flowing.

Readable streams also support the async iterator protocol, which is the cleanest way to consume them in modern Node.js. You can use for-await-of to iterate over chunks: 'for await (const chunk of stream) { ... }'. This is syntactic sugar over the paused mode read protocol and automatically handles backpressure. The stream is paused between iterations, so the consumer controls the pace. Combined with async/await, this makes stream code as readable as synchronous code.

The highWaterMark option controls the internal buffer size. For file streams, it defaults to 65536 bytes (64KB). A higher value means fewer reads but more memory usage. A lower value means more frequent reads with less memory. For most applications, the default is fine. You can also specify an encoding option (e.g., 'utf8') to receive string chunks instead of Buffer chunks, which is convenient for text files.

Code examples

Reading a File with createReadStream

const fs = require('fs');

// Create a test file first
fs.writeFileSync('stream-test.txt', 'Hello from streams!\nLine 2\nLine 3');

// Create a readable stream
const stream = fs.createReadStream('stream-test.txt', {
  encoding: 'utf8',
  highWaterMark: 16 // Small buffer to demonstrate chunking
});

let chunkCount = 0;

stream.on('data', (chunk) => {
  chunkCount++;
  console.log(`Chunk ${chunkCount} (${chunk.length} chars): "${chunk}"`);
});

stream.on('end', () => {
  console.log(`\nStream ended. Total chunks: ${chunkCount}`);
  fs.unlinkSync('stream-test.txt');
});

stream.on('error', (err) => {
  console.error('Stream error:', err.message);
});

With a 16-byte highWaterMark, the 35-character file is read in three chunks. In real applications with the default 64KB buffer, most small files arrive in a single chunk. The key benefit is that even multi-gigabyte files use only highWaterMark bytes of memory.

Async Iteration Over Streams

const fs = require('fs');
const { Readable } = require('stream');

// Create a custom readable stream for demo
function createNumberStream(max) {
  let current = 1;
  return new Readable({
    objectMode: true,
    read() {
      if (current <= max) {
        this.push({ number: current, squared: current * current });
        current++;
      } else {
        this.push(null); // Signal end of stream
      }
    }
  });
}

async function main() {
  const stream = createNumberStream(5);

  // Modern approach: for-await-of
  const results = [];
  for await (const item of stream) {
    results.push(`${item.number}^2 = ${item.squared}`);
  }

  console.log('Results:');
  results.forEach(r => console.log(r));
}

main();

for-await-of is the cleanest way to consume streams. The custom readable stream pushes objects (objectMode:true) until pushing null to signal completion. This pattern works with any readable stream including file streams and HTTP responses.

Flowing vs Paused Mode

const { Readable } = require('stream');

// Create a stream that produces letters
const letters = ['A', 'B', 'C', 'D', 'E'];
let index = 0;

const stream = new Readable({
  read() {
    if (index < letters.length) {
      this.push(letters[index++]);
    } else {
      this.push(null);
    }
  }
});

// Initially paused. Attaching 'data' switches to flowing mode.
stream.on('data', (chunk) => {
  console.log('Received:', chunk.toString());

  // Pause after 'C'
  if (chunk.toString() === 'C') {
    stream.pause();
    console.log('Paused stream');

    // Resume after a delay
    setTimeout(() => {
      console.log('Resuming...');
      stream.resume();
    }, 100);
  }
});

stream.on('end', () => {
  console.log('Stream ended');
});

Attaching a 'data' listener puts the stream in flowing mode. Calling pause() stops data emission, and resume() restarts it. This is manual backpressure control. In practice, pipe() handles this automatically.

Key points

Concepts covered

createReadStream, data Event, end Event, Flowing vs Paused Mode, Highwater Mark