الأحداث والتدفقات (Events & Streams) — Node.js و Express

الأحداث والتدفقات

نصف Node مبني على فكرتين: الأحداث (شيء حدث، من يهتم؟) والتدفقات (بيانات تصل قطعة قطعة). الخادم HTTP نفسه، وقراءة الملفات، والاتصالات بقاعدة البيانات — كلها تستخدمهما.

EventEmitter

const { EventEmitter } = require("node:events");

class OrderService extends EventEmitter {
  place(order) {
    // ... حفظ الطلب ...
    this.emit("order:placed", order);          // أعلن: من يهتم فليستمع
  }
}

const orders = new OrderService();
orders.on("order:placed", (o) => console.log("📧 بريد تأكيد إلى", o.email));
orders.on("order:placed", (o) => console.log("📦 إشعار المخزن بالطلب", o.id));
orders.once("order:placed", () => console.log("🎉 أول طلب اليوم!"));   // مرة واحدة فقط
orders.place({ id: 1, email: "sara@example.com" });
orders.place({ id: 2, email: "omar@example.com" });

- المستمعون يُستدعون بالترتيب وبشكل متزامن عند emit.

- حدث error خاص: إن أُطلق ولا مستمع له، يُرمى الخطأ ويسقط البرنامج. استمع لـ error دائماً على أي EventEmitter أو تدفق.

- انزع المستمعين عندما لا تحتاجهم (off) — تحذير "MaxListenersExceeded" غالباً يعني تسرباً.

- الفائدة: فصل الأجزاء — خدمة الطلبات لا تعرف شيئاً عن البريد أو المخزن.

لماذا التدفقات؟

fs.readFile على ملف سجلات 4GB يحاول وضعه كله في الذاكرة. التدفق يقرأ قطعة صغيرة (64KB افتراضياً)، يعالجها، ويطلب التالية — الذاكرة ثابتة مهما كبر الملف.

| النوع | المثال |

|---|---|

| Readable (مصدر) | fs.createReadStream، طلب HTTP الوارد |

| Writable (وجهة) | fs.createWriteStream، استجابة HTTP |

| Duplex | اتصال TCP |

| Transform (يقرأ ويكتب معدّلاً) | الضغط zlib.createGzip()، التشفير، تحويل CSV |

pipeline: ربط آمن

const fs = require("node:fs");
const os = require("node:os");
const path = require("node:path");
const zlib = require("node:zlib");
const { pipeline } = require("node:stream/promises");

async function main() {
  const src = path.join(os.tmpdir(), "kodax-big.log");
  fs.writeFileSync(src, "سطر سجل\n".repeat(50_000));
  await pipeline(fs.createReadStream(src), zlib.createGzip(), fs.createWriteStream(src + ".gz"));
  console.log("الأصل:", fs.statSync(src).size, "بايت ← مضغوط:", fs.statSync(src + ".gz").size);
  fs.rmSync(src); fs.rmSync(src + ".gz");
}
main().catch(console.error);

استخدم pipeline بدل .pipe() القديم: ينقل الأخطاء من أي مرحلة ويغلق كل التدفقات وينظفها عند الفشل.

الضغط العكسي (Backpressure)

إذا كان المصدر أسرع من الوجهة (قراءة قرص سريع وإرسال لشبكة بطيئة)، تتراكم البيانات في الذاكرة. التدفقات تحل ذلك: write() تعيد false عندما يمتلئ مخزن الوجهة، فيتوقف المصدر حتى حدث drain. pipeline يدير ذلك تلقائياً — سبب إضافي لاستخدامه.

سطراً بسطر

const fs = require("node:fs");
const os = require("node:os");
const path = require("node:path");
const readline = require("node:readline");

async function countErrors(file) {
  const rl = readline.createInterface({ input: fs.createReadStream(file), crlfDelay: Infinity });
  let errors = 0;
  for await (const line of rl) if (line.includes("ERROR")) errors++;   // سطر واحد في الذاكرة
  return errors;
}

