الـ Streams والـ Backpressure: ملفات أكبر من الـ memory

import fs from "node:fs";
const mb = (n) => `${Math.round(n / 1024 / 1024)}MB`;

// نكتب ملف logs كبير، سطر سطر، ونحترم الـ backpressure
const out = fs.createWriteStream("access.log");
let waits = 0;
for (let i = 0; i < 1_000_000; i++) {
  const line = `${i},GET,/courses/${i % 50},${i % 7 === 0 ? 500 : 200},${(i * 37) % 900}\n`;
  if (!out.write(line)) {
    waits++;
    await new Promise((resolve) => out.once("drain", resolve));
  }
}
out.end();
await new Promise((resolve) => out.on("finish", resolve));

console.log("حجم الملف:", mb(fs.statSync("access.log").size));
console.log("مرات الاستنى على drain:", waits > 100 ? "كتير (ده طبيعي)" : waits);

$ node main.js
حجم الملف: 29MB
مرات الاستنى على drain: كتير (ده طبيعي)

import fs from "node:fs";
const mb = (n) => `${Math.round(n / 1024 / 1024)}MB`;

const before = process.memoryUsage().rss;
const text = fs.readFileSync("access.log", "utf8");
const errors = text.split("\n").filter((l) => l.includes(",500,")).length;
const after = process.memoryUsage().rss;

console.log("errors:", errors);
console.log("الـ memory زادت حوالي:", mb(after - before));

$ node main.js
errors: 142858
الـ memory زادت حوالي: 123MB

import fs from "node:fs";
import readline from "node:readline";
const mb = (n) => `${Math.round(n / 1024 / 1024)}MB`;

const before = process.memoryUsage().rss;
const lines = readline.createInterface({
  input: fs.createReadStream("access.log"),
  crlfDelay: Infinity,
});

let errors = 0;
for await (const line of lines) {
  if (line.includes(",500,")) errors++;
}
const after = process.memoryUsage().rss;

console.log("errors:", errors);
console.log("الـ memory زادت حوالي:", mb(Math.max(0, after - before)));

$ node main.js
errors: 142858
الـ memory زادت حوالي: 23MB

import fs from "node:fs";
import { pipeline } from "node:stream/promises";
import { Transform } from "node:stream";
import zlib from "node:zlib";
const mb = (n) => `${Math.round(n / 1024 / 1024)}MB`;

// Transform: بياخد chunks وبيطلع chunks تانية
let rest = "";
const onlyErrors = new Transform({
  transform(chunk, _enc, done) {
    const lines = (rest + chunk).split("\n");
    rest = lines.pop();
    const keep = lines.filter((l) => l.includes(",500,")).join("\n");
    done(null, keep ? keep + "\n" : "");
  },
});

await pipeline(
  fs.createReadStream("access.log"),
  onlyErrors,
  zlib.createGzip(),
  fs.createWriteStream("errors.log.gz"),
);
console.log("errors.log.gz:", mb(fs.statSync("errors.log.gz").size));

$ node main.js
errors.log.gz: 1MB

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

try {
  await pipeline(
    fs.createReadStream("missing.log"),
    fs.createWriteStream("copy.log"),
  );
} catch (err) {
  console.log("pipeline وقع:", err.code);
}
console.log("البرنامج لسه شغال");

$ node main.js
pipeline وقع: ENOENT
البرنامج لسه شغال

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

// access.log: id,method,path,status,ms

const lines = readline.createInterface({
  input: fs.createReadStream("access.log"),
  crlfDelay: Infinity,
});

const stats = {};
for await (const line of lines) {
  if (!line) continue;
  const [, , , status, ms] = line.split(",");
  stats[status] ??= { count: 0, totalMs: 0 };
  stats[status].count++;
  stats[status].totalMs += Number(ms);
}

const report = {};
for (const [status, s] of Object.entries(stats)) {
  report[status] = { count: s.count, avgMs: Math.round(s.totalMs / s.count) };
}

await pipeline(Readable.from([JSON.stringify(report)]), fs.createWriteStream("report.json"));
console.log(fs.readFileSync("report.json", "utf8"));

$ node main.js
{"200":{"count":857142,"avgMs":450},"500":{"count":142858,"avgMs":449}}