QueuePubSub

by Paweł Niemczyk

JavaScript 1.7

class Subscriber {
  constructor(name, onBroadcast) {
    this.name = name;
    this.kind = "subscriber";
    if (typeof onBroadcast !== "function") {
      throw TypeError("onBroadcast must be a function");
    }

    this.onBroadcast = onBroadcast;
  }
}

class QueuePubSub {
  constructor(queue, listeners) {
    this.queue = queue || [];
    this.listeners = listeners || [];
  }

  static subscriber(name, onBroadcast) {
    return new Subscriber(name, onBroadcast);
  }

  subscriber(name, onBroadcast) {
    return new Subscriber(name, onBroadcast);
  }

  clearQueue() {
    this.queue = [];
  }

  broadcast(destination, payload) {
    if (!destination) {
      throw TypeError("Destination must be provided");
    }

    const event = { destination, payload };

    let pushToQueue = true;

    this.listeners.forEach(listener => {
      if (destination === listener.name) {
        listener.onBroadcast(event);
        pushToQueue = false;
      } else if (destination === "all") {
        listener.onBroadcast(event);
      }
    });

    if (pushToQueue) this.queue.push(event);
  }

  broadcastQueue(listener) {
    const published = [];
    this.queue.forEach(event => {
      if (event.destination === "all") {
        listener.onBroadcast(event);
      } else if (event.destination === listener.name) {
        listener.onBroadcast(event);
        published.push(event);
      }
    });
    published.forEach(event => {
      this.queue = this.queue.filter(curEvent => event !== curEvent);
    });
  }

  subscribe(listener) {
    if (listener && listener.kind === "subscriber") {
      if (typeof listener.onBroadcast !== "function") {
        throw TypeError("onBroadcast must be a function");
      }

      if (!listener.name) {
        throw TypeError("Subscriber must have a name");
      }
    }

    if (this.listeners.indexOf(listener) === -1) {
      this.listeners.push(listener);
      if (this.queue.length) {
        this.broadcastQueue(listener);
      }
    }
 ...