More fixes for DDP rewrite. 'livedata' tests pass.

packages/livedata/livedata_server.js

@@ -43,6 +43,7 @@ _.extend(Meteor._SessionDocumentView.prototype, {
     _.each(self.dataByKey, function (precedenceList, key) {
       ret[key] = precedenceList[0].value;
     });
+    return ret;
   },
 
   clearField: function (subscriptionId, key, changeCollector, clearCollector) {
@@ -144,7 +145,8 @@ _.extend(Meteor._SessionCollectionView.prototype, {
     var cleared = [];
     diffObjects(prevDV.getFields(), nowDV.getFields(), {
       both: function (key, prev, now) {
-        fields[key] = now;
+        if (!_.isEqual(prev, now))
+          fields[key] = now;
       },
       rightOnly: function (key, now) {
         fields[key] = now;
@@ -311,7 +313,7 @@ _.extend(Meteor._LivedataSession.prototype, {
   sendRemoved: function (collectionName, ids) {
     var self = this;
     if (self._isSending)
-      self.send({msg: "removed", ids: ids});
+      self.send({msg: "removed", collection: collectionName, ids: ids});
   },
 
   getSendCallbacks: function () {

packages/livedata/livedata_tests.js

@@ -295,51 +295,58 @@ if (Meteor.isClient) {
       // A helper for testing incoming set and unset messages
       // XXX should this be extracted as a general helper together with
       // eavesdropOnCollection?
-      var testSetAndUnset = function(expectation) {
-        test.equal(_.map(messages, function(msg) {
-          var result = {};
-          if (msg.set)
-            result.set = msg.set.name;
-          if (msg.unset)
-            result.unset = true;
-          return result;
-        }), expectation);
+      var expectMessages = function(expectedAddedMessageCount,
+                                    expectedRemovedMessageCount,
+                                    expectedNamesInCollection) {
+        var actualAddedMessageCount = 0;
+        var actualRemovedMessageCount = 0;
+        _.each(messages, function (msg) {
+          if (msg.msg === 'added')
+            ++actualAddedMessageCount;
+          else if (msg.msg === 'removed')
+            actualRemovedMessageCount += msg.ids.length;
+          else
+            test.fail({unexpected: JSON.stringify(msg)});
+        });
+        test.equal(actualAddedMessageCount, expectedAddedMessageCount);
+        test.equal(actualRemovedMessageCount, expectedRemovedMessageCount);
+        expectedNamesInCollection.sort();
+        test.equal(_.pluck(objectsWithUsers.find({}, {sort: ['name']}).fetch(),
+                           'name'),
+                   expectedNamesInCollection);
         messages.length = 0; // clear messages without creating a new object
       };
 
       Meteor.subscribe("objectsWithUsers", expect(function() {
-        testSetAndUnset([{set: "owned by none"}]);
-        test.equal(objectsWithUsers.find().count(), 1);
-
+        expectMessages(1, 0, ["owned by none"]);
         Meteor.apply("setUserId", [1], {wait: true}, afterFirstSetUserId);
       }));
 
       var afterFirstSetUserId = expect(function() {
-        testSetAndUnset([
-          {unset: true},
-          {set: "owned by one - a"},
-          {set: "owned by one/two - a"},
-          {set: "owned by one/two - b"}]);
-        test.equal(objectsWithUsers.find().count(), 3);
-
+        expectMessages(3, 1, [
+          "owned by one - a",
+          "owned by one/two - a",
+          "owned by one/two - b"]);
         Meteor.apply("setUserId", [2], {wait: true}, afterSecondSetUserId);
       });
 
       var afterSecondSetUserId = expect(function() {
-        testSetAndUnset([
-          {unset: true},
-          {set: "owned by two - a"},
-          {set: "owned by two - b"}]);
-        test.equal(objectsWithUsers.find().count(), 4);
-
+        expectMessages(2, 1, [
+          "owned by one/two - a",
+          "owned by one/two - b",
+          "owned by two - a",
+          "owned by two - b"]);
         Meteor.apply("setUserId", [2], {wait: true}, afterThirdSetUserId);
       });
 
       var afterThirdSetUserId = expect(function() {
         // Nothing should have been sent since the results of the
         // query are the same ("don't flap data on the wire")
-        testSetAndUnset([]);
-        test.equal(objectsWithUsers.find().count(), 4);
+        expectMessages(0, 0, [
+          "owned by one/two - a",
+          "owned by one/two - b",
+          "owned by two - a",
+          "owned by two - b"]);
         undoEavesdrop();
       });
     }, function(test, expect) {

packages/mongo-livedata/collection.js

@@ -68,6 +68,17 @@ Meteor.Collection = function (name, options) {
       // Apply an update.
       // XXX better specify this interface (not in terms of a wire message)?
       update: function (msg) {
+        // Handle removed separately, since it's the only one without an 'id'
+        // field.
+        if (msg.msg === 'removed') {
+          _.each(msg.ids, function (removedId) {
+            if (!self._collection.findOne(removedId))
+              throw new Error("Expected to find a document to remove");
+            self._collection.remove(removedId);
+          });
+          return;
+        }
+
         var doc = self._collection.findOne(msg.id);
 
         // Is this a "replace the whole doc" message coming from the quiescence
@@ -89,12 +100,6 @@ Meteor.Collection = function (name, options) {
           if (doc)
             throw new Error("Expected not to find a document already present for an add");
           self._collection.insert(_.extend({_id: msg.id}, msg.fields));
-        } else if (msg.msg === 'removed') {
-          if (!doc)
-            throw new Error("Expected to find a document to remove");
-          _.each(msg.ids, function (removedId) {
-            self._collection.remove(removedId);
-          });
         } else if (msg.msg === 'changed') {
           if (!doc)
             throw new Error("Expected to find a document to change");