End pending method when connection drops

Author: thani-sh

lib/ddp-client.js

@@ -48,9 +48,16 @@ var DDPClient = function(opts) {
   self._nextId = 0;
   self._callbacks = {};
   self._updatedCallbacks = {};
+  self._pendingMethods = {};
   self._observers = {};
 };
 
+
+DDPClient.ERRORS = {
+  DISCONNECTED: new Error("DDPClient: Disconnected from DDP server")
+};
+
+
 /**
  * Inherits from EventEmitter
  */
@@ -80,6 +87,7 @@ DDPClient.prototype._prepareHandlers = function() {
 
   self.socket.on("close", function(event) {
     self.emit("socket-close", event.code, event.reason);
+    self._endPendingMethodCalls();
     self._recoverNetworkError();
   });
 
@@ -320,6 +328,24 @@ DDPClient.prototype.connect = function(connected) {
   }
 };
 
+DDPClient.prototype._endPendingMethodCalls = function() {
+  var self = this;
+  var ids = _.keys(self._pendingMethods);
+  self._pendingMethods = {};
+
+  ids.forEach(function (id) {
+    if(self._callbacks[id]) {
+      self._callbacks[id](DDPClient.ERRORS.DISCONNECTED);
+      delete self._callbacks[id];
+    }
+
+    if(self._updatedCallbacks[id]) {
+      self._updatedCallbacks[id]();
+      delete self._updatedCallbacks[id];
+    }
+  });
+};
+
 DDPClient.prototype._makeSockJSConnection = function() {
   var self = this;
 
@@ -389,13 +415,23 @@ DDPClient.prototype.call = function(name, params, callback, updatedCallback) {
   var self = this;
   var id = self._getNextId();
 
-  if (callback) {
-    self._callbacks[id] = callback;
-  }
+  self._callbacks[id] = function () {
+    delete self._pendingMethods[id];
 
-  if (updatedCallback) {
-    self._updatedCallbacks[id] = updatedCallback;
-  }
+    if (callback) {
+      callback.apply(this, arguments);
+    }
+  };
+
+  self._updatedCallbacks[id] = function () {
+    delete self._pendingMethods[id];
+
+    if (updatedCallback) {
+      updatedCallback.apply(this, arguments);
+    }
+  };
+
+  self._pendingMethods[id] = true;
 
   self._send({
     msg    : "method",

test/ddp-client.js

@@ -134,6 +134,68 @@ describe('Automatic reconnection', function() {
       done();
     }, 15);
   });
+
+  it('should save currently running method calls', function() {
+    var ddpclient = new DDPClient();
+    ddpclient._getNextId = sinon.stub().returns('_test');
+    ddpclient._send = Function.prototype;
+
+    ddpclient.connect();
+    ddpclient.call();
+
+    assert("_test" in ddpclient._pendingMethods)
+  });
+
+  it('should remove id when callback is called', function() {
+    var ddpclient = new DDPClient();
+    ddpclient._getNextId = sinon.stub().returns('_test');
+    ddpclient._send = Function.prototype;
+
+    ddpclient.connect();
+    ddpclient.call();
+
+    assert("_test" in ddpclient._pendingMethods)
+
+    ddpclient._callbacks._test();
+    assert(!("_test" in ddpclient._pendingMethods))
+  });
+
+  it('should remove id when updated-callback is called', function() {
+    var ddpclient = new DDPClient();
+    ddpclient._getNextId = sinon.stub().returns('_test');
+    ddpclient._send = Function.prototype;
+
+    ddpclient.connect();
+    ddpclient.call();
+
+    assert("_test" in ddpclient._pendingMethods)
+
+    ddpclient._updatedCallbacks._test();
+    assert(!("_test" in ddpclient._pendingMethods))
+  });
+
+  it('should end method calls which could not be completed', function() {
+    var ddpclient = new DDPClient();
+    var callback = sinon.spy();
+    var updatedCallback = sinon.spy();
+
+    ddpclient._pendingMethods = { _test: true };
+    ddpclient._callbacks = { _test: callback };
+    ddpclient._updatedCallbacks = { _test: updatedCallback };
+
+    ddpclient.connect();
+    ddpclient.socket.emit('close', {});
+
+    assert(callback.calledOnce);
+    assert(callback.calledWithExactly(DDPClient.ERRORS.DISCONNECTED));
+
+    assert(updatedCallback.calledOnce);
+
+    // callbacks should be removed after calling them
+    assert(!("_test" in ddpclient._callbacks));
+    assert(!("_test" in ddpclient._updatedCallbacks));
+    assert(!("_test" in ddpclient._pendingMethods));
+  });
 });
 
 
@@ -408,4 +470,4 @@ function WithRequestGet(getFn, fn) {
   fn();
 
   request.get = originalGet;
-}
+}