Node.js Daq Pub/Sub system mach

publish/subscribe code event chenmoHbogh 'ej code jangbogh pIm. publisher Sovbe' vay' Qoy, subscribers Sovbe' event nuqDaq. process law' vaD message broker (Redis, RabbitMQ) lo' 'ach pattern mach, in-process chenmoH laH. 'oH yaj vaj broker yaj Dun.

wa'DIch yIch

pub/sub wej Dochmey: topics (pong channel), subscribers (function topic Daq registered) 'ej publish operation (topic subscriber Hoch Qu').

class EventBus {
  #topics = new Map();

  subscribe(topic, handler) {
    if (!this.#topics.has(topic)) {
      this.#topics.set(topic, new Set());
    }
    this.#topics.get(topic).add(handler);

    // return an unsubscribe function — the caller owns cleanup
    return () => this.#topics.get(topic)?.delete(handler);
  }

  publish(topic, payload) {
    const handlers = this.#topics.get(topic);
    if (!handlers) return;
    for (const handler of handlers) {
      handler(payload);
    }
  }
}

in-memory event bus 20 line mach. Node EventEmitter rap 'ach: async handling, error isolation, wildcard topics chenmoH Dun.

lo'

const bus = new EventBus();

const unsubscribe = bus.subscribe('order.created', (order) => {
  console.log(`Sending confirmation email for order ${order.id}`);
});

bus.subscribe('order.created', (order) => {
  console.log(`Reserving inventory for order ${order.id}`);
});

bus.publish('order.created', { id: 42, total: 89.99 });
// Sending confirmation email for order 42
// Reserving inventory for order 42

unsubscribe(); // the email handler stops receiving events

cha' concern - email 'ej inventory - rap event Qu', wa' Sov pagh. wej concern (analytics) chel vaj order.created publish code touch pagh.

async subscribers

publish naive handlers synchronous Qu' 'ej promise Sovbe'. handler async Qu' (database ghItlh, API Qu') vaj fail Sov poQ 'ej slow handler pIm block pagh.

async publish(topic, payload) {
  const handlers = this.#topics.get(topic);
  if (!handlers) return [];

  const results = await Promise.allSettled(
    [...handlers].map((handler) => handler(payload)),
  );

  for (const result of results) {
    if (result.status === 'rejected') {
      console.error('Subscriber failed:', result.reason);
    }
  }

  return results;
}

Promise.allSettled handlers Hoch concurrent Qu', wa' succeed pagh fail. subscriber Qagh vaj pIm Qu', fail Sov.

wildcard 'ej namespace topics

real systems event ghom (order.*) subscribe. exact topic strings Daq subscribers store 'ach publish exact 'ej wildcard Hoch legh.

class EventBus {
  #topics = new Map();

  subscribe(pattern, handler) {
    if (!this.#topics.has(pattern)) this.#topics.set(pattern, new Set());
    this.#topics.get(pattern).add(handler);
    return () => this.#topics.get(pattern)?.delete(handler);
  }

  #matches(pattern, topic) {
    if (pattern === topic) return true;
    if (!pattern.endsWith('.*')) return false;
    const prefix = pattern.slice(0, -1); // keep the trailing dot
    return topic.startsWith(prefix);
  }

  async publish(topic, payload) {
    const matched = [...this.#topics.entries()]
      .filter(([pattern]) => this.#matches(pattern, topic))
      .flatMap(([, handlers]) => [...handlers]);

    return Promise.allSettled(matched.map((handler) => handler(payload)));
  }
}

bus.subscribe('order.*', logAnyOrderEvent) order.created, order.cancelled, order. namespace Hoch tlhap.

real broker poQ

in-process bus wa' application Daq modules decouple Dun. 'ach naDev poQ vaj Redis Pub/Sub, RabbitMQ, Kafka:

• process pagh service law' delivery. event bus wa' Node process neH. cha' app instance load balancer Daq subscribers rap pagh.
• Durability. publish DI' vay' subscribe pagh vaj event Qaw'. real broker message store, subscriber qen vaj tlhap.
• Backpressure 'ej ordering. Promise.allSettled concurrency nob, flow control pagh. slow consumer throttle pagh.
• At-least-once delivery ack. real brokers consumer ack, crash vaj redeliver.

in-process version Dun

"subscribers" Hoch wa' Node process Daq vaj (webhook handler, email sender, audit logger, wa' service) vaj EventBus mach Dun. Kafka cha' function rap codebase Daq Qagh pagh Qotlh. naDev bI'ang 'ej process boundaries delivery poQ vaj broker.