Prevent pending request leak when disconnect the SDK

doc/6/core-classes/kuzzle/disconnect/index.md

@@ -9,7 +9,9 @@ description: Disconnect the SDK
 
 Closes the current connection to Kuzzle.
 The SDK then enters the `offline` state.
-A call to `disconnect()` will not trigger a `disconnected` event. This event is only triggered on unexpected disconnection.
+A call to `disconnect()` will not trigger a `disconnected` event. This event is only triggered on unexpected disconnection.  
+
+If there are still pending requests during the `disconnect` call, a `discarded` event will be issued for each of them.  
 
 ## Arguments
 

src/protocols/abstract/common.js

@@ -5,20 +5,15 @@ const
   uuidv4 = require('../../uuidv4'),
   KuzzleEventEmitter = require('../../eventEmitter');
 
-// read-only properties
-let
-  _host,
-  _port,
-  _ssl;
-
 class AbstractWrapper extends KuzzleEventEmitter {
 
   constructor (host, options = {}) {
     super();
 
-    _host = host;
-    _port = typeof options.port === 'number' ? options.port : 7512;
-    _ssl = typeof options.sslConnection === 'boolean' ? options.sslConnection : false;
+    this._pendingRequests = new Map();
+    this._host = host;
+    this._port = typeof options.port === 'number' ? options.port : 7512;
+    this._ssl = typeof options.sslConnection === 'boolean' ? options.sslConnection : false;
 
     this.id = uuidv4();
     this.state = 'offline';
@@ -31,21 +26,25 @@ class AbstractWrapper extends KuzzleEventEmitter {
   }
 
   get host () {
-    return _host;
+    return this._host;
   }
 
   get port () {
-    return _port;
+    return this._port;
   }
 
   get ssl () {
-    return _ssl;
+    return this._ssl;
   }
 
   get connected () {
     return this.state === 'online';
   }
 
+  get pendingRequests () {
+    return this._pendingRequests;
+  }
+
   /**
    * @abstract
    * @returns {Promise<any>}
@@ -67,6 +66,8 @@ class AbstractWrapper extends KuzzleEventEmitter {
    * Called when the client's connection is established
    */
   clientConnected (state, wasConnected) {
+    this.on('disconnected', () => this.clear());
+
     this.state = state || 'ready';
     this.emit(wasConnected && 'reconnect' || 'connect');
   }
@@ -76,6 +77,7 @@ class AbstractWrapper extends KuzzleEventEmitter {
    */
   close () {
     this.state = 'offline';
+    this.clear();
   }
 
   query (request) {
@@ -85,8 +87,12 @@ class AbstractWrapper extends KuzzleEventEmitter {
 Discarded request: ${JSON.stringify(request)}`));
     }
 
+    this._pendingRequests.set(request.requestId, request);
+
     return new Promise((resolve, reject) => {
       this.once(request.requestId, response => {
+        this._pendingRequests.delete(request.requestId);
+
         if (response.error) {
           const error = new KuzzleError(response.error);
 
@@ -110,6 +116,18 @@ Discarded request: ${JSON.stringify(request)}`));
     return this.state === 'ready';
   }
 
+  /**
+   * Clear pendings requests.
+   * Emits an event for each discarded pending request.
+   */
+  clear () {
+    for (const request of this._pendingRequests.values()) {
+      this.emit('discarded', request);
+    }
+
+    this._pendingRequests.clear();
+  }
+
 }
 
 module.exports = AbstractWrapper;

test/protocol/query.test.js → test/protocol/common.test.js

@@ -4,33 +4,23 @@ const
   KuzzleError = require('../../src/KuzzleError'),
   AbstractWrapper = require('../../src/protocols/abstract/common');
 
-describe('Protocol query management', () => {
-  describe('#query', () => {
-    let
-      protocol;
-
-    beforeEach(function () {
-      protocol = new AbstractWrapper('somewhere');
-      protocol.send = function(request) {
-        protocol.emit(request.requestId, request.response);
-      };
-      sendSpy = sinon.spy(protocol, 'send');
-    });
-
+describe('Common Protocol', () => {
+  let
+    sendSpy,
+    protocol;
+
+  beforeEach(function () {
+    protocol = new AbstractWrapper('somewhere');
+    protocol.send = function(request) {
+      protocol.emit(request.requestId, request.response);
+    };
+    sendSpy = sinon.spy(protocol, 'send');
   });
 
   describe('#query', () => {
-    let
-      sendSpy,
-      protocol;
 
     beforeEach(() => {
-      protocol = new AbstractWrapper('somewhere');
       protocol.isReady = sinon.stub().returns(true);
-      protocol.send = function(request) {
-        protocol.emit(request.requestId, request.response);
-      };
-      sendSpy = sinon.spy(protocol, 'send');
       protocol._emitRequest = sinon.stub().resolves();
     });
 
@@ -50,11 +40,21 @@ describe('Protocol query management', () => {
       const request = {requestId: 'bar', response: {}};
 
       protocol.query(request);
+
       should(sendSpy)
         .be.calledOnce()
         .be.calledWith(request);
     });
 
+    it('should add the requests to pending requests', () => {
+      protocol.send = () => {};
+      const request = {requestId: 'bar', response: {}};
+
+      protocol.query(request);
+
+      should(protocol.pendingRequests.get('bar')).be.eql(request);
+    });
+
     it('should fire a "queryError" event and reject if an error occurred', () => {
       const
         eventStub = sinon.stub(),
@@ -103,6 +103,7 @@ describe('Protocol query management', () => {
           should(res.error).be.null();
           should(res.result).be.exactly(response.result);
           should(res.status).be.exactly(42);
+          should(protocol.pendingRequests.get('bar')).be.undefined();
         });
     });
 
@@ -151,6 +152,36 @@ describe('Protocol query management', () => {
           should(error.count).eql(42);
         });
     });
+  });
+
+  describe('#clear', () => {
+    it('should send "discarded" event for each pending request', () => {
+      const
+        request1 = { requestId: '12345', body: 'foobar' },
+        request2 = { requestId: '54321', body: 'barfoo' };
+
+      protocol._pendingRequests.set(request1, request1);
+      protocol._pendingRequests.set(request2, request2);
+
+      const listener = sinon.stub();
+      protocol.on('discarded', listener);
+
+      protocol.clear();
 
+      should(listener).be.calledTwice();
+      should(listener.getCall(0).args).be.eql([request1]);
+      should(listener.getCall(1).args).be.eql([request2]);
+    });
+  });
+
+  describe('#close', () => {
+    it('should set state to "offline" and clear pending requests', () => {
+      protocol.clear = sinon.stub();
+
+      protocol.close();
+
+      should(protocol.state).be.eql('offline');
+      should(protocol.clear).be.calledOnce();
+    });
   });
 });