Streams

import { createReadStream, createWriteStream } from 'node:fs';
import { Readable, Writable, Transform, pipeline } from 'node:stream';
import { pipeline as pipelineAsync } from 'node:stream/promises';
import { createGzip, createGunzip } from 'node:zlib';
import { createHash } from 'node:crypto';

// 1) Read a big file in chunks
const rs = createReadStream('input.txt', { highWaterMark: 64 * 1024 });
for await (const chunk of rs) {
    process(chunk);     // each chunk is a Buffer
}

// 2) Pipe — automatic backpressure handling
import { createReadStream } from 'node:fs';
createReadStream('input.txt')
    .pipe(createGzip())
    .pipe(createWriteStream('input.txt.gz'));

// 3) pipeline — the modern way (handles errors + cleanup)
await pipelineAsync(
    createReadStream('big.log'),
    createGzip(),
    createWriteStream('big.log.gz'),
);

// 4) Transform — modify data on the way through
const toUpper = new Transform({
    transform(chunk, _encoding, callback) {
        callback(null, chunk.toString().toUpperCase());
    },
});

await pipelineAsync(
    createReadStream('input.txt'),
    toUpper,
    createWriteStream('UPPER.txt'),
);

// 5) Line-by-line via readline
import readline from 'node:readline';
const rl = readline.createInterface({ input: createReadStream('events.ndjson') });
let count = 0;
for await (const line of rl) {
    const evt = JSON.parse(line);
    if (evt.type === 'error') count++;
}
console.log('errors:', count);

// 6) Compute a hash without loading the file
const hash = createHash('sha256');
await pipelineAsync(createReadStream('image.png'), hash);
console.log(hash.digest('hex'));

// 7) Stream HTTP response — never buffer big payloads
import http from 'node:http';
const server = http.createServer((req, res) => {
    res.writeHead(200, { 'content-type': 'video/mp4' });
    pipeline(
        createReadStream('movie.mp4'),
        res,
        (err) => err && console.error('pipe failed', err),
    );
});

// 8) Object mode — pass JS objects, not Buffers
const objectStream = new Readable({
    objectMode: true,
    read() {
        this.push({ id: 1, name: 'Ada' });
        this.push({ id: 2, name: 'Bo'  });
        this.push(null);            // end
    },
});

const printer = new Writable({
    objectMode: true,
    write(obj, _enc, cb) {
        console.log(obj.name);
        cb();
    },
});

objectStream.pipe(printer);

// 9) Create a Readable from an iterable
const rs2 = Readable.from(async function* () {
    for (let i = 0; i < 5; i++) {
        yield Buffer.from(`row ${i}\n`);
    }
}());

// 10) Backpressure — write returns false when buffer is full
const ws = createWriteStream('out.txt');
for (let i = 0; i < 1_000_000; i++) {
    const ok = ws.write(`line ${i}\n`);
    if (!ok) await new Promise(r => ws.once('drain', r));
}
ws.end();

// 11) Convert between async iterables and streams
import { Readable } from 'node:stream';
const webStream = Readable.toWeb(nodeStream);          // → WHATWG ReadableStream
const nodeStream2 = Readable.fromWeb(webStream);       // back to Node

// 12) Common pitfalls
//   • Forgetting to handle 'error' event → process.exit on uncaught
//   • Mixing pipe() with async/await without proper error handling
//   • Reading + writing without await → race conditions on big files
//   • Using readFile / writeFile when streaming would save GBs of memory

import { createReadStream } from 'node:fs';
import { pipeline } from 'node:stream/promises';
await pipeline(createReadStream('big.log'), process.stdout);