Beginning a system for merging subscriptions in the session.

So far just one v. small test, and enough of a shim to get it to pass.

packages/livedata/livedata_server.js

@@ -1,5 +1,63 @@
 var Fiber = __meteor_bootstrap__.require('fibers');
 
+// This file contains classes:
+// * LivedataSession - The server's connection to a single DDP client
+// * LivedataSubscription - A single subscription for a single client
+// * LivedataServer - An entire server that may talk to > 1 client.  A DDP endpoint.
+
+
+Meteor._SessionDocumentView = function (id) {
+  var self = this;
+  self.id = id;
+  self.existsIn = {}; // set of subId
+  self.dataByKey = {}; // key-> [ {subId -> value} by precedence]
+};
+
+Meteor._SessionCollectionView = function (collectionName, sessionCallbacks) {
+  var self = this;
+  self.collectionName = collectionName;
+  self.documents = {};
+  self.callbacks = sessionCallbacks;
+};
+
+_.extend(Meteor._SessionCollectionView.prototype, {
+  added: function (subscriptionId, doc) {
+    var self = this;
+    var docView = self.documents[doc._id];
+    if (docView) {
+      if (_.has(docView.existsIn, subscriptionId)) {
+        throw new Error("Duplicate add for " + doc._id);
+      }
+      docView.existsIn[subscriptionId] = true;
+      //XXX: Deal with fields here.
+    } else {
+      docView = new Meteor._SessionDocumentView(doc._id);
+      self.documents[doc._id] = docView;
+      docView.existsIn[subscriptionId] = true;
+      //XXX: Deal with fields here
+      self.callbacks.added(self.collectionName, doc);
+    }
+  },
+  changed: function (subscriptionId, id, changed, cleared) {
+    var self = this;
+  },
+  removed: function (subscriptionId, ids) {
+    var self = this;
+    _.each(ids, function (id) {
+      var docView = self.documents[id];
+      if (!docView) {
+        throw new Error("Removed nonexistent document " + id);
+      }
+      delete docView.existsIn[subscriptionId];
+      if (_.isEmpty(docView.existsIn)) {
+        //XXX: Stop splitting it up
+        self.callbacks.removed(self.collectionName, [id]);
+      } else {
+        //XXX: DO STUFF
+      }
+    });
+  }
+});
 /******************************************************************************/
 /* LivedataSession                                                            */
 /******************************************************************************/
@@ -657,6 +715,7 @@ Meteor._LivedataServer = function () {
   self.stream_server = new Meteor._StreamServer;
 
   self.stream_server.register(function (socket) {
+    // socket implements the SockJSConnection interface
     socket.meteor_session = null;
 
     var sendError = function (reason, offending_message) {

packages/livedata/package.js

@@ -34,4 +34,5 @@ Package.on_test(function (api) {
   api.add_files('livedata_connection_tests.js', ['client']);
   api.add_files('livedata_tests.js', ['client', 'server']);
   api.add_files('livedata_test_service.js', ['client', 'server']);
+  api.add_files('session_view_tests.js', ['server']);
 });

packages/livedata/session_view_tests.js

@@ -0,0 +1,46 @@
+var newView = function(test) {
+  var results = [];
+  var view = new Meteor._SessionCollectionView('test', {
+    added: function (collection, doc) {
+      results.push({fun: 'added', doc:doc});
+    },
+    changed: function (collection, id, changed, cleared) {
+      results.push({fun: 'changed', id: id, changed: changed, cleared: cleared});
+    },
+    removed: function (collection, ids) {
+      results.push({fun: 'removed', ids: ids});
+    }
+  });
+  var ret = {
+    view: view,
+    results: results
+  };
+  _.each(["added", "changed", "removed"], function (it) {
+    ret[it] = _.bind(view[it], view);
+  });
+  ret.expectResult = function (result) {
+    test.equal(results.shift(), result);
+  };
+  ret.expectNoResult = function () {
+    test.equal(results, []);
+  };
+  return ret;
+};
+
+Tinytest.add('livedata - sessionview - basic', function (test) {
+  var v = newView(test);
+
+  v.added("A", {_id: "A1"});
+  v.expectResult({fun: 'added', doc: {_id: "A1"}});
+  v.expectNoResult();
+
+  v.added("B", {_id: "A1"});
+  v.expectNoResult();
+
+  v.removed("A", ["A1"]);
+  v.expectNoResult();
+
+  v.removed("B", ["A1"]);
+  v.expectResult({fun: 'removed', ids: ["A1"]});
+  v.expectNoResult();
+});

packages/mongo-livedata/mongo_driver.js

@@ -286,8 +286,10 @@ _.each(['forEach', 'map', 'rewind', 'fetch', 'count'], function (method) {
   };
 });
 
-// Called by livedata_server to automatically publish cursors returned from a
-// publish handler over DDP.
+// When you call Meteor.publish() with a function that returns a Cursor, we need
+// to transmute it into the equivalent subscription.  This is the function that
+// does that.
+
 Cursor.prototype._publishCursor = function (sub) {
   var self = this;
   var collection = self._cursorDescription.collectionName;