Node.js: Node.js Streams

Last updated: 2026-08-26

Charlie was assigned an urgent task: to analyze 2GB of access logs on the server. As soon as he loaded them using fs.readFile, the memory immediately ran out—causing an OOM crash. His team leader said, “No matter how big the file is, don’t try to swallow it all at once—use Stream to process it bit by bit!” Charlie switched to createReadStream to process the logs line by line, and the memory usage remained stable at 50MB. From then on, he told everyone he met: “Stream is the Swiss Army knife for processing big data in Node.js.”

1. Basic Concepts of Streams

Stream is an abstract interface in Node.js for processing streaming data, where data is transmitted in chunks, like a stream of water, rather than being loaded into memory all at once.

• Key Concept: Processing data in chunks, allowing operations to begin without waiting for all the data to be available
• Inheritance Relationship: All streams are instances of EventEmitter and use events to notify of state changes.
• Use Cases: Reading and writing large files, network transmission, data compression and conversion
• Built-in modules: fs, zlib, crypto, http, and others all provide stream interfaces.
• Global Objects: The stream module provides the base classes Readable, Writable, Duplex, and Transform

(1) Four Types of Flows

(2) Two Types of Flow

The Readable stream has two operating modes:

2. A Detailed Explanation of Stream Events

All streams are based on EventEmitter and use an event-driven mechanism to pass data and state.

(1) General Events

▶ Example: Listening for Readable stream events

const fs = require('fs');
const rs = fs.createReadStream('./access.log', { highWaterMark: 64 * 1024 });

rs.on('data', (chunk) => {
  console.log(`Received ${chunk.length} Byte`);
});

rs.on('end', () => {
  console.log('Finished reading');
});

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

▶ Example: Listening for Writable stream events

const ws = fs.createWriteStream('./output.txt');

ws.on('finish', () => {
  console.log('All data has been written');
});

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

ws.write('Hello Stream');
ws.end();

• data Events are automatically triggered in flow mode
• The difference between end and finish: end indicates that reading is complete, and finish indicates that writing is complete.
• error Events must be listened to; otherwise, uncaught exceptions will cause the process to crash.
• close Not all streams trigger the event; it depends on the underlying implementation.

3. Chaining with pipe()

pipe() is Stream's most powerful method; it automatically connects the output of a readable stream to the input of a writable stream and automatically handles backpressure.

▶ Example: (1) Basic Usage of "pipe"

readable.pipe(writable);

pipe() Returns the target stream, so it can be chained:

readable.pipe(transform1).pipe(transform2).pipe(writable);

▶ Example: Copying Files

const fs = require('fs');
fs.createReadStream('./source.txt')
  .pipe(fs.createWriteStream('./dest.txt'));

▶ Example: Streaming HTTP Responses

const http = require('http');
const fs = require('fs');

http.createServer((req, res) => {
  res.writeHead(200, { 'Content-Type': 'text/plain' });
  fs.createReadStream('./large.txt').pipe(res);
}).listen(3000);

▶ Example: (2) Stream Pipe Data Flow Diagram

flowchart LR
    A[Readable<br/>Data Source] -->|chunk| B[Transform<br/>Data Transformation]
    B -->|chunk| C[Writable<br/>Data Endpoint]
    C -.->|backpressure| A
    B -.->|backpressure| A

    style A fill:#4CAF50,color:#fff
    style B fill:#FF9800,color:#fff
    style C fill:#2196F3,color:#fff

flowchart LR
    A[Readable<br/>Data Source] -->|chunk| B[Transform<br/>Data Transformation]
    B -->|chunk| C[Writable<br/>Data Endpoint]
    C -.->|backpressure| A
    B -.->|backpressure| A

    style A fill:#4CAF50,color:#fff
    style B fill:#FF9800,color:#fff
    style C fill:#2196F3,color:#fff

• pipe() Automatically manages the data flow rate; the read end pauses when the write end is busy
• pipe() Errors are not handled; events must be monitored separately for each stream error
• Using stream.pipeline() automatically handles cleanup and error propagation, making it safer than pipe()

4. fs Streams and File Operations

