bigbluebutton-Github/bigbluebutton-html5/imports/ui/services/LocalCollectionSynchronizer/LocalCollectionSynchronizer.js

import SubscriptionRegistry from '/imports/ui/services/subscription-registry/subscriptionRegistry';
import CollectionEventsBroker from '/imports/ui/services/LiveDataEventBroker/LiveDataEventBroker';

/*
This class connects a local collection with the LiveDataEventBroker, propagating the changes of a server-side published cursor to a local collection. 

It also guarantee that in case of a reconnection or a re-subscription, the data is only removed after subscription is ready, avoiding the situation of missing data during re-synchronization.
*/

class LocalCollectionSynchronizer {
  constructor(serverCollection, localCollection, options = {}) {
    this.serverCollection = serverCollection;
    this.localCollection = localCollection;

    this.lastSubscriptionId = '';
    this.options = options;
    this.ignoreDeletes = false;
  }

  /*
  This method allow to enable/disable the ignoreDeletes feature.
  When enabled, system will skip the received deletes ( not apply on local collection ).
  Don't panic: these deletes will be processed when the subscription gets ready - see removeOldSubscriptionData method.
  */
  setIgnoreDeletes(value) {
    this.ignoreDeletes = value;
  }

  // Replicates remote collection to local collection ( avoiding cleanup during the forced resync )
  setupListeners() {
    const self = this;

    const addedCallback = function (item) {
      const subscription = SubscriptionRegistry
        .getSubscription(self.serverCollection._name);
      if (item.id === 'publication-stop-marker' && item.stopped) {
        self.ignoreDeletes = true;
        return;
      }
      // If the subscriptionId changes means the subscriptions was redone
      // or theres more than one subscription per collection
      if (subscription && (self.lastSubscriptionId !== subscription.subscriptionId)) {
        const wasEmpty = self.lastSubscriptionId === '';
        self.lastSubscriptionId = subscription.subscriptionId;
        if (!wasEmpty) {
          self.callWhenSubscriptionReady(() => {
            self.ignoreDeletes = false;
            Session.set('globalIgnoreDeletes', false);
            self.removeOldSubscriptionData();
          });
        }
      }

      const selector = { referenceId: item.referenceId };
      const itemExistInCollection = self.localCollection.findOne(selector);

      const { _id, ...restItem } = item;
      const payload = {
        subscriptionId: self.lastSubscriptionId,
        ...restItem,
      };

      if (itemExistInCollection) {
        self.localCollection.update(selector, payload);
      } else {
        self.localCollection.insert(payload);
      }
    };

    const changedCallback = function (item) {
      const {
        _id,
        ...restFields
      } = item;
      const selector = { referenceId: item.referenceId };
      const payload = {
        $set: {
          ...restFields,
          subscriptionId: self.lastSubscriptionId,
        },
      };
      self.localCollection.update(selector, payload);
    };

    const removedCallback = function (item) {
      const globalIgnoreDeletes = Session.get('globalIgnoreDeletes');
      if (self.ignoreDeletes || globalIgnoreDeletes) {
        return;
      }
      const selector = { referenceId: item.referenceId };

      self.localCollection.remove(selector);
    };

    CollectionEventsBroker.addListener(this.serverCollection._name, 'added', addedCallback);
    CollectionEventsBroker.addListener(this.serverCollection._name, 'changed', changedCallback);
    CollectionEventsBroker.addListener(this.serverCollection._name, 'removed', removedCallback);
  }

  /*
  This method calls the function received as parameter when the subscription gets ready.
*/
  callWhenSubscriptionReady(func) {
    const temp = (res) => {
      setTimeout(() => {
        const subscription = SubscriptionRegistry.getSubscription(this.serverCollection._name);
        if (subscription.ready()) {
          func();
          return res();
        }
        return temp(res);
      }, 100);
    };
    const tempPromise = new Promise((res) => {
      temp(res);
    });
    return tempPromise;
  }

  /*
  This method removes data from previous subscriptions after the current one is ready.
  */
  removeOldSubscriptionData() {
    const subscription = SubscriptionRegistry.getSubscription(this.serverCollection._name);

    const selector = {
      subscriptionId: {
        $nin: [subscription.subscriptionId],
      },
    };

    const oldSubscriptionItems = this.localCollection.find(selector).fetch();
    oldSubscriptionItems.forEach((item) => {
      this.localCollection.remove({
        meetingId: item.meetingId,
        ...selector,
      });
    });
  }
}

export default LocalCollectionSynchronizer;