Streams & Buffers

Node.js Intermediate — Streams & Buffers

Node.js Intermediate

Chapter 1 of 8 · Streams & Buffers

Streams & Buffers

Course 1 showed you how to read files with fs.readFile and fs.readFileSync. These work fine for small files — but try loading a 4 GB log file and Node reads the entire thing into memory before you can touch it. Streams solve this by processing data in small chunks as it arrives, keeping memory usage flat regardless of file size.

Buffers — Raw Binary Data in Node

Before streams make sense, you need to understand Buffers. A Buffer is Node's representation of raw binary data — a fixed-size chunk of memory outside the V8 heap. When a stream hands you data, it's a Buffer by default.

// Creating buffers
const buf1 = Buffer.alloc(10);              // 10 zero-filled bytes (safe default)
const buf2 = Buffer.from('Hello, Node!');    // from a string (UTF-8 by default)
const buf3 = Buffer.from([0x48, 0x69]);      // from byte values → "Hi"

// Reading and inspecting
console.log(buf2.toString());               // 'Hello, Node!'
console.log(buf2.toString('hex'));          // '48656c6c6f2c...'
console.log(buf2.toString('base64'));       // 'SGVsbG8sIE5vZGUh'
console.log(buf2.length);                   // 12 (bytes — not characters!)

// Concatenating
const combined = Buffer.concat([buf2, buf3]);
console.log(combined.toString());           // 'Hello, Node!Hi'
    

💡 Why Not Just Use Strings?

Strings in V8 are UTF-16 encoded. Binary data — images, compressed files, network packets — doesn't map cleanly to strings and will be corrupted by encoding conversions. Buffers give you direct byte access. When you need text, call .toString('utf8') yourself.

The Four Stream Types

Readable

A source of data you consume from. Examples: fs.createReadStream, http.IncomingMessage, process.stdin.

Writable

A destination you write data into. Examples: fs.createWriteStream, http.ServerResponse, process.stdout.

Duplex

Both readable and writable, but independently. A TCP socket is a classic example — you read incoming data and write outgoing data on the same socket.

Transform

A Duplex where output is derived from input. Examples: zlib.createGzip(), CSV parsers, encryption streams — data goes in one form, comes out another.

Readable Streams

Readable streams have two modes: flowing (data events fire automatically) and paused (you pull chunks manually). Attaching a data event listener switches to flowing mode.

const fs = require('fs');

const readable = fs.createReadStream('large-file.txt', {
    encoding: 'utf8',
    highWaterMark: 64 * 1024   // chunk size: 64 KB (default is 16 KB)
});

// Flowing mode — events drive the reading
readable.on('data', (chunk) => {
    console.log(`Received: ${chunk.length} bytes`);
});

readable.on('end', () => {
    console.log('No more data');
});

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

// Async iteration — cleanest modern approach (Node 12+)
async function processFile(path) {
    const readable = fs.createReadStream(path, { encoding: 'utf8' });

    for await (const chunk of readable) {
        console.log(`Got ${chunk.length} chars`);
    }
    console.log('Done');
}
    

Writable Streams

You push data in with write() and signal completion with end(). The return value of write() is critical — false means the internal buffer is full and you should pause.

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

// write() returns false when the buffer is full (backpressure signal)
const ok = writable.write('First line\n');

if (!ok) {
    // Don't write more until 'drain' fires
    writable.once('drain', () => {
        writable.write('Second line\n');
    });
}

// end() takes an optional final chunk, then closes the stream
writable.end('Last line\n', () => {
    console.log('File written and closed');
});

writable.on('error', console.error);
    

Transform Streams

Transform streams sit between a readable and a writable, modifying data as it passes through. You implement _transform() to process each chunk and _flush() for any cleanup at the end.

// Built-in transform: gzip compression via zlib
const zlib = require('zlib');

fs.createReadStream('input.txt')
    .pipe(zlib.createGzip())
    .pipe(fs.createWriteStream('input.txt.gz'));
    

// Custom transform — converts text to uppercase as it streams through
const { Transform } = require('stream');

class UpperCaseTransform extends Transform {
    _transform(chunk, encoding, callback) {
        // chunk is a Buffer by default — convert, transform, push
        this.push(chunk.toString().toUpperCase());
        callback();   // signal: ready for the next chunk
    }

    _flush(callback) {
        // Called once at the end — push any remaining buffered output here
        callback();
    }
}

