Source: components/event-queue.js

  1. const Bluebird = require("bluebird");
  3. /**
  4.  * Handles the processing order of incoming Matrix events.
  5.  *
  6.  * Events can be pushed to the queue and will be processed when their
  7.  * corresponding data is ready and they are at the head of line.
  8.  * Different types of queues can be chosen for the processing order of events.
  9.  *
  10.  * Abstract Base Class. Use the factory method `create` to create new instances.
  11.  */
  12. class EventQueue {
  13.     /**
  14.      * Private constructor.
  15.      *
  16.      * @constructor
  17.      * @param {"none"|"single"|"per_room"} type The type of event queue to create.
  18.      * @param {consumeCallback} consumeFn Function which is called when an event
  19.      *     is consumed.
  20.      */
  21.     constructor(type, consumeFn) {
  22.         this.type = type;
  23.         this._queues = {
  24.             // $identifier: {
  25.             //  events: [ {dataReady: } ],
  26.             //  consuming: true|false
  27.             // }
  28.         };
  29.         this.consumeFn = consumeFn;
  30.     }
  32.     /**
  33.      * Push the event and its related data to the queue.
  34.      *
  35.      * @param {IMatrixEvent} event The event to enqueue.
  36.      * @param {Promise<object>} dataReady Promise containing data related to the event.
  37.      */
  38.     push(event, dataReady) {
  39.         const queue = this._getQueue(event);
  40.         queue.events.push({
  41.             dataReady: dataReady
  42.         });
  43.     }
  45.     _getQueue(event) {
  46.         const identifier = this.type === "per_room" ? event.room_id : "none";
  47.         if (!this._queues[identifier]) {
  48.             this._queues[identifier] = {
  49.                 events: [],
  50.                 consuming: false
  51.             };
  52.         }
  53.         return this._queues[identifier];
  54.     }
  56.     /**
  57.      * Starts consuming the queue.
  58.      *
  59.      * As long as events are enqueued they will continue to be consumed.
  60.      */
  61.     consume() {
  62.         Object.keys(this._queues).forEach((identifier) => {
  63.             if (!this._queues[identifier].consuming) {
  64.                 this._queues[identifier].consuming = true;
  65.                 this._takeNext(identifier);
  66.             }
  67.         });
  68.     }
  70.     _takeNext(identifier) {
  71.         const events = this._queues[identifier].events;
  72.         if (events.length === 0) {
  73.             this._queues[identifier].consuming = false;
  74.             return;
  75.         }
  76.         const entry = events.shift();
  78.         Bluebird.resolve(entry.dataReady).asCallback(this.consumeFn);
  79.         entry.dataReady.finally(() => this._takeNext(identifier));
  80.     }
  82.     /**
  83.      * Factory for EventQueues.
  84.      *
  85.      * @param {"none"|"single"|"per_room"} opts.type Type of the queue to create.
  86.      * @param {consumeCallback} consumeFn Function which is called when an event
  87.      *     is consumed.
  88.      * @return {EventQueue} The newly created EventQueue.
  89.      */
  90.     static create(opts, consumeFn) {
  91.         const type = opts.type;
  92.         /* eslint-disable no-use-before-define */
  93.         if (type == "single") {
  94.             return new EventQueueSingle(consumeFn);
  95.         }
  96.         if (type == "per_room") {
  97.             return new EventQueuePerRoom(consumeFn);
  98.         }
  99.         if (type == "none") {
  100.             return new EventQueueNone(consumeFn);
  101.         }
  102.         /* eslint-enable no-use-before-define */
  103.         throw Error(`Invalid EventQueue type '${type}'.`);
  104.     }
  105. }
  107. /**
  108.  * EventQueue for which all events are enqueued in their order of arrival.
  109.  *
  110.  * The foremost event is processed as soon as its data is available.
  111.  */
  112. class EventQueueSingle extends EventQueue {
  113.     constructor(consumeFn) {
  114.         super("single", consumeFn);
  115.     }
  116. }
  118. /**
  119.  * EventQueue for which one queue per room is utilized.
  120.  *
  121.  * Events at the head of line are processed as soon as their data is available.
  122.  */
  123. class EventQueuePerRoom extends EventQueue {
  124.     constructor(consumeFn) {
  125.         super("per_room", consumeFn);
  126.     }
  127. }
  129. /**
  130.  * Dummy EventQueue for which no queue is utilized.
  131.  *
  132.  * Every event is handled as soon as its data is available.
  133.  */
  134. class EventQueueNone extends EventQueue {
  135.     constructor(consumeFn) {
  136.         super("none", consumeFn);
  137.     }
  139.     push(event, dataReady) {
  140.         // consume the event instantly
  141.         Bluebird.resolve(dataReady).asCallback(this.consumeFn);
  142.     }
  144.     consume() {
  145.         // no-op for EventQueueNone
  146.     }
  147. }
  149. /**
  150.  * @callback consumeCallback
  151.  * @param {Error} [err] The error in case the data could not be retrieved.
  152.  * @param {object} data The data associated with the consumed event.
  153.  */
  155. module.exports = {
  156.     EventQueue,
  157.     EventQueueSingle,
  158.     EventQueuePerRoom,
  159.     EventQueueNone,
  160. };