How to Create Custom Stream in Node.js: Simple Guide

In Node.js, you create a custom stream by extending the Readable, Writable, or Duplex classes from the stream module and implementing required methods like _read() or _write(). This lets you control how data is read, written, or transformed in your stream.

Syntax

To create a custom stream, import the stream module and extend one of its classes:

• Readable for streams that produce data.
• Writable for streams that consume data.
• Duplex for streams that do both.

Implement the required internal methods like _read(size) for readable streams or _write(chunk, encoding, callback) for writable streams.

javascript

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

class CustomReadable extends Readable {
  _read(size) {
    // Implement how data is pushed
  }
}

const stream = new CustomReadable();

Example

This example creates a custom readable stream that emits numbers from 1 to 5, one per second, then ends.

javascript

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

class NumberStream extends Readable {
  constructor(max) {
    super();
    this.current = 1;
    this.max = max;
  }

  _read() {
    const i = this.current++;
    if (i > this.max) {
      this.push(null); // No more data
    } else {
      setTimeout(() => {
        this.push(String(i));
      }, 1000);
    }
  }
}

const numberStream = new NumberStream(5);
numberStream.on('data', chunk => {
  console.log('Received:', chunk.toString());
});
numberStream.on('end', () => {
  console.log('Stream ended');
});

Output

Received: 1
Received: 2
Received: 3
Received: 4
Received: 5
Stream ended

Common Pitfalls

Common mistakes when creating custom streams include:

• Not calling push(null) to signal the end of a readable stream, causing it to never end.
• For writable streams, forgetting to call the callback in _write(), which blocks further writes.
• Not handling backpressure properly by respecting the return value of push() or write().

javascript

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

// Wrong: missing callback call
class BadWritable extends Writable {
  _write(chunk, encoding, callback) {
    console.log('Writing:', chunk.toString());
    // callback() missing here causes stream to hang
  }
}

// Correct:
class GoodWritable extends Writable {
  _write(chunk, encoding, callback) {
    console.log('Writing:', chunk.toString());
    callback(); // Signal that write is done
  }
}

Quick Reference

Summary tips for custom streams:

• Extend Readable, Writable, or Duplex from stream.
• Implement _read(size) for readable streams to push data.
• Implement _write(chunk, encoding, callback) for writable streams and always call callback().
• Call push(null) to end readable streams.
• Handle backpressure by checking return values of push() and write().

Key Takeaways

Create custom streams by extending Node.js stream classes and implementing required methods.

Always call push(null) to signal the end of readable streams.

In writable streams, call the callback in _write() to avoid blocking.

Handle backpressure by respecting push() and write() return values.

Use custom streams to control how data flows in your Node.js applications.