const file = path.join(os.tmpdir(), "kodax-app.log");
fs.writeFileSync(file, ["INFO start", "ERROR db timeout", "INFO ok", "ERROR disk full"].join("\n"));
countErrors(file).then((n) => { console.log("أخطاء:", n); fs.rmSync(file); });

for await على التدفقات أوضح طريقة لمعالجتها في الكود الحديث.

في الخوادم

- رفع ملف كبير: مرّر req (تدفق قراءة) مباشرة إلى التخزين بدل تجميعه في Buffer.

- تنزيل ملف: pipeline(fs.createReadStream(file), res).

- تصدير CSV لمليون صف: اقرأ من قاعدة البيانات بمؤشر (cursor) واكتب للاستجابة صفاً بصف.

تمرين

اكتب سكربتاً يقرأ ملف CSV كبيراً سطراً بسطر، ويكتب ملفاً جديداً فيه الصفوف التي عمود "الحالة" فيها "paid" فقط، باستخدام Transform و pipeline. قِس استهلاك الذاكرة (process.memoryUsage().rss) مع ملف 100MB.

const { EventEmitter } = require("node:events");
const { Transform } = require("node:stream");
const { pipeline } = require("node:stream/promises");
const fs = require("node:fs");
const os = require("node:os");
const path = require("node:path");

// ===== 1) أحداث: خدمة لا تعرف من يستمع إليها =====
class ImportJob extends EventEmitter {}
const job = new ImportJob();
let processed = 0;
job.on("progress", (n) => { processed = n; });
job.on("error", (err) => console.error("خطأ في الاستيراد:", err.message));   // استمع لـ error دائماً
job.once("done", (stats) => console.log("✅ انتهى:", stats));

// ===== 2) Transform: CSV ← نحتفظ بالطلبات المدفوعة ونحسب المجموع =====
function filterPaid() {
  let header = null;
  let buffered = "";
  let total = 0;
  let count = 0;
  return new Transform({
    readableObjectMode: false,
    transform(chunk, _enc, callback) {
      buffered += chunk.toString("utf8");
      const lines = buffered.split("\n");
      buffered = lines.pop();                       // آخر سطر قد يكون ناقصاً — نكمله مع القطعة التالية
      const out = [];
      for (const line of lines) {
        if (!header) { header = line; out.push(line); continue; }
        const [id, status, amount] = line.split(",");
        if (status === "paid") { out.push(line); total += Number(amount); count++; }
      }
      job.emit("progress", count);
      callback(null, out.length ? out.join("\n") + "\n" : "");
    },
    flush(callback) {
      job.emit("done", { paidOrders: count, total: Math.round(total * 100) / 100 });
      callback();
    },
  });
}

async function main() {
  const dir = fs.mkdtempSync(path.join(os.tmpdir(), "kodax-stream-"));
  const input = path.join(dir, "orders.csv");
  const output = path.join(dir, "paid.csv");

  // ملف "كبير" للتجربة
  const rows = ["id,status,amount"];
  for (let i = 1; i <= 20000; i++) rows.push(`${i},${i % 3 === 0 ? "pending" : "paid"},${(i % 50) + 0.5}`);
  fs.writeFileSync(input, rows.join("\n") + "\n");

  const before = process.memoryUsage().rss;
  await pipeline(fs.createReadStream(input, { highWaterMark: 16 * 1024 }), filterPaid(), fs.createWriteStream(output));
  const after = process.memoryUsage().rss;

  console.log("أسطر الناتج:", fs.readFileSync(output, "utf8").trim().split("\n").length - 1, "| آخر تقدم:", processed);
  console.log("زيادة الذاكرة تقريباً:", Math.round((after - before) / 1024 / 1024), "MB");
  fs.rmSync(dir, { recursive: true, force: true });
}

main().catch((err) => job.emit("error", err));