Merge pull request #321 from share/delay-op-for-query-poll

Delay ops sent by query subscriptions until after polling is complete

Author: nateps

lib/query-emitter.js

@@ -63,7 +63,12 @@ QueryEmitter.prototype._emitTiming = function(action, start) {
 };
 
 QueryEmitter.prototype._update = function(op) {
+  // Note that `op` should not be projected or sanitized yet. It's possible for
+  // a query to filter on a field that's not in the projection. skipPoll checks
+  // to see if an op could possibly affect a query, so it should get passed the
+  // full op. The onOp listener function must call backend.sanitizeOp()
   var id = op.d;
+  var pollCallback = this._defaultCallback;
 
   // Check if the op's id matches the query before updating the query results
   // and send it through immediately if it does. The current snapshot
@@ -90,28 +95,35 @@ QueryEmitter.prototype._update = function(op) {
   // that removed the doc from the query to cause the client-side computed
   // list to update.
   if (this.ids.indexOf(id) !== -1) {
-    // Note that this op is not projected or sanitized yet. It's possible for a
-    // query to filter on a field that's not in the projection. skipPoll checks
-    // to see if an op could possibly affect a query, so it should get passed
-    // the full op. The onOp listener function must call backend.sanitizeOp()
-    this.onOp(op);
+    var emitter = this;
+    pollCallback = function(err) {
+      // Send op regardless of polling error. Clients handle subscription to ops
+      // on the documents that currently match query results independently from
+      // updating which docs match the query
+      emitter.onOp(op);
+      if (err) emitter.onError(err);
+    };
   }
 
   // Ignore if the database or user function says we don't need to poll
   try {
-    if (this.db.skipPoll(this.collection, id, op, this.query)) return this._defaultCallback();
-    if (this.skipPoll(this.collection, id, op, this.query)) return this._defaultCallback();
+    if (
+      this.db.skipPoll(this.collection, id, op, this.query) ||
+      this.skipPoll(this.collection, id, op, this.query)
+    ) {
+      return pollCallback();
+    }
   } catch (err) {
-    return this._defaultCallback(err);
+    return pollCallback(err);
   }
   if (this.canPollDoc) {
     // We can query against only the document that was modified to see if the
     // op has changed whether or not it matches the results
-    this.queryPollDoc(id, this._defaultCallback);
+    this.queryPollDoc(id, pollCallback);
   } else {
     // We need to do a full poll of the query, because the query uses limits,
     // sorts, or something special
-    this.queryPoll(this._defaultCallback);
+    this.queryPoll(pollCallback);
   }
 };
 

test/client/query-subscribe.js

@@ -83,6 +83,40 @@ module.exports = function(options) {
       });
     });
 
+    it('subscribed query removes document from results before sending delete op to other clients', function(done) {
+      var connection1 = this.backend.connect();
+      var connection2 = this.backend.connect();
+      var matchAllDbQuery = this.matchAllDbQuery;
+      async.parallel([
+        function(cb) {
+          connection1.get('dogs', 'fido').create({age: 3}, cb);
+        },
+        function(cb) {
+          connection1.get('dogs', 'spot').create({age: 5}, cb);
+        }
+      ], function(err) {
+        if (err) return done(err);
+        var query = connection2.createSubscribeQuery('dogs', matchAllDbQuery, null, function(err) {
+          if (err) return done(err);
+          connection1.get('dogs', 'fido').del();
+        });
+        var removed = false;
+        connection2.get('dogs', 'fido').on('del', function() {
+          expect(removed).equal(true);
+          done();
+        });
+        query.on('remove', function(docs, index) {
+          removed = true;
+          expect(util.pluck(docs, 'id')).eql(['fido']);
+          expect(util.pluck(docs, 'data')).eql([{age: 3}]);
+          expect(index).a('number');
+          var results = util.sortById(query.results);
+          expect(util.pluck(results, 'id')).eql(['spot']);
+          expect(util.pluck(results, 'data')).eql([{age: 5}]);
+        });
+      });
+    });
+
     it('subscribed query does not get updated after destroyed', function(done) {
       var connection = this.backend.connect();
       var connection2 = this.backend.connect();