Node.js Streams: A Comprehensive Guide

Streams process data piece by piece, enabling efficient handling of large files and real-time data without loading everything into memory.

Stream Types#

Readable: Source of data (fs.createReadStream, http request)
Writable: Destination for data (fs.createWriteStream, http response)
Duplex: Both readable and writable (TCP socket)
Transform: Modify data as it passes through (compression, encryption)

Reading Streams#

import { createReadStream } from 'fs';
import { pipeline } from 'stream/promises';

// Basic readable stream
const readable = createReadStream('large-file.txt', {
  encoding: 'utf8',
  highWaterMark: 64 * 1024, // 64KB chunks
});

// Event-based reading
readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes`);
});

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

readable.on('error', (err) => {
  console.error('Error:', err);
});

// Async iteration (preferred)
async function processFile(path: string): Promise<void> {
  const stream = createReadStream(path, { encoding: 'utf8' });

  for await (const chunk of stream) {
    console.log(`Processing ${chunk.length} characters`);
  }
}

// Pausing and resuming
readable.pause();
// Do something
readable.resume();

Writing Streams#

import { createWriteStream } from 'fs';

const writable = createWriteStream('output.txt');

// Write data
writable.write('Hello, ');
writable.write('World!\n');

// End stream (signals no more data)
writable.end('Final line');

// Handle events
writable.on('finish', () => {
  console.log('All data written');
});

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

// Handle backpressure
function writeData(stream: Writable, data: string[]): Promise<void> {
  return new Promise((resolve, reject) => {
    let i = 0;

    function write() {
      let ok = true;

      while (i < data.length && ok) {
        const chunk = data[i];
        i++;

        if (i === data.length) {
          // Last chunk
          stream.write(chunk, (err) => {
            if (err) reject(err);
            else resolve();
          });
        } else {
          // Check if we can continue
          ok = stream.write(chunk);
        }
      }

      if (i < data.length) {
        // Wait for drain event
        stream.once('drain', write);
      }
    }

    stream.on('error', reject);
    write();
  });
}

Transform Streams#

import { Transform, TransformCallback } from 'stream';

// Custom transform stream
class UppercaseTransform extends Transform {
  _transform(
    chunk: Buffer,
    encoding: BufferEncoding,
    callback: TransformCallback
  ): void {
    const upper = chunk.toString().toUpperCase();
    this.push(upper);
    callback();
  }
}

// Usage
const uppercase = new UppercaseTransform();
process.stdin.pipe(uppercase).pipe(process.stdout);

// Line-by-line transform
class LineSplitter extends Transform {
  private buffer = '';

  _transform(chunk: Buffer, encoding: string, callback: TransformCallback): void {
    this.buffer += chunk.toString();
    const lines = this.buffer.split('\n');
    this.buffer = lines.pop() || '';

    for (const line of lines) {
      this.push(line + '\n');
    }

    callback();
  }

  _flush(callback: TransformCallback): void {
    if (this.buffer) {
      this.push(this.buffer);
    }
    callback();
  }
}

// JSON transform
class JSONParser extends Transform {
  constructor() {
    super({ objectMode: true }); // Enable object mode
  }

  _transform(chunk: Buffer, encoding: string, callback: TransformCallback): void {
    try {
      const obj = JSON.parse(chunk.toString());
      this.push(obj);
      callback();
    } catch (err) {
      callback(err as Error);
    }
  }
}

Pipeline#

import { pipeline } from 'stream/promises';
import { createReadStream, createWriteStream } from 'fs';
import { createGzip, createGunzip } from 'zlib';

// Compress file
async function compressFile(input: string, output: string): Promise<void> {
  await pipeline(
    createReadStream(input),
    createGzip(),
    createWriteStream(output)
  );
  console.log('Compression complete');
}

// Decompress file
async function decompressFile(input: string, output: string): Promise<void> {
  await pipeline(
    createReadStream(input),
    createGunzip(),
    createWriteStream(output)
  );
  console.log('Decompression complete');
}

// Multiple transforms
await pipeline(
  createReadStream('input.txt'),
  new LineSplitter(),
  new UppercaseTransform(),
  createWriteStream('output.txt')
);

// With error handling
try {
  await pipeline(source, transform, destination);
} catch (err) {
  console.error('Pipeline failed:', err);
}

HTTP Streaming#

import http from 'http';
import { createReadStream } from 'fs';
import { pipeline } from 'stream/promises';

// Stream file response
const server = http.createServer(async (req, res) => {
  if (req.url === '/video') {
    const stat = await fs.promises.stat('video.mp4');

    res.writeHead(200, {
      'Content-Type': 'video/mp4',
      'Content-Length': stat.size,
    });

    await pipeline(createReadStream('video.mp4'), res);
  }
});

// Stream with range support (video seeking)
const server = http.createServer(async (req, res) => {
  const stat = await fs.promises.stat('video.mp4');
  const fileSize = stat.size;
  const range = req.headers.range;

  if (range) {
    const parts = range.replace(/bytes=/, '').split('-');
    const start = parseInt(parts[0], 10);
    const end = parts[1] ? parseInt(parts[1], 10) : fileSize - 1;

    res.writeHead(206, {
      'Content-Range': `bytes ${start}-${end}/${fileSize}`,
      'Accept-Ranges': 'bytes',
      'Content-Length': end - start + 1,
      'Content-Type': 'video/mp4',
    });

    createReadStream('video.mp4', { start, end }).pipe(res);
  } else {
    res.writeHead(200, {
      'Content-Length': fileSize,
      'Content-Type': 'video/mp4',
    });

    createReadStream('video.mp4').pipe(res);
  }
});

// Stream request body
app.post('/upload', async (req, res) => {
  const writeStream = createWriteStream('upload.bin');

  await pipeline(req, writeStream);

  res.json({ message: 'Upload complete' });
});

Async Generators#

import { Readable } from 'stream';

// Create readable from async generator
async function* generateData() {
  for (let i = 0; i < 100; i++) {
    yield `Line ${i}\n`;
    await sleep(10);
  }
}

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

// Database cursor as stream
async function* fetchUsers(batchSize = 100) {
  let cursor: string | undefined;

  while (true) {
    const users = await db.user.findMany({
      take: batchSize,
      cursor: cursor ? { id: cursor } : undefined,
      skip: cursor ? 1 : 0,
    });

    if (users.length === 0) break;

    for (const user of users) {
      yield user;
    }

    cursor = users[users.length - 1].id;
  }
}

// Stream to response
app.get('/users/export', async (req, res) => {
  res.setHeader('Content-Type', 'application/json');
  res.write('[');

  let first = true;
  for await (const user of fetchUsers()) {
    if (!first) res.write(',');
    res.write(JSON.stringify(user));
    first = false;
  }

  res.write(']');
  res.end();
});

Backpressure#

// Handling backpressure properly
function copyFile(source: string, dest: string): Promise<void> {
  return new Promise((resolve, reject) => {
    const readable = createReadStream(source);
    const writable = createWriteStream(dest);

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

      if (!canContinue) {
        // Destination is full, pause reading
        readable.pause();

        writable.once('drain', () => {
          // Destination ready, resume reading
          readable.resume();
        });
      }
    });

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

    writable.on('finish', resolve);
    writable.on('error', reject);
    readable.on('error', reject);
  });
}

// Or use pipeline (handles backpressure automatically)
await pipeline(
  createReadStream(source),
  createWriteStream(dest)
);

Object Mode Streams#

// Streams that handle objects instead of buffers
class ObjectTransform extends Transform {
  constructor() {
    super({ objectMode: true });
  }

  _transform(obj: any, encoding: string, callback: TransformCallback): void {
    // Transform object
    const transformed = {
      ...obj,
      processed: true,
      timestamp: new Date(),
    };

    this.push(transformed);
    callback();
  }
}

// Usage
const transform = new ObjectTransform();

transform.write({ id: 1, name: 'Test' });
transform.on('data', (obj) => {
  console.log(obj); // { id: 1, name: 'Test', processed: true, timestamp: ... }
});

Best Practices#

Memory:
✓ Use streams for large files
✓ Set appropriate highWaterMark
✓ Handle backpressure
✓ Clean up on error

Error Handling:
✓ Always handle 'error' events
✓ Use pipeline for cleanup
✓ Destroy streams on error
✓ Check stream state

Performance:
✓ Tune chunk size for workload
✓ Use object mode for objects
✓ Avoid unnecessary transforms
✓ Profile memory usage

Streams enable efficient data processing in Node.js. Use pipeline for composing streams safely, handle backpressure to prevent memory issues, and leverage async iteration for cleaner code. Streams are essential for processing large files and real-time data.