The fs module provides read and write streams related to the file system.

(1) createReadStream / createWriteStream

▶ Example: Reading a File Segment Within a Specified Range

const fs = require('fs');
const rs = fs.createReadStream('./big.bin', {
  start: 100,
  end: 199,
  highWaterMark: 32
});

rs.on('data', (chunk) => {
  console.log(chunk.length);
});

(2) readFile vs createReadStream Comparison

▶ Example: Processing Large Files Block by Block

const fs = require('fs');
let totalBytes = 0;

const rs = fs.createReadStream('./2gb.log');
rs.on('data', (chunk) => {
  totalBytes += chunk.length;
});
rs.on('end', () => {
  console.log(`Total ${totalBytes} Byte`);
});

• Large files must be processed using streams to avoid OOM errors
• highWaterMark The smaller the memory, the more power-efficient it is, but the more frequent the system calls.
• The start/end options enable reading files in chunks

5. Backpressure Mechanism

Backpressure is at the heart of flow control: when the write side cannot keep up with the read side’s push rate, “backpressure” is applied to the write side to pause the read side and prevent memory backlog.

(1) pipe() automatically handles backpressure

When using pipe(), backpressure is automatically managed internally by Node.js, requiring no manual intervention.

(2) Manual Backpressure Control

When pipe() is not used, you must manually evaluate the return value of write():

const fs = require('fs');
const rs = fs.createReadStream('./source.txt');
const ws = fs.createWriteStream('./dest.txt');

rs.on('data', (chunk) => {
  const canContinue = ws.write(chunk);
  if (!canContinue) {
    rs.pause();
    ws.once('drain', () => {
      rs.resume();
    });
  }
});

rs.on('end', () => {
  ws.end();
});

▶ Example: Observing the Effect of Back Pressure

const rs = fs.createReadStream('./big.log', { highWaterMark: 1024 });
const ws = fs.createWriteStream('./out.log', { highWaterMark: 512 });

let paused = 0;
rs.on('data', (chunk) => {
  const ok = ws.write(chunk);
  if (!ok) {
    paused++;
    rs.pause();
    ws.once('drain', () => rs.resume());
  }
});
rs.on('end', () => {
  console.log(`Backpressure triggered ${paused} times`);
  ws.end();
});

• write() Return false indicates that the internal buffer is full and reading should be paused
• drain This event indicates that the buffer has been flushed and writing can resume.
• Ignoring backpressure will cause memory usage to continue to grow, eventually leading to an OOM
• pipeline() is more highly recommended than pipe(); it automatically handles error propagation and resource cleanup.

6. Transform Stream

A Transform stream is a subclass of Duplex; its output is the transformed result of the input, and it is commonly used for data compression, encryption, and format conversion.

(1) The Difference Between Transform and Duplex

▶ Example: Custom Transform Stream (Case Conversion)

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

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

process.stdin.pipe(upper).pipe(process.stdout);

▶ Example: zlib-compressed files

const fs = require('fs');
const zlib = require('zlib');

fs.createReadStream('./access.log')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('./access.log.gz'));

▶ Example: Decompressing a zlib file

const fs = require('fs');
const zlib = require('zlib');

fs.createReadStream('./access.log.gz')
  .pipe(zlib.createGunzip())
  .pipe(fs.createWriteStream('./access_restored.log'));

• In _transform(chunk, encoding, callback), callback(null, data) pushes the transformation results
• callback(err) Passable error
• zlib.createGzip() / createGunzip() are the most commonly used built-in Transform streams
• You can also implement _flush(callback) processing for the remaining data when customizing the Transform stream.

7. Pausing and Resuming a Stream

The Readable stream is in pause mode by default and can be toggled using pause() and resume().

▶ Example: Pause and Resume Controls

const fs = require('fs');
const rs = fs.createReadStream('./big.log');
let count = 0;

rs.on('data', (chunk) => {
  count++;
  if (count % 10 === 0) {
    rs.pause();
    console.log(`Processed ${count} chunks, pause 1 second`);
    setTimeout(() => rs.resume(), 1000);
  }
});

▶ Example: Reading a file line by line (readline)