fs.createReadStream('input.txt')
    .pipe(new UpperCaseTransform())
    .pipe(fs.createWriteStream('output.txt'));
    

⚠️ Always Call callback() in _transform()

Forgetting to call callback() stalls the stream silently — no error, no data, no progress. If processing fails, pass the error: callback(err). This is the most common transform stream bug.

pipe() and pipeline()

pipe() connects a readable to a writable and manages the data flow loop for you. pipeline() (from stream/promises, Node 10+) is the modern replacement — it properly destroys all streams if any one of them errors, preventing resource leaks.

// pipe() — simple but leaks streams on error
fs.createReadStream('input.txt')
    .pipe(zlib.createGzip())
    .pipe(fs.createWriteStream('output.gz'))
    .on('finish', () => console.log('Done'));
    

// pipeline() — preferred in production
const { pipeline } = require('stream/promises');

async function compress(input, output) {
    await pipeline(
        fs.createReadStream(input),
        zlib.createGzip(),
        fs.createWriteStream(output)
    );
    console.log(`Compressed ${input} → ${output}`);
}

compress('large-file.txt', 'large-file.txt.gz').catch(console.error);
    

Backpressure

Backpressure is what happens when a fast readable outpaces a slow writable — the internal buffer fills, memory climbs, and the whole point of streaming is defeated. pipe() and pipeline() handle this automatically by pausing the readable when the writable's buffer is full and resuming when it drains.

Readable ──data──▶ Writable
                        │
               buffer fills up
                        │
               write() returns false
                        │
        ◀── pause() ────┘    ← pipe/pipeline does this for you
                        │
               'drain' event fires
                        │
        ──── resume() ──▶    ← pipe/pipeline does this for you
    

// Manual backpressure — exactly what pipe() does under the hood
function copyFile(src, dest) {
    const r = fs.createReadStream(src);
    const w = fs.createWriteStream(dest);

    r.on('data', (chunk) => {
        const ok = w.write(chunk);
        if (!ok) {
            r.pause();                          // writable is full — stop reading
            w.once('drain', () => r.resume());  // resume when it drains
        }
    });

    r.on('end', () => w.end());
    r.on('error', (e) => w.destroy(e));
    w.on('error', console.error);
}

// In practice — just use pipeline():
await pipeline(fs.createReadStream(src), fs.createWriteStream(dest));
    

Quick Reference

Concept	Key API	When to use
`Buffer`	`Buffer.alloc(n)`, `Buffer.from()`, `.toString()`	Working with binary data or explicit encoding conversions
Readable stream	`fs.createReadStream()`, `on('data')`, `for await`	Reading large files or network data chunk by chunk
Writable stream	`fs.createWriteStream()`, `write()`, `end()`	Writing output without building it entirely in memory
Transform stream	`extends Transform`, `_transform(chunk, enc, cb)`	Modifying data in-flight: compress, encrypt, parse, convert
`pipe()`	`readable.pipe(writable)`	Quick connections — attach error handlers manually
`pipeline()`	`await pipeline(r, t, w)`	Production use — automatic cleanup on error, async/await friendly
Backpressure	`write()` → `false`, `'drain'` event	Handled automatically by `pipe()` / `pipeline()`

⚠️ Common Mistakes

Using pipe() in production without error handling — if the writable errors, the readable keeps flowing and memory climbs. Use pipeline() or attach 'error' listeners to every stream.

Forgetting callback() in _transform() — the stream stalls silently. No error message, no data flowing, no indication why. Always call callback() or callback(err).

Mixing event listeners and async iteration on the same stream — attaching a 'data' listener switches to flowing mode and will conflict with for await...of. Pick one approach per stream.

Coding Challenges

Challenge 1 — Line-by-Line File Reader

Write a function countLines(filePath) that uses a readable stream (not fs.readFile) to count the number of lines in a file. Handle the edge case where the final chunk doesn't end with a newline. The function should return a Promise that resolves with the total line count.

View sample solution ↗

Challenge 2 — Custom Transform: Word Censor

Build a CensorTransform class that extends Transform. It should accept an array of words to censor in its constructor, and replace each occurrence of those words in the stream with *** (case-insensitive). Test it by piping a readable through it to process.stdout.

View sample solution ↗

Challenge 3 — File Compressor with pipeline()

Write an async function compressFile(inputPath) that uses pipeline() to gzip-compress a file, saving it as inputPath + '.gz'. After compression, log the original size, compressed size, and percentage reduction. Handle errors with a clear error message and non-zero exit code.

View sample solution ↗