refactor(pool): use cancellation token for connection establishment

Author: mbroadst

lib/core/connection/connect.js

@@ -266,16 +266,25 @@ function makeConnection(family, options, cancellationToken, _callback) {
   socket.setTimeout(connectionTimeout);
   socket.setNoDelay(noDelay);
 
+  let cancellationHandler;
   function errorHandler(eventName) {
     return err => {
       SOCKET_ERROR_EVENTS.forEach(event => socket.removeAllListeners(event));
+      if (cancellationHandler) {
+        cancellationToken.removeListener('cancel', cancellationHandler);
+      }
+
       socket.removeListener('connect', connectHandler);
       callback(connectionFailureError(eventName, err));
     };
   }
 
   function connectHandler() {
     SOCKET_ERROR_EVENTS.forEach(event => socket.removeAllListeners(event));
+    if (cancellationHandler) {
+      cancellationToken.removeListener('cancel', cancellationHandler);
+    }
+
     if (socket.authorizationError && rejectUnauthorized) {
       return callback(socket.authorizationError);
     }
@@ -286,7 +295,8 @@ function makeConnection(family, options, cancellationToken, _callback) {
 
   SOCKET_ERROR_EVENTS.forEach(event => socket.once(event, errorHandler(event)));
   if (cancellationToken) {
-    cancellationToken.once('cancel', errorHandler('cancel'));
+    cancellationHandler = errorHandler('cancel');
+    cancellationToken.once('cancel', cancellationHandler);
   }
 
   socket.once('connect', connectHandler);

lib/core/connection/pool.js

@@ -91,9 +91,13 @@ var Pool = function(topology, options) {
   this.topology = topology;
 
   this.s = {
-    state: DISCONNECTED
+    state: DISCONNECTED,
+    cancellationToken: new EventEmitter()
   };
 
+  // we don't care how many connections are listening for cancellation
+  this.s.cancellationToken.setMaxListeners(Infinity);
+
   // Add the options
   this.options = Object.assign(
     {
@@ -629,6 +633,9 @@ Pool.prototype.unref = function() {
 function destroy(self, connections, options, callback) {
   stateTransition(self, DESTROYING);
 
+  // indicate that in-flight connections should cancel
+  self.s.cancellationToken.emit('cancel');
+
   eachAsync(
     connections,
     (conn, cb) => {
@@ -748,6 +755,10 @@ Pool.prototype.reset = function(callback) {
     return;
   }
 
+  // signal in-flight connections should be cancelled
+  this.s.cancellationToken.emit('cancel');
+
+  // destroy existing connections
   const connections = this.availableConnections.concat(this.inUseConnections);
   eachAsync(
     connections,
@@ -984,7 +995,7 @@ function createConnection(pool, callback) {
   }
 
   pool.connectingConnections++;
-  connect(pool.options, (err, connection) => {
+  connect(pool.options, pool.s.cancellationToken, (err, connection) => {
     pool.connectingConnections--;
 
     if (err) {

test/unit/core/connect.test.js

@@ -131,10 +131,10 @@ describe('Connect Tests', function() {
 
   it('should allow a cancellaton token', function(done) {
     const cancellationToken = new EventEmitter();
-    setTimeout(() => cancellationToken.emit('cancel'), 100);
+    setTimeout(() => cancellationToken.emit('cancel'), 500);
     // set no response handler for mock server, effecively blackhole requests
 
-    connect(test.connectOptions, cancellationToken, (err, conn) => {
+    connect({ host: '240.0.0.1' }, cancellationToken, (err, conn) => {
       expect(err).to.exist;
       expect(err).to.match(/connection establishment was cancelled/);
       expect(conn).to.not.exist;