Piping & Transform Streams

Difficulty: Intermediate

Piping is the mechanism that connects a readable stream to a writable stream, automatically handling data flow and backpressure. It is the preferred way to move data between streams because it handles all the complexity of read/write coordination, pausing, and resuming. The pipe method was one of the earliest features that made Node.js streams elegant and powerful.

The stream.pipe(destination) method takes a writable stream as its argument and returns the destination stream, allowing you to chain multiple pipes. For example: readStream.pipe(transformStream).pipe(writeStream). When you pipe, the readable stream automatically pauses when the writable stream's buffer is full (backpressure) and resumes when it drains. This prevents memory exhaustion without any manual intervention.

However, pipe has a significant limitation: it does not propagate errors. If the source stream errors, the destination stream is not automatically closed, leading to resource leaks. The stream.pipeline() function (from 'stream' or 'stream/promises') solves this. It connects multiple streams together and properly handles cleanup when any stream in the chain errors or ends. Always use pipeline instead of pipe in production code. The promise-based version (from 'stream/promises') returns a Promise that resolves when the pipeline completes.

Transform streams are duplex streams (both readable and writable) that modify data as it passes through. They sit in the middle of a pipeline, receiving chunks from a readable stream, processing them, and pushing the results to a writable stream. Common use cases include compression (zlib.createGzip()), encryption, data parsing, format conversion, and filtering. You create a custom Transform by extending the Transform class and implementing the _transform(chunk, encoding, callback) method.

The _transform method receives each chunk, processes it, and calls this.push() to send the transformed data downstream. Call callback() when you are done processing the chunk (or callback(error) if an error occurred). You can push zero, one, or multiple chunks for each input chunk. The _flush(callback) method is called when the stream ends and gives you a chance to push any remaining buffered data. Transform streams are the building blocks of stream processing pipelines, enabling composable data transformations.

Code examples

Piping Streams Together

const fs = require('fs');
const { pipeline } = require('stream/promises');
const { Transform } = require('stream');

// Create a test file
fs.writeFileSync('input.txt', 'hello world\nnode streams\nare powerful');

// Transform: uppercase each line
const toUpperCase = new Transform({
  transform(chunk, encoding, callback) {
    callback(null, chunk.toString().toUpperCase());
  }
});

// Using pipe (simple but no error propagation)
const readStream = fs.createReadStream('input.txt', 'utf8');
const writeStream = fs.createWriteStream('output.txt');

readStream.pipe(toUpperCase).pipe(writeStream);

writeStream.on('finish', () => {
  const result = fs.readFileSync('output.txt', 'utf8');
  console.log('Piped result:');
  console.log(result);

  fs.unlinkSync('input.txt');
  fs.unlinkSync('output.txt');
});

The readable stream pipes through the transform (uppercase) and into the writable stream. pipe() handles backpressure automatically. The chain reads like Unix pipes: input | transform | output.

pipeline() with Proper Error Handling

const { pipeline } = require('stream/promises');
const { Transform, Readable, Writable } = require('stream');

// Create a number-generating readable stream
function numberSource(max) {
  let n = 1;
  return new Readable({
    objectMode: true,
    read() {
      this.push(n <= max ? { value: n++ } : null);
    }
  });
}

// Transform: double the value
const doubler = new Transform({
  objectMode: true,
  transform(chunk, enc, cb) {
    cb(null, { ...chunk, value: chunk.value * 2 });
  }
});

// Transform: format as string
const formatter = new Transform({
  objectMode: true,
  transform(chunk, enc, cb) {
    cb(null, `Value: ${chunk.value}\n`);
  }
});

// Collect output
let output = '';
const collector = new Writable({
  write(chunk, enc, cb) {
    output += chunk.toString();
    cb();
  }
});

async function main() {
  await pipeline(
    numberSource(5),
    doubler,
    formatter,
    collector
  );
  console.log(output.trim());
}

main().catch(console.error);

pipeline connects four streams: source -> doubler -> formatter -> collector. If any stream errors, pipeline cleans up all streams and rejects the promise. This is much safer than chaining pipe() calls which do not handle errors.

Custom Transform Stream: Line Counter

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

class LineCounter extends Transform {
  constructor(options) {
    super(options);
    this.lineNumber = 0;
    this.buffer = '';
  }

  _transform(chunk, encoding, callback) {
    this.buffer += chunk.toString();
    const lines = this.buffer.split('\n');

    // Keep last partial line in buffer
    this.buffer = lines.pop();

    // Number complete lines
    for (const line of lines) {
      this.lineNumber++;
      this.push(`${this.lineNumber}: ${line}\n`);
    }
    callback();
  }

  _flush(callback) {
    // Process remaining buffer
    if (this.buffer.length > 0) {
      this.lineNumber++;
      this.push(`${this.lineNumber}: ${this.buffer}\n`);
    }
    callback();
  }
}

// Test it
const input = Readable.from(['Hello\nWorld\n', 'Node.js\nStreams']);
const counter = new LineCounter();
let result = '';

counter.on('data', (chunk) => { result += chunk; });
counter.on('end', () => { console.log(result.trim()); });

input.pipe(counter);

This Transform stream adds line numbers. _transform handles each chunk, splitting into lines and buffering partial lines. _flush processes any remaining data when the stream ends. This demonstrates the common pattern of accumulating a buffer across chunks.

Key points

Concepts covered

pipe, pipeline, Transform Class, Stream Chaining, Error Handling