const fs = require('fs');
const readline = require('readline');

const rl = readline.createInterface({
  input: fs.createReadStream('./access.log'),
  crlfDelay: Infinity
});

rl.on('line', (line) => {
  if (line.includes('ERROR')) {
    console.log(line);
  }
});

rl.on('close', () => {
  console.log('The file has been read.');
});

• readline.createInterface Wraps a Readable stream as an interface for reading line by line
• crlfDelay: Infinity Correctly handles \r\n line breaks
• After pause(), the data event is no longer triggered until resume() is called.
• pipe() will switch the stream to live mode

8. Comprehensive Example: Log Processing Pipeline

Build a complete data processing pipeline: Read large log files → Parse line by line → Filter out error lines → Compress the output.

const fs = require('fs');
const zlib = require('zlib');
const { Transform } = require('stream');

const filterError = new Transform({
  transform(chunk, encoding, callback) {
    const lines = chunk.toString().split('\n');
    const errors = lines.filter(l => l.includes('ERROR')).join('\n');
    callback(null, errors ? errors + '\n' : '');
  }
});

const src = fs.createReadStream('./app.log');
const dest = fs.createWriteStream('./errors.log.gz');

src
  .pipe(filterError)
  .pipe(zlib.createGzip())
  .pipe(dest);

dest.on('finish', () => {
  console.log('The error log has been written in compressed form.');
});

src.on('error', (err) => {
  console.error('Read failed:', err.message);
});

• Every stage of the pipeline is a stream; data flows through in chunks, resulting in extremely low memory usage.
• pipe() Automatically handles back pressure between adjacent streams
• Errors must be monitored separately on each stream, or handled uniformly using pipeline()
• Transform stream filterError to retain only rows containing ERROR

❓ FAQ

• When should you use Stream? Use Stream when the amount of data exceeds available memory, when you need to process data as you read it, or when working with real-time data sources; for small files, it’s simpler to use readFile.
• Does pipe() automatically handle backpressure? Yes, pipe() internally checks the return value from the writable end write() and automatically calls pause()/resume() to manage the flow rate.
• How do I read a file line by line? Use readline.createInterface({ input: createReadStream(path) }) and listen for the line event to retrieve the data line by line.
• What is the difference between Transform and Duplex? In Duplex, reading and writing are independent and unrelated; in Transform, the output is generated by transforming the input, requiring only the implementation of the _transform() method.
• How should stream errors be handled? Listen for error events on each stream, or use stream.pipeline() to automatically propagate errors and clean up resources, thereby preventing process crashes caused by uncaught exceptions.
• What is an appropriate value for highWaterMark? The default of 64 KB is suitable for most scenarios; for processing large blocks of binary data, you can increase it to 256 KB, and when memory is limited, you can reduce it to 16 KB.
• Which should you choose: pipe() or pipeline()? We recommend using stream.pipeline(), as it automatically propagates errors and cleans up resources; pipe() does not handle errors and does not automatically close streams.

📖 Summary

• Core Concepts and Usage of Stream Basics
• A Detailed Explanation of the Core Concepts and Usage of Stream Events
• Core Concepts and Usage of Chainable Connections with pipe()
• Core Concepts and Usage of FS Streams and File Operations
• Core Concepts and Usage of the Backpressure Mechanism
• Core Concepts and Usage of the Transform Stream
• Key Concepts and Usage of Pausing and Resuming Streams
• Comprehensive Example: Core Concepts and Usage of the Log Processing Pipeline

📝 Exercises

1. Complete all the code examples in this lesson and make sure each one runs correctly.
2. Modify the comprehensive example and add your own extensions
3. Review the official documentation, identify 1–2 APIs not covered in this lesson, and write test code for them.
4. Reflection: How would you apply what you’ve learned in this lesson to a real-world project?
5. Try to combine what you’ve learned in this lesson with material from previous lessons to build a small project.

Web-Tutorial Tech Team

A team of developers maintaining programming tutorials. Each tutorial is written and reviewed by developers with expertise in that field. We work to keep our content accurate and reliable — if you spot an issue, please let us know.

About Us