Construire un système pub/sub simple avec Node.js

Le publish/subscribe découple le code qui produit un événement de celui qui y réagit : l'éditeur ne sait pas et ne se soucie pas de qui écoute, et les abonnés ne savent pas et ne se soucient pas d'où vient l'événement. On fait appel à un broker de messages comme Redis ou RabbitMQ dès qu'il faut cela entre plusieurs processus, mais le patron lui-même est assez petit pour être construit de zéro en mémoire, et comprendre cette version rend les versions adossées à un broker bien moins mystérieuses.

L'idée de base

Un système pub/sub a trois éléments : des topics (canaux nommés), des abonnés (fonctions enregistrées sur un topic) et une opération de publication qui appelle chaque abonné actuellement enregistré sur un topic.

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);
    }
  }
}

Voilà un bus d'événements en mémoire fonctionnel en moins de 20 lignes. L'EventEmitter intégré de Node fait le même travail, mais il vaut la peine de construire le sien une fois, pour les éléments qu'un vrai broker ajoute : gestion asynchrone, isolation des erreurs et topics avec jokers.

Utilisation

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

Deux préoccupations indépendantes, l'envoi d'e-mails et l'inventaire, réagissent au même événement sans se connaître. Si vous ajoutez plus tard une troisième préoccupation (par exemple l'analytique), vous l'abonnez sans toucher au code qui publie order.created.

Gérer les abonnés asynchrones

Le publish naïf ci-dessus appelle les handlers de façon synchrone et ignore s'ils renvoient une promesse. Si un handler fait du travail asynchrone (écriture en base, appel d'API), on veut généralement savoir s'il a échoué, et on ne veut pas qu'un handler lent bloque les autres.

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 exécute tous les handlers en parallèle et laisse chacun réussir ou échouer indépendamment : un abonné défaillant n'empêche pas les autres de s'exécuter, et vous savez lequel a échoué, au lieu qu'un rejet non géré avale l'erreur.

Topics avec jokers et espaces de noms

Les systèmes réels veulent souvent s'abonner à une famille d'événements (order.*) plutôt qu'à un topic exact. Approche simple : stocker les abonnés sous des chaînes de topic exactes, mais vérifier à la fois les correspondances exactes et celles avec joker lors de la publication.

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)));
  }
}

Désormais, bus.subscribe('order.*', logAnyOrderEvent) capte order.created, order.cancelled et tout ce qui se trouve sous l'espace de noms order., en plus des handlers abonnés au topic exact.

Quand cela ne suffit plus : il faut un vrai broker

Ce bus en mémoire est réellement utile pour découpler des modules au sein d'une même application. Il atteint ses limites dès que vous avez besoin de l'un des éléments suivants, et c'est là que Redis Pub/Sub, RabbitMQ ou Kafka prennent le relais :

• Livraison entre processus ou services. Votre bus d'événements n'existe qu'à l'intérieur d'un seul processus Node. Deux instances de votre application derrière un répartiteur de charge ne partagent pas leurs abonnés.
• Durabilité. Si personne n'est abonné au moment de la publication, l'événement ci-dessus est tout simplement perdu. Un vrai broker (ou au moins une file) conserve les messages pour qu'un abonné démarré plus tard puisse encore les recevoir.
• Contre-pression et garanties d'ordre. Promise.allSettled donne de la concurrence, pas du contrôle de flux : un consommateur lent n'est pas ralenti, il prend simplement du retard pendant que les producteurs continuent d'émettre.
• Livraison « au moins une fois » avec accusé de réception. Les vrais brokers permettent à un consommateur d'acquitter explicitement un message et le redélivrent si le consommateur plante avant l'acquittement.

Quand la version en mémoire est le bon choix

Si tous vos « abonnés » se trouvent dans un seul processus Node (découpler un handler de webhook d'un expéditeur d'e-mails et d'un journal d'audit, le tout dans un seul service), un petit EventBus comme celui-ci représente souvent la bonne quantité d'infrastructure. Sortir Kafka pour relier deux fonctions de la même base de code, c'est résoudre un problème que vous n'avez pas encore. Commencez ici, et passez à un broker quand vous avez réellement besoin d'une livraison entre processus, pas avant.