Hooked redis into the core, cleaned up tests that depended on old ObservableCollection, made sure reload happens when connection to redis is resumed

Author: theodorDiaconu

lib/cache/ObservableCollection.js

@@ -1,11 +1,11 @@
 import { DiffSequence } from 'meteor/diff-sequence';
 import { _ } from 'meteor/underscore';
 import { LocalCollection, Minimongo } from 'meteor/minimongo';
-import cloneDeep from 'lodash.clonedeep';
 import fieldProjectionIsExclusion from './lib/fieldProjectionIsExclusion';
 import getChannels from './lib/getChannels';
 import extractFieldsFromFilters from './lib/extractFieldsFromFilters';
 import { MongoIDMap } from './mongoIdMap';
+import { EJSON } from 'meteor/ejson';
 import isRemovedNonExistent from '../utils/isRemovedNonExistent';
 import getStrategy from '../processors/getStrategy';
 
@@ -24,19 +24,20 @@ const { Matcher } = Minimongo;
 
 export default class ObservableCollection {
     /**
-     * @param observer
-     * @param cursor
-     * @param config
+     * Instantiate the collection
+     * @param {*} param
      */
-    constructor(multiplexer, matcher, sorter, cursorDescription) {
+    constructor({ multiplexer, matcher, sorter, cursorDescription }) {
         this.multiplexer = multiplexer;
+        this.matcher = matcher;
         this.cursorDescription = cursorDescription;
 
         this.collectionName = this.cursorDescription.collectionName;
         this.collection = Mongo.Collection.__getCollectionByName(
             cursorDescription.collectionName
         );
 
+        // Here we apply the logic of changing the cursor based on the collection-level configuration
         if (this.collection._redisOplog) {
             const { cursor } = this.collection._redisOplog;
             if (cursor) {
@@ -102,7 +103,6 @@ export default class ObservableCollection {
         }
 
         this.channels = getChannels(this.collectionName, this.options);
-        this.testDocEligibility = this._createTestDocEligibility();
         this.fieldsOfInterest = this._getFieldsOfInterest();
         this.__isInitialized = false;
 
@@ -128,8 +128,8 @@ export default class ObservableCollection {
      * @returns {*}
      */
     isEligible(doc) {
-        if (this.testDocEligibility) {
-            return this.testDocEligibility(doc);
+        if (this.matcher) {
+            return this.matcher.documentMatches(doc).result;
         }
 
         return true;
@@ -140,7 +140,7 @@ export default class ObservableCollection {
      * @returns {boolean}
      */
     isEligibleByDB(_id) {
-        if (this.testDocEligibility) {
+        if (this.matcher) {
             return !!this.collection.findOne(
                 _.extend({}, this.selector, { _id }),
                 { fields: { _id: 1 } }
@@ -191,7 +191,7 @@ export default class ObservableCollection {
      * @param safe {Boolean} If this is set to true, it assumes that the object is cleaned
      */
     add(doc, safe = false) {
-        doc = cloneDeep(doc);
+        doc = EJSON.clone(doc);
 
         if (!safe) {
             if (this.fieldsArray) {
@@ -257,7 +257,7 @@ export default class ObservableCollection {
         }
 
         let storedDoc = this.store.get(docId);
-        let oldDoc = cloneDeep(storedDoc);
+        let oldDoc = EJSON.clone(storedDoc);
 
         LocalCollection._modify(storedDoc, modifier);
 
@@ -303,42 +303,6 @@ export default class ObservableCollection {
         return false;
     }
 
-    /**
-     * Used at initialization
-     *
-     * Creates the function that checks if the document is valid
-     *
-     * @returns {null}
-     * @private
-     */
-    _createTestDocEligibility() {
-        const self = this;
-
-        if (_.keys(this.selector).length) {
-            try {
-                const matcher = new Matcher(this.selector);
-
-                return function(object) {
-                    return matcher.documentMatches(object).result;
-                };
-            } catch (e) {
-                // The logic here is that if our matcher is too complex for minimongo
-                // We put our matching function to query db
-                if (
-                    e.toString().indexOf('Unrecognized logical operator') >= 0
-                ) {
-                    return function(object) {
-                        return self.isEligibleByDB(object._id);
-                    };
-                } else {
-                    throw e;
-                }
-            }
-        }
-
-        return null;
-    }
-
     /**
      * Used at initialization
      *

lib/cache/testing/ObservableCollection.test.js

@@ -1,2 +1 @@
-import './ObservableCollection.test';
 import './filterFieldsForFetching.test';

lib/cache/testing/index.js

@@ -5,6 +5,7 @@ import RedisSubscriptionManager from './redis/RedisSubscriptionManager';
 import PubSubManager from './redis/PubSubManager';
 import { getRedisListener } from './redis/getRedisClient';
 import deepExtend from 'deep-extend';
+import reload from './processors/actions/reload';
 
 let isInitialized = false;
 
@@ -29,6 +30,11 @@ export default (config = {}) => {
         onConnect() {
             // this will be executed initially, but since there won't be any observable collections, nothing will happen
             // PublicationFactory.reloadAll();
+            RedisSubscriptionManager.getAllRedisSubscribers().forEach(
+                redisSubscriber => {
+                    reload(redisSubscriber.observableCollection);
+                }
+            );
         },
     });
 

lib/init.js

@@ -26,12 +26,12 @@ export default class RedisOplogObserveDriver {
         );
 
         // TODO send by object
-        this.observableCollection = new ObservableCollection(
+        this.observableCollection = new ObservableCollection({
             multiplexer,
             matcher,
             sorter,
-            cursorDescription
-        );
+            cursorDescription,
+        });
 
         if (strategy === Strategy.DEDICATED_CHANNELS) {
             let oc = this.observableCollection;

lib/mongo/RedisOplogObserveDriver.js

@@ -1,5 +1,3 @@
-import { diff } from 'deep-diff';
-import cloneDeep from 'lodash.clonedeep';
 import { DDP } from 'meteor/ddp';
 import isRemovedNonExistent from '../utils/isRemovedNonExistent';
 import { Mongo, MongoInternals } from 'meteor/mongo';

lib/mongo/extendObserveChanges.js

@@ -8,15 +8,8 @@ import getDedicatedChannel from '../../utils/getDedicatedChannel';
 import Config from '../../config';
 import OptimisticInvocation from '../OptimisticInvocation';
 
-const getWriteFence = function(optimistic) {
-    if (optimistic && DDPServer._CurrentWriteFence) {
-        return DDPServer._CurrentWriteFence.get();
-    }
-    return null;
-};
-
-const dispatchEvents = function(fence, collectionName, channels, events) {
-    if (fence) {
+const dispatchEvents = function(optimistic, collectionName, channels, events) {
+    if (optimistic) {
         OptimisticInvocation.withValue(true, () => {
             events.forEach(event => {
                 const docId = event[RedisPipe.DOC]._id;
@@ -58,37 +51,40 @@ const dispatchUpdate = function(
     docIds,
     fields
 ) {
-    const fence = getWriteFence(optimistic);
-    const uid = fence ? RedisSubscriptionManager.uid : null;
+    const uid = optimistic ? RedisSubscriptionManager.uid : null;
+
     const events = docIds.map(docId => ({
         [RedisPipe.EVENT]: Events.UPDATE,
         [RedisPipe.FIELDS]: fields,
         [RedisPipe.DOC]: { _id: docId },
         [RedisPipe.UID]: uid,
     }));
-    dispatchEvents(fence, collectionName, channels, events);
+
+    dispatchEvents(optimistic, collectionName, channels, events);
 };
 
 const dispatchRemove = function(optimistic, collectionName, channels, docIds) {
-    const fence = getWriteFence(optimistic);
-    const uid = fence ? RedisSubscriptionManager.uid : null;
+    const uid = optimistic ? RedisSubscriptionManager.uid : null;
+
     const events = docIds.map(docId => ({
         [RedisPipe.EVENT]: Events.REMOVE,
         [RedisPipe.DOC]: { _id: docId },
         [RedisPipe.UID]: uid,
     }));
-    dispatchEvents(fence, collectionName, channels, events);
+
+    dispatchEvents(optimistic, collectionName, channels, events);
 };
 
 const dispatchInsert = function(optimistic, collectionName, channels, docId) {
-    const fence = getWriteFence(optimistic);
-    const uid = fence ? RedisSubscriptionManager.uid : null;
+    const uid = optimistic ? RedisSubscriptionManager.uid : null;
+
     const event = {
         [RedisPipe.EVENT]: Events.INSERT,
         [RedisPipe.DOC]: { _id: docId },
         [RedisPipe.UID]: uid,
     };
-    dispatchEvents(fence, collectionName, channels, [event]);
+
+    dispatchEvents(optimistic, collectionName, channels, [event]);
 };
 
 export { dispatchInsert, dispatchUpdate, dispatchRemove };

lib/mongo/lib/dispatchers.js

@@ -12,6 +12,7 @@ export default function(cursorDescription, ordered, callbacks) {
             callbacks
         );
     }
+
     // You may not filter out _id when observing changes, because the id is a core
     // part of the observeChanges API.
 
@@ -64,11 +65,7 @@ export default function(cursorDescription, ordered, callbacks) {
                     // At a bare minimum, using the oplog requires us to have an oplog, to
                     // want unordered callbacks, and to not want a callback on the polls
                     // that won't happen.
-                    return (
-                        self._oplogHandle &&
-                        !ordered &&
-                        !callbacks._testOnlyPollCallback
-                    );
+                    return !ordered && !callbacks._testOnlyPollCallback;
                 },
                 function() {
                     // We need to be able to compile the selector. Fall back to polling for