feat: support connection establishment cancellation

This allows us to cancel in-flight connection establishment in
cases where connections might be blackholed, or waiting for an
attempt to time out in `connectTimeoutMS`

NODE-2363

Author: mbroadst

lib/core/connection/connect.js

@@ -14,14 +14,19 @@ const MIN_SUPPORTED_WIRE_VERSION = WIRE_CONSTANTS.MIN_SUPPORTED_WIRE_VERSION;
 const MIN_SUPPORTED_SERVER_VERSION = WIRE_CONSTANTS.MIN_SUPPORTED_SERVER_VERSION;
 let AUTH_PROVIDERS;
 
-function connect(options, callback) {
+function connect(options, cancellationToken, callback) {
+  if (typeof cancellationToken === 'function') {
+    callback = cancellationToken;
+    cancellationToken = undefined;
+  }
+
   const ConnectionType = options && options.connectionType ? options.connectionType : Connection;
   if (AUTH_PROVIDERS == null) {
     AUTH_PROVIDERS = defaultAuthProviders(options.bson);
   }
 
   const family = options.family !== void 0 ? options.family : 0;
-  makeConnection(family, options, (err, socket) => {
+  makeConnection(family, options, cancellationToken, (err, socket) => {
     if (err) {
       callback(err, socket); // in the error case, `socket` is the originating error event name
       return;
@@ -219,7 +224,7 @@ function parseSslOptions(family, options) {
 }
 
 const SOCKET_ERROR_EVENTS = new Set(['error', 'close', 'timeout', 'parseError']);
-function makeConnection(family, options, _callback) {
+function makeConnection(family, options, cancellationToken, _callback) {
   const useSsl = typeof options.ssl === 'boolean' ? options.ssl : false;
   const keepAlive = typeof options.keepAlive === 'boolean' ? options.keepAlive : true;
   let keepAliveInitialDelay =
@@ -240,6 +245,7 @@ function makeConnection(family, options, _callback) {
     if (err && socket) {
       socket.destroy();
     }
+
     _callback(err, ret);
   };
 
@@ -264,7 +270,7 @@ function makeConnection(family, options, _callback) {
     return err => {
       SOCKET_ERROR_EVENTS.forEach(event => socket.removeAllListeners(event));
       socket.removeListener('connect', connectHandler);
-      callback(connectionFailureError(eventName, err), eventName);
+      callback(connectionFailureError(eventName, err));
     };
   }
 
@@ -279,6 +285,10 @@ function makeConnection(family, options, _callback) {
   }
 
   SOCKET_ERROR_EVENTS.forEach(event => socket.once(event, errorHandler(event)));
+  if (cancellationToken) {
+    cancellationToken.once('cancel', errorHandler('cancel'));
+  }
+
   socket.once('connect', connectHandler);
 }
 
@@ -359,6 +369,8 @@ function connectionFailureError(type, err) {
       return new MongoNetworkError(`connection timed out`);
     case 'close':
       return new MongoNetworkError(`connection closed`);
+    case 'cancel':
+      return new MongoNetworkError(`connection establishment was cancelled`);
     default:
       return new MongoNetworkError(`unknown network error`);
   }

test/unit/core/connect.test.js

@@ -129,6 +129,19 @@ describe('Connect Tests', function() {
     }
   );
 
+  it('should allow a cancellaton token', function(done) {
+    const cancellationToken = new EventEmitter();
+    setTimeout(() => cancellationToken.emit('cancel'), 100);
+    // set no response handler for mock server, effecively blackhole requests
+
+    connect(test.connectOptions, cancellationToken, (err, conn) => {
+      expect(err).to.exist;
+      expect(err).to.match(/connection establishment was cancelled/);
+      expect(conn).to.not.exist;
+      done();
+    });
+  });
+
   describe('runCommand', function() {
     const metadata = { requires: { topology: 'single' } };
     class MockConnection extends EventEmitter {