Concurrency Pattern - Pub/Sub

concurrency-pattern-pubsub.js

/**
 * PubSub (Publish-Subscribe) Pattern Implementation
 *
 * A lightweight event bus that decouples producers (publishers) from
 * consumers (subscribers). Publishers emit named events with data;
 * subscribers register callbacks that fire when those events are published.
 */
class PubSub {
  #events = new Map();

  subscribe(eventName, callback) {
    // Only support function as callback
    if (typeof callback !== "function") {
      throw new TypeError(`Subscriber for ${eventName} must be a function`);
    }
    if (!this.#events.has(eventName)) {
      this.#events.set(eventName, new Set());
    }

    this.#events.get(eventName).add(callback); // Set ignores duplicate refs

    return () => {
      this.unsubscribe(eventName, callback);
    };
  }

  unsubscribe(eventName, callback) {
    this.#events.get(eventName)?.delete(callback);
  }

  // Used async/await to support async publishing of event
  async publish(eventName, data) {
    const subscribers = this.#events.get(eventName);
    if (!subscribers?.size) {
      return;
    }

    // Snapshot before iterating so mid-publish unsubscribes don't skew results
    const subscribersSnapshot = [...subscribers];

    const result = await Promise.allSettled(
      subscribersSnapshot.map((callback) => callback(data)),
    );

    for (let i = 0; i < result.length; i++) {
      if (result[i].status === "rejected") {
        console.error(
          `Subscriber ${i} threw an error for ${eventName}`,
          results[i].reason,
        );
      }
    }
  }

  clearEvent(eventName) {
    this.#events.delete(eventName);
  }

  clear() {
    this.#events.clear();
  }
}

const cart = {
  items: [],
  total: 0,
};
const product1 = {
  name: "Earphone",
  price: 2500,
};
const product2 = {
  name: "Keyboard",
  price: 1100,
};

const lockProductPrice = (product) => {
  console.log(`Price of ${product.name} is locked at ₹ ${product.price}`);
};

const updateCart = (product) => {
  const existingItem = cart.items.find((obj) => obj.name === product.name);
  if (existingItem) {
    existingItem.qty += 1;
  } else {
    cart.items.push({ ...product, qty: 1 });
  }
  cart.total = cart.items.reduce((acc, item) => {
    return acc + item.price * item.qty;
  }, 0);
  console.log(`Cart Total: ₹ ${cart.total}`);
};

const onNotification = (notification) => {
  console.log("New notification:", notification.message);
};

// Subscriber
const pubSub = new PubSub();

pubSub.subscribe("cart:add", lockProductPrice);
pubSub.subscribe("cart:add", updateCart);
const unsubscribeNotification = pubSub.subscribe(
  "notification:new",
  onNotification,
);

// Publisher
await pubSub.publish("cart:add", product1);
await pubSub.publish("notification:new", { message: "*** Item Added to Cart" });

await pubSub.publish("cart:add", product2);
await pubSub.publish("notification:new", { message: "*** Item Added to Cart" });

await pubSub.publish("cart:add", product1); // triggers qty increment for Earphone
await pubSub.publish("notification:new", { message: "*** Item Added to Cart" });

// Unsubscribe from Notification
unsubscribeNotification();
// Does not publish the event since we unsubscribed from notification
await pubSub.publish("notification:new", { message: "Use Supercoins" });