بناء Event Bus و Task Scheduler متقدم في Pure JavaScript

عند بناء تطبيقات الويب المعقدة والمعتمدة على البيانات اللحظية (Real-time Applications)، نواجه تحديين رئيسيين على المستوى المعماري:

• تنسيق الاتصالات بين أجزاء التطبيق دون إنشاء ارتباط قوي بينها (Decoupled Communication).
• التحكم في عدد مهام الـ Async التي تعمل في الوقت نفسه لتجنب استهلاك موارد المتصفح أو الضغط غير الضروري على السيرفر.

في هذه المقالة، سنبني مكتبة صغيرة بدون أي اعتمادات خارجية باستخدام Pure JavaScript و ES Modules، وتحتوي على مكوّنين رئيسيين:

• Event Bus لإدارة الأحداث والاشتراكات مع دعم AbortSignal.
• Priority-based Task Scheduler للتحكم في الـ Concurrency وترتيب المهام حسب الأولوية.

المعمارية والمشاكل البرمجية الشائعة

قبل أن نكتب أي كود، من المهم أن نفهم المشاكل التي يحاول هذا التصميم حلها.

نظرة عامة على المعمارية

المشروع مقسم إلى مكوّنين مستقلين، بحيث يكون لكل مكوّن مسؤولية واضحة:

• EventEmitter: مسؤول عن تسجيل الـ Listeners وإرسال الأحداث وإلغاء الاشتراكات.
• TaskScheduler: مسؤول عن إدارة Queue للمهام، والتحكم في عدد المهام التي تعمل في الوقت نفسه، وترتيب المهام حسب الـ Priority.

هذا الفصل بين المسؤوليات يجعل كل مكوّن قابلًا للاستخدام بشكل مستقل داخل مشاريع مختلفة.

هيكل ملفات المشروع المعماري

تعرض الشجرة التالية التنظيم المعماري المتبع في المشروع:

الكود المصدري والمكونات الأساسية

1. الـ Event Bus المتطور مع دعم AbortSignal

تتيح هذه الفئة إمكانية إنشاء Event Bus بسيط يدعم الاشتراك في الأحداث، وإلغاء الاشتراكات يدويًا، بالإضافة إلى إلغاء الاشتراك تلقائيًا عند استخدام AbortSignal.

export class EventEmitter {
  #events = new Map();

  on(event, listener, options = {}) {
    if (typeof listener !== 'function') {
      throw new TypeError('Listener must be a function');
    }

    if (!this.#events.has(event)) {
      this.#events.set(event, new Set());
    }

    const listeners = this.#events.get(event);

    // Integrates AbortSignal to auto-clean listeners without manual off()
    if (options.signal) {
      if (options.signal.aborted) return () => {};
      options.signal.addEventListener(
        'abort',
        () => {
          this.off(event, listener);
        },
        { once: true },
      );
    }

    const wrappedListener = {
      fn: listener,
      once: !!options.once,
    };

    listeners.add(wrappedListener);
    return () => this.off(event, listener); // Higher-order cleanup function
  }

  emit(event, ...args) {
    if (!this.#events.has(event)) return false;

    const listeners = Array.from(this.#events.get(event));
    for (const item of listeners) {
      item.fn(...args);
      if (item.once) {
        this.off(event, item.fn);
      }
    }
    return true;
  }

  off(event, listener) {
    if (!this.#events.has(event)) return;

    const listeners = this.#events.get(event);
    for (const item of listeners) {
      if (item.fn === listener) {
        listeners.delete(item);
        break;
      }
    }

    if (listeners.size === 0) {
      this.#events.delete(event); 
    }
  }
}

2. منظم الـ Micro-tasks وحد التزامن

بعد بناء الـ Event Bus، نحتاج إلى مكوّن آخر مسؤول عن إدارة المهام غير المتزامنة.

الفكرة هنا بسيطة: بدلًا من تشغيل جميع المهام مباشرة، نضعها داخل Queue، ثم نحدد عدد المهام التي يمكن أن تعمل في الوقت نفسه من خلال قيمة concurrency.

بالإضافة إلى ذلك، يمكن إعطاء كل مهمة priority لتحديد ترتيب تنفيذها داخل الـ Queue.

export class TaskScheduler {
  #concurrency;
  #running = 0;
  #queue = [];

  constructor(concurrency = 3) {
    this.#concurrency = concurrency;
  }

  add(taskFn, { priority = 0 } = {}) {
    return new Promise((resolve, reject) => {
      this.#queue.push({ taskFn, priority, resolve, reject });

      // Sort tasks: highest priority jumps to the front of queue
      this.#queue.sort((a, b) => b.priority - a.priority); 

      this.#next();
    });
  }

  #next() {
    if (this.#running >= this.#concurrency || this.#queue.length === 0) {
      return;
    }

    const { taskFn, resolve, reject } = this.#queue.shift();
    this.#running++;

    // Schedules tasks directly into the engine microtask queue
    queueMicrotask(async () => {
      try {
        const result = await taskFn();
        resolve(result);
      } catch (error) {
        reject(error);
      } finally {
        this.#running--;
        this.#next();
      }
    });
  }
}

ملاحظة: استخدام queueMicrotask() هنا يحدد توقيت بدء تنفيذ الـ task داخل الـ Microtask Queue، بينما العملية غير المتزامنة نفسها قد تستمر عبر دورة الـ Event Loop، مثل fetch() أو setTimeout().

العرض التوضيحي وتجربة التشغيل

بعد الانتهاء من المكوّنين، يمكننا اختبار الـ Event Bus والـ Task Scheduler من خلال مثال بسيط.

import { EventEmitter, TaskScheduler } from '../index.js';

console.log('=== 1. Event Bus with AbortSignal Demo ===');
const bus = new EventEmitter();
const controller = new AbortController();

bus.on(
  'order:created',
  data => {
    console.log(`[Notification] Order #${data.id} placed for $${data.total}`);
  },
  { signal: controller.signal },
);

bus.emit('order:created', { id: 101, total: 250 });

// Cleanly detach event listener via AbortController
controller.abort();
bus.emit('order:created', { id: 102, total: 500 }); // Will NOT execute!

console.log('\n=== 2. Concurrency-Controlled Task Scheduler Demo ===');
const scheduler = new TaskScheduler(2); // Max 2 parallel requests

const fakeFetch = (id, delay) => () =>
  new Promise(resolve => {
    console.log(`[Start] Task ${id}`);
    setTimeout(() => {
      console.log(`[Done] Task ${id}`);
      resolve();
    }, delay);
  });

// Adding tasks with different priorities
scheduler.add(fakeFetch('Low-Priority 1', 1000), { priority: 1 });
scheduler.add(fakeFetch('Low-Priority 2', 800), { priority: 1 });
scheduler.add(fakeFetch('CRITICAL TASK', 500), { priority: 10 }); // Priority 10 jumps ahead!
scheduler.add(fakeFetch('Normal 1', 600), { priority: 5 });

في الجزء الأول، يتم إنشاء Listener مرتبط بـ AbortSignal. بمجرد استدعاء:

controller.abort();

يتم إلغاء الاشتراك، وبالتالي لن يتم تنفيذ الـ Listener عند إرسال الحدث مرة أخرى.

أما في الجزء الثاني، فقد حددنا:

const scheduler = new TaskScheduler(2);

وهذا يعني أن الحد الأقصى لعدد المهام التي يمكن تشغيلها في الوقت نفسه هو مهمتان.

وعند إضافة مهام ذات أولويات مختلفة، يتم ترتيب الـ Queue بحيث تحصل المهمة ذات الأولوية الأعلى على فرصة التنفيذ أولًا.

تثبيت المشروع وتجربة التشغيل

إذا كنت تريد تجربة المشروع من المستودع مباشرة، يمكنك تشغيله باستخدام Node.js:

git clone https://github.com/Hanafi6/js-advanced-bus-scheduler
cd js-advanced-bus-scheduler
node examples/demo.js

أما إذا كنت تنشئ المشروع من الصفر، فيمكنك تهيئة مشروع Node.js باستخدام:

npm init

ثم تأكد من تفعيل ES Modules داخل package.json:

{
  "type": "module"
}

بعد ذلك يمكنك تشغيل المثال:

node examples/demo.js

فهم الـ Concurrency والـ Priority

الـ Scheduler هنا يجمع بين مفهومين مختلفين:

Concurrency

الـ concurrency تحدد عدد المهام التي يمكن أن تكون قيد التنفيذ في الوقت نفسه.

على سبيل المثال:

const scheduler = new TaskScheduler(2);

يعني أن scheduler لن يسمح بأكثر من مهمتين قيد التنفيذ في الوقت نفسه.

Priority

الـ priority تحدد ترتيب المهام الموجودة في الـ Queue.

فمثلًا:

scheduler.add(taskA, { priority: 1 });
scheduler.add(taskB, { priority: 10 });
scheduler.add(taskC, { priority: 5 });

سيجعل taskB صاحبة الأولوية الأعلى داخل الـ Queue.

ملاحظات وحدود التصميم

رغم أن هذا التصميم صغير وبسيط، إلا أن استخدامه في نظام Production كبير يحتاج إلى التفكير في بعض الحالات الإضافية.

أحد أهم هذه الحالات هو Priority Starvation، حيث يمكن لمهام ذات الأولوية المنخفضة أن تنتظر لفترة طويلة إذا استمر وصول مهام جديدة ذات أولوية أعلى.

كذلك، زيادة قيمة الـ concurrency ليست دائمًا أفضل؛ لأن تشغيل عدد كبير من المهام في الوقت نفسه قد يزيد استهلاك الشبكة والـ CPU والذاكرة بدلًا من تحسين الأداء.

لذلك يجب اختيار قيمة الـ concurrency بناءً على طبيعة العمل، وليس باعتبارها رقمًا ثابتًا يصلح لكل التطبيقات.

الخلاصة

من خلال عدد قليل من الـ APIs في Pure JavaScript، استطعنا بناء مكوّنين يمكن استخدامهما كأساس لأنظمة أكثر تعقيدًا:

• Event Bus لفصل أجزاء التطبيق عن بعضها وتقليل الـ coupling.
• AbortSignal integration لإدارة دورة حياة الـ subscriptions بشكل أكثر أمانًا.
• Task Queue لتنظيم العمليات غير المتزامنة.
• Concurrency Control لمنع تشغيل عدد غير محدود من المهام في الوقت نفسه.
• Priority Scheduling لإعطاء المهام المهمة أولوية أعلى داخل الـ Queue.

الفكرة الأساسية ليست في كتابة أكبر قدر ممكن من الكود، وإنما في فهم كيف يمكن استخدام Event Loop و Microtasks و Promises و Queues لبناء abstraction صغيرة تحل مشكلة حقيقية في التطبيقات الحديثة.