Node.js Stream Processing

Streams enable efficient data processing for large datasets. Here's how to use them.

Stream Types#

const { Readable, Writable, Transform, Duplex } = require('stream');
const fs = require('fs');

// Four types of streams:
// 1. Readable - source of data
// 2. Writable - destination for data
// 3. Duplex - both readable and writable
// 4. Transform - modifies data as it passes through

// Reading a file as stream
const readStream = fs.createReadStream('large-file.txt');
readStream.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes`);
});

// Writing to a stream
const writeStream = fs.createWriteStream('output.txt');
writeStream.write('Hello, ');
writeStream.write('World!');
writeStream.end();

Readable Streams#

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

// Creating a readable stream
class CounterStream extends Readable {
  constructor(max) {
    super();
    this.max = max;
    this.current = 0;
  }

  _read() {
    if (this.current <= this.max) {
      this.push(String(this.current++));
    } else {
      this.push(null); // Signal end of stream
    }
  }
}

const counter = new CounterStream(10);
counter.on('data', (chunk) => console.log(chunk.toString()));

// Using Readable.from() for iterables
const numbers = Readable.from([1, 2, 3, 4, 5]);

// Async generator as source
async function* generateData() {
  for (let i = 0; i < 5; i++) {
    yield `Item ${i}\n`;
    await new Promise(resolve => setTimeout(resolve, 100));
  }
}

const asyncStream = Readable.from(generateData());

// Object mode for non-string data
const objectStream = new Readable({
  objectMode: true,
  read() {
    this.push({ id: 1, name: 'Item' });
    this.push(null);
  },
});

Writable Streams#

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

// Creating a writable stream
class LogStream extends Writable {
  _write(chunk, encoding, callback) {
    console.log(`LOG: ${chunk.toString()}`);
    callback(); // Signal completion
  }

  _writev(chunks, callback) {
    // Handle multiple chunks at once
    chunks.forEach(({ chunk }) => {
      console.log(`LOG: ${chunk.toString()}`);
    });
    callback();
  }
}

const logger = new LogStream();
logger.write('Hello');
logger.write('World');
logger.end();

// Object mode writable
class DatabaseWriter extends Writable {
  constructor() {
    super({ objectMode: true });
    this.records = [];
  }

  _write(record, encoding, callback) {
    this.records.push(record);
    console.log(`Saved record: ${record.id}`);
    callback();
  }

  _final(callback) {
    console.log(`Total records: ${this.records.length}`);
    callback();
  }
}

// Handle errors
const writer = new DatabaseWriter();
writer.on('error', (err) => console.error('Write error:', err));
writer.on('finish', () => console.log('All writes complete'));

Transform Streams#

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

// Basic transform
class UppercaseTransform extends Transform {
  _transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
}

// JSON parsing transform
class JSONParser extends Transform {
  constructor() {
    super({ objectMode: true });
    this.buffer = '';
  }

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

    lines.forEach(line => {
      if (line.trim()) {
        try {
          this.push(JSON.parse(line));
        } catch (err) {
          this.emit('error', err);
        }
      }
    });
    callback();
  }

  _flush(callback) {
    if (this.buffer.trim()) {
      try {
        this.push(JSON.parse(this.buffer));
      } catch (err) {
        this.emit('error', err);
      }
    }
    callback();
  }
}

// Chunking transform
class ChunkSplitter extends Transform {
  constructor(chunkSize) {
    super();
    this.chunkSize = chunkSize;
    this.buffer = Buffer.alloc(0);
  }

  _transform(chunk, encoding, callback) {
    this.buffer = Buffer.concat([this.buffer, chunk]);

    while (this.buffer.length >= this.chunkSize) {
      this.push(this.buffer.slice(0, this.chunkSize));
      this.buffer = this.buffer.slice(this.chunkSize);
    }
    callback();
  }

  _flush(callback) {
    if (this.buffer.length > 0) {
      this.push(this.buffer);
    }
    callback();
  }
}

Piping Streams#

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

// Basic piping
const source = fs.createReadStream('input.txt');
const dest = fs.createWriteStream('output.txt');
source.pipe(dest);

// Chain multiple transforms
fs.createReadStream('file.txt')
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream('file.txt.gz'));

// Decompress
fs.createReadStream('file.txt.gz')
  .pipe(zlib.createGunzip())
  .pipe(fs.createWriteStream('file-restored.txt'));

// Pipeline with error handling (recommended)
const { pipeline } = require('stream');

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

// Promise-based pipeline
const { pipeline: pipelinePromise } = require('stream/promises');

async function processFile() {
  await pipelinePromise(
    fs.createReadStream('input.txt'),
    new UppercaseTransform(),
    fs.createWriteStream('output.txt')
  );
  console.log('Done');
}

Backpressure Handling#

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

// Manual backpressure handling
const readable = fs.createReadStream('large-file.txt');
const writable = fs.createWriteStream('output.txt');

readable.on('data', (chunk) => {
  const canContinue = writable.write(chunk);

  if (!canContinue) {
    // Pause reading until writer drains
    readable.pause();
    writable.once('drain', () => {
      readable.resume();
    });
  }
});

readable.on('end', () => {
  writable.end();
});

// Automatic with pipe (handles backpressure)
readable.pipe(writable);

// Slow consumer simulation
class SlowWriter extends Writable {
  constructor() {
    super({ highWaterMark: 1024 }); // Buffer size
  }

  _write(chunk, encoding, callback) {
    // Simulate slow processing
    setTimeout(() => {
      console.log(`Processed ${chunk.length} bytes`);
      callback();
    }, 100);
  }
}

// Fast producer
class FastReader extends Readable {
  constructor() {
    super({ highWaterMark: 1024 });
    this.counter = 0;
  }

  _read() {
    if (this.counter < 100) {
      const data = `Data chunk ${this.counter++}\n`;
      const canPush = this.push(data);
      if (!canPush) {
        console.log('Backpressure: reader paused');
      }
    } else {
      this.push(null);
    }
  }
}

Async Iteration#

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

// Async iteration over stream
async function processLines(filename) {
  const stream = fs.createReadStream(filename);

  for await (const chunk of stream) {
    console.log(`Chunk: ${chunk.length} bytes`);
  }
}

// Line-by-line processing
async function readLines(filename) {
  const rl = readline.createInterface({
    input: fs.createReadStream(filename),
    crlfDelay: Infinity,
  });

  for await (const line of rl) {
    console.log(`Line: ${line}`);
  }
}

// Custom async iterable stream
async function* csvParser(stream) {
  let buffer = '';

  for await (const chunk of stream) {
    buffer += chunk.toString();
    const lines = buffer.split('\n');
    buffer = lines.pop();

    for (const line of lines) {
      yield line.split(',');
    }
  }

  if (buffer.trim()) {
    yield buffer.split(',');
  }
}

// Usage
async function parseCSV(filename) {
  const stream = fs.createReadStream(filename);

  for await (const row of csvParser(stream)) {
    console.log(row);
  }
}

HTTP Streaming#

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

// Stream file response
const server = http.createServer((req, res) => {
  const stream = fs.createReadStream('large-file.txt');
  res.writeHead(200, { 'Content-Type': 'text/plain' });
  stream.pipe(res);
});

// Stream with compression
const zlib = require('zlib');

const compressedServer = http.createServer((req, res) => {
  const acceptEncoding = req.headers['accept-encoding'] || '';

  if (acceptEncoding.includes('gzip')) {
    res.writeHead(200, {
      'Content-Type': 'text/plain',
      'Content-Encoding': 'gzip',
    });
    fs.createReadStream('file.txt')
      .pipe(zlib.createGzip())
      .pipe(res);
  } else {
    res.writeHead(200, { 'Content-Type': 'text/plain' });
    fs.createReadStream('file.txt').pipe(res);
  }
});

// Stream request body
const uploadServer = http.createServer((req, res) => {
  if (req.method === 'POST') {
    const writeStream = fs.createWriteStream('upload.txt');
    req.pipe(writeStream);

    writeStream.on('finish', () => {
      res.writeHead(200);
      res.end('Upload complete');
    });
  }
});

Error Handling#

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

// Proper error handling with pipeline
async function safeProcess() {
  try {
    await pipeline(
      fs.createReadStream('input.txt'),
      new Transform({
        transform(chunk, encoding, callback) {
          try {
            // Process chunk
            callback(null, chunk);
          } catch (err) {
            callback(err);
          }
        },
      }),
      fs.createWriteStream('output.txt')
    );
  } catch (err) {
    console.error('Stream processing failed:', err);
    // Cleanup handled automatically by pipeline
  }
}

// AbortController for cancellation
const { AbortController } = require('abort-controller');

async function cancellableProcess(signal) {
  try {
    await pipeline(
      fs.createReadStream('input.txt'),
      fs.createWriteStream('output.txt'),
      { signal }
    );
  } catch (err) {
    if (err.name === 'AbortError') {
      console.log('Processing cancelled');
    } else {
      throw err;
    }
  }
}

const controller = new AbortController();
setTimeout(() => controller.abort(), 5000);
cancellableProcess(controller.signal);

Best Practices#

Memory Management:
✓ Use streams for large data
✓ Set appropriate highWaterMark
✓ Handle backpressure properly
✓ Clean up resources on error

Error Handling:
✓ Use pipeline() over pipe()
✓ Handle errors on all streams
✓ Implement _destroy() for cleanup
✓ Use AbortController for cancellation

Performance:
✓ Use object mode sparingly
✓ Batch small writes
✓ Consider worker threads for CPU tasks
✓ Profile memory usage

Patterns:
✓ Prefer transform streams
✓ Use async iteration when possible
✓ Chain streams with pipeline
✓ Implement proper backpressure

Node.js streams enable efficient processing of large data sets. Use readable streams for data sources, writable for destinations, and transform for modifications. Handle backpressure properly, use pipeline for error handling, and leverage async iteration for cleaner code.