Streams and Pipes

Challenge

You need to process a 5GB log file, transform the data, and write results efficiently without loading everything in memory. Streams are your solution.

Explanation

Streams are Node.js’s primary abstraction for handling data flow. They allow processing data in chunks, reducing memory footprint and improving performance for large datasets.

Stream Types

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

// Readable Stream
const readStream = fs.createReadStream('input.txt');

// Writable Stream
const writeStream = fs.createWriteStream('output.txt');

// Duplex Stream (both read and write)
const duplexStream = fs.createReadStream('input.txt')
  .pipe(fs.createWriteStream('output.txt'));

// Transform Stream (modify data)
const transform = new Transform({
  transform(chunk, encoding, callback) {
    const data = chunk.toString().toUpperCase();
    this.push(data);
    callback();
  }
});

readStream.pipe(transform).pipe(writeStream);

Practical Stream Pipeline

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

// Process, transform, and compress data
fs.createReadStream('input.txt', { encoding: 'utf8' })
  .pipe(new Transform({
    transform(chunk, encoding, callback) {
      // Transform each chunk
      const lines = chunk.split('\n');
      const transformed = lines
        .filter(line => line.length > 0)
        .map(line => line.trim().toUpperCase())
        .join('\n');

      this.push(transformed);
      callback();
    }
  }))
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('output.txt.gz'));

Stream Error Handling

const fs = require('fs');

const stream = fs.createReadStream('file.txt');

// Handle individual events
stream.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes`);
});

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

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

// Or use pipeline (recommended)
const { pipeline } = require('stream');

pipeline(
  fs.createReadStream('input.txt'),
  transformStream,
  fs.createWriteStream('output.txt'),
  (err) => {
    if (err) console.error('Pipeline failed:', err);
    else console.log('Pipeline succeeded');
  }
);

Backpressure Handling

const fs = require('fs');

const readable = fs.createReadStream('large.txt');
const writable = fs.createWriteStream('output.txt');

let chunks = 0;

readable.on('data', (chunk) => {
  chunks++;

  // Write and check for backpressure
  const ok = writable.write(chunk);

  if (!ok) {
    console.log('Backpressure: pausing read');
    readable.pause();
  }
});

// Resume when writable is ready
writable.on('drain', () => {
  console.log(`${chunks} chunks written, resuming read`);
  readable.resume();
});

// Or use pipe (handles backpressure automatically)
readable.pipe(writable);