Streams

import fs from "node:fs";

const rs = fs.createReadStream("big.txt", {
  encoding:   "utf8",    // omit for Buffer chunks
  highWaterMark: 64 * 1024,  // chunk size hint (bytes)
  start: 0,
  end:   1023,           // inclusive — first 1 KB only
});

// Event-based
rs.on("data",  (chunk) => process.stdout.write(chunk));
rs.on("end",   () => console.log("done"));
rs.on("error", (err) => console.error(err));
rs.on("close", () => { /* fd closed */ });

// Async iterator (recommended — automatic backpressure)
for await (const chunk of rs) {
  process.stdout.write(chunk);
}

// Pause / resume
rs.pause();
rs.resume();
rs.isPaused();     // boolean
rs.readableLength; // bytes in internal buffer

import fs from "node:fs";

const ws = fs.createWriteStream("out.txt", {
  encoding:      "utf8",
  flags:         "w",         // "a" to append
  highWaterMark: 16384,
  mode:          0o644,
});

// write() returns false when buffer is full
const ok = ws.write("hello\n");
if (!ok) {
  // backpressure — pause the source
  rs.pause();
  ws.once("drain", () => rs.resume());
}

ws.end("final chunk\n");  // flush + close
ws.end();                 // no final data

ws.on("finish", () => console.log("all flushed"));
ws.on("close",  () => console.log("fd closed"));
ws.on("error",  console.error);

ws.writableLength;  // bytes in buffer
ws.writableEnded;   // end() called
ws.writableFinished;// all data flushed

// Simple pipe (handles backpressure automatically)
readable.pipe(writable);

// Chain transforms
fs.createReadStream("in.gz")
  .pipe(zlib.createGunzip())
  .pipe(fs.createWriteStream("out.txt"));

// Unpipe
readable.unpipe(writable);
readable.unpipe();   // unpipe all

// DANGER: pipe does NOT propagate errors — handle each stage
readable
  .on("error", cleanup)
  .pipe(transform)
  .on("error", cleanup)
  .pipe(writable)
  .on("error", cleanup);

import { pipeline } from "node:stream/promises";
import fs   from "node:fs";
import zlib from "node:zlib";

// Propagates errors and cleans up all streams automatically
await pipeline(
  fs.createReadStream("in.gz"),
  zlib.createGunzip(),
  fs.createWriteStream("out.txt"),
);

// With transform function (Node 16+)
await pipeline(
  fs.createReadStream("data.txt"),
  async function* (source) {
    for await (const chunk of source) {
      yield chunk.toString().toUpperCase();
    }
  },
  fs.createWriteStream("upper.txt"),
);

// stream/consumers (Node 16+)
import { text, json, buffer, arrayBuffer, blob } from "node:stream/consumers";

const str  = await text(readable);
const obj  = await json(readable);
const buf  = await buffer(readable);

// Manual
async function collect(stream) {
  const chunks = [];
  for await (const chunk of stream) chunks.push(Buffer.from(chunk));
  return Buffer.concat(chunks);
}

import { Readable } from "node:stream";

// From async generator (simplest)
function range(start, end) {
  return Readable.from(async function* () {
    for (let i = start; i <= end; i++) yield `${i}\n`;
  }());
}

// Class-based
class CounterStream extends Readable {
  constructor(opts) {
    super({ ...opts, objectMode: true });
    this.n = 0;
  }
  _read(size) {
    if (this.n >= 10) {
      this.push(null);  // signal end
    } else {
      this.push(this.n++);
    }
  }
}

// Readable.from — wrap any iterable or async iterable
const letters = Readable.from(["a", "b", "c"]);
const numbers = Readable.from([1, 2, 3], { objectMode: true });

import { Writable } from "node:stream";

class LogStream extends Writable {
  _write(chunk, encoding, callback) {
    console.log(chunk.toString());
    callback();          // signal done — call with error to abort
  }
  _final(callback) {
    console.log("stream ended");
    callback();
  }
}

import { Transform } from "node:stream";

class UpperCase extends Transform {
  _transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
}

// Inline via object (uses the same Transform import)
const upper = new Transform({
  transform(chunk, enc, cb) {
    cb(null, chunk.toString().toUpperCase());
  },
});

// objectMode: true — chunks can be any JS value, not just string/Buffer
const readable = new Readable({ objectMode: true, read() {} });
readable.push({ id: 1, name: "Alice" });
readable.push({ id: 2, name: "Bob" });
readable.push(null);  // end

for await (const obj of readable) {
  console.log(obj.name);
}

// Source must respect writable's capacity
writable.on("drain", () => {
  // buffer cleared — safe to write again
  source.resume();
});

if (!writable.write(data)) {
  source.pause();   // stop producing until drain
}

// With pipeline() or async iterators, backpressure is automatic

// Standard I/O
process.stdin    // Readable
process.stdout   // Writable
process.stderr   // Writable

// Compression
import zlib from "node:zlib";
zlib.createGzip()         // Readable → compressed
zlib.createGunzip()       // compressed → original
zlib.createBrotliCompress()
zlib.createBrotliDecompress()
zlib.createDeflate()
zlib.createInflate()

// Crypto
import crypto from "node:crypto";
const iv = crypto.randomBytes(12);        // GCM needs a unique IV per key
crypto.createCipheriv("aes-256-gcm", key, iv)    // Transform — encrypt
crypto.createDecipheriv("aes-256-gcm", key, iv)  // Transform — decrypt
crypto.createHash("sha256")       // Transform — hash computation
crypto.createHmac("sha256", key)  // Transform — HMAC

// Net
import net from "node:net";
net.createConnection(port, host)  // Duplex socket

import { pipeline, finished } from "node:stream/promises";

// finished — resolves when stream is finished/closed, rejects on error
await finished(writable);

// pipeline — cleans up all streams on error
try {
  await pipeline(src, transform, dest);
} catch (err) {
  console.error("Pipeline failed:", err);
}