No commit message

creator/packages/ddp-server/README.md

@@ -0,0 +1,4 @@
+# ddp-server
+[Source code of released version](https://github.com/meteor/meteor/tree/master/packages/ddp-server) | [Source code of development version](https://github.com/meteor/meteor/tree/devel/packages/ddp-server)
+***
+

creator/packages/ddp-server/crossbar.js

@@ -0,0 +1,167 @@
+// A "crossbar" is a class that provides structured notification registration.
+// See _match for the definition of how a notification matches a trigger.
+// All notifications and triggers must have a string key named 'collection'.
+
+DDPServer._Crossbar = function (options) {
+  var self = this;
+  options = options || {};
+
+  self.nextId = 1;
+  // map from collection name (string) -> listener id -> object. each object has
+  // keys 'trigger', 'callback'.  As a hack, the empty string means "no
+  // collection".
+  self.listenersByCollection = {};
+  self.listenersByCollectionCount = {};
+  self.factPackage = options.factPackage || "livedata";
+  self.factName = options.factName || null;
+};
+
+_.extend(DDPServer._Crossbar.prototype, {
+  // msg is a trigger or a notification
+  _collectionForMessage: function (msg) {
+    var self = this;
+    if (! _.has(msg, 'collection')) {
+      return '';
+    } else if (typeof(msg.collection) === 'string') {
+      if (msg.collection === '')
+        throw Error("Message has empty collection!");
+      return msg.collection;
+    } else {
+      throw Error("Message has non-string collection!");
+    }
+  },
+
+  // Listen for notification that match 'trigger'. A notification
+  // matches if it has the key-value pairs in trigger as a
+  // subset. When a notification matches, call 'callback', passing
+  // the actual notification.
+  //
+  // Returns a listen handle, which is an object with a method
+  // stop(). Call stop() to stop listening.
+  //
+  // XXX It should be legal to call fire() from inside a listen()
+  // callback?
+  listen: function (trigger, callback) {
+    var self = this;
+    var id = self.nextId++;
+
+    var collection = self._collectionForMessage(trigger);
+    var record = {trigger: EJSON.clone(trigger), callback: callback};
+    if (! _.has(self.listenersByCollection, collection)) {
+      self.listenersByCollection[collection] = {};
+      self.listenersByCollectionCount[collection] = 0;
+    }
+    self.listenersByCollection[collection][id] = record;
+    self.listenersByCollectionCount[collection]++;
+
+    if (self.factName && Package['facts-base']) {
+      Package['facts-base'].Facts.incrementServerFact(
+        self.factPackage, self.factName, 1);
+    }
+
+    return {
+      stop: function () {
+        if (self.factName && Package['facts-base']) {
+          Package['facts-base'].Facts.incrementServerFact(
+            self.factPackage, self.factName, -1);
+        }
+        delete self.listenersByCollection[collection][id];
+        self.listenersByCollectionCount[collection]--;
+        if (self.listenersByCollectionCount[collection] === 0) {
+          delete self.listenersByCollection[collection];
+          delete self.listenersByCollectionCount[collection];
+        }
+      }
+    };
+  },
+
+  // Fire the provided 'notification' (an object whose attribute
+  // values are all JSON-compatibile) -- inform all matching listeners
+  // (registered with listen()).
+  //
+  // If fire() is called inside a write fence, then each of the
+  // listener callbacks will be called inside the write fence as well.
+  //
+  // The listeners may be invoked in parallel, rather than serially.
+  fire: function (notification) {
+    var self = this;
+
+    var collection = self._collectionForMessage(notification);
+
+    if (! _.has(self.listenersByCollection, collection)) {
+      return;
+    }
+
+    var listenersForCollection = self.listenersByCollection[collection];
+    var callbackIds = [];
+    _.each(listenersForCollection, function (l, id) {
+      if (self._matches(notification, l.trigger)) {
+        callbackIds.push(id);
+      }
+    });
+
+    // Listener callbacks can yield, so we need to first find all the ones that
+    // match in a single iteration over self.listenersByCollection (which can't
+    // be mutated during this iteration), and then invoke the matching
+    // callbacks, checking before each call to ensure they haven't stopped.
+    // Note that we don't have to check that
+    // self.listenersByCollection[collection] still === listenersForCollection,
+    // because the only way that stops being true is if listenersForCollection
+    // first gets reduced down to the empty object (and then never gets
+    // increased again).
+    _.each(callbackIds, function (id) {
+      if (_.has(listenersForCollection, id)) {
+        listenersForCollection[id].callback(notification);
+      }
+    });
+  },
+
+  // A notification matches a trigger if all keys that exist in both are equal.
+  //
+  // Examples:
+  //  N:{collection: "C"} matches T:{collection: "C"}
+  //    (a non-targeted write to a collection matches a
+  //     non-targeted query)
+  //  N:{collection: "C", id: "X"} matches T:{collection: "C"}
+  //    (a targeted write to a collection matches a non-targeted query)
+  //  N:{collection: "C"} matches T:{collection: "C", id: "X"}
+  //    (a non-targeted write to a collection matches a
+  //     targeted query)
+  //  N:{collection: "C", id: "X"} matches T:{collection: "C", id: "X"}
+  //    (a targeted write to a collection matches a targeted query targeted
+  //     at the same document)
+  //  N:{collection: "C", id: "X"} does not match T:{collection: "C", id: "Y"}
+  //    (a targeted write to a collection does not match a targeted query
+  //     targeted at a different document)
+  _matches: function (notification, trigger) {
+    // Most notifications that use the crossbar have a string `collection` and
+    // maybe an `id` that is a string or ObjectID. We're already dividing up
+    // triggers by collection, but let's fast-track "nope, different ID" (and
+    // avoid the overly generic EJSON.equals). This makes a noticeable
+    // performance difference; see https://github.com/meteor/meteor/pull/3697
+    if (typeof(notification.id) === 'string' &&
+        typeof(trigger.id) === 'string' &&
+        notification.id !== trigger.id) {
+      return false;
+    }
+    if (notification.id instanceof MongoID.ObjectID &&
+        trigger.id instanceof MongoID.ObjectID &&
+        ! notification.id.equals(trigger.id)) {
+      return false;
+    }
+
+    return _.all(trigger, function (triggerValue, key) {
+      return !_.has(notification, key) ||
+        EJSON.equals(triggerValue, notification[key]);
+    });
+  }
+});
+
+// The "invalidation crossbar" is a specific instance used by the DDP server to
+// implement write fence notifications. Listener callbacks on this crossbar
+// should call beginWrite on the current write fence before they return, if they
+// want to delay the write fence from firing (ie, the DDP method-data-updated
+// message from being sent).
+DDPServer._InvalidationCrossbar = new DDPServer._Crossbar({
+  factName: "invalidation-crossbar-listeners"
+});

creator/packages/ddp-server/crossbar_tests.js

@@ -0,0 +1,49 @@
+// White box tests of invalidation crossbar matching function.
+// Note: the current crossbar match function is designed specifically
+// to ensure that a modification that targets a specific ID does not
+// notify a query that is watching a different specific ID. (And to
+// keep separate collections separate.) Other than that, there's no
+// deep meaning to the matching function, and it could be changed later
+// as long as it preserves that property.
+Tinytest.add('livedata - crossbar', function (test) {
+  var crossbar = new DDPServer._Crossbar;
+  test.isTrue(crossbar._matches({collection: "C"},
+                                {collection: "C"}));
+  test.isTrue(crossbar._matches({collection: "C", id: "X"},
+                                {collection: "C"}));
+  test.isTrue(crossbar._matches({collection: "C"},
+                                {collection: "C", id: "X"}));
+  test.isTrue(crossbar._matches({collection: "C", id: "X"},
+                                {collection: "C"}));
+
+  test.isFalse(crossbar._matches({collection: "C", id: "X"},
+                                 {collection: "C", id: "Y"}));
+
+  // Test that stopped listens definitely don't fire.
+  var calledFirst = false;
+  var calledSecond = false;
+  var trigger = {collection: "C"};
+  var secondHandle;
+  crossbar.listen(trigger, function (notification) {
+    // This test assumes that listeners will be called in the order
+    // registered. It's not wrong for the crossbar to do something different,
+    // but the test won't be valid in that case, so make it fail so we notice.
+    calledFirst = true;
+    if (calledSecond) {
+      test.fail({
+        type: "test_assumption_failed",
+        message: "test assumed that listeners would be called in the order registered"
+      });
+    } else {
+      secondHandle.stop();
+    }
+  });
+  secondHandle = crossbar.listen(trigger, function (notification) {
+    // This should not get invoked, because it should be stopped by the other
+    // listener!
+    calledSecond = true;
+  });
+  crossbar.fire(trigger);
+  test.isTrue(calledFirst);
+  test.isFalse(calledSecond);
+});