feat: support socket timeouts on a per-connection level

Author: mbroadst

lib/cmap/connection.js

@@ -51,6 +51,16 @@ class Connection extends EventEmitter {
       this.emit('close');
     });
 
+    stream.on('timeout', () => {
+      this.closed = true;
+      this[kQueue].forEach(op =>
+        op.cb(new MongoNetworkError(`connection ${this.id} to ${this.address} timed out`))
+      );
+      this[kQueue].clear();
+
+      this.emit('close');
+    });
+
     // hook the message stream up to the passed in stream
     stream.pipe(this[kMessageStream]);
     this[kMessageStream].pipe(stream);
@@ -159,7 +169,7 @@ function messageHandler(conn) {
 
     const callback = operationDescription.cb;
     if (operationDescription.socketTimeoutOverride) {
-      this[kStream].setSocketTimeout(this.socketTimeout);
+      conn[kStream].setTimeout(conn.socketTimeout);
     }
 
     try {
@@ -230,7 +240,7 @@ function write(command, options, callback) {
 
   if (typeof options.socketTimeout === 'number') {
     operationDescription.socketTimeoutOverride = true;
-    this[kStream].setSocketTimeout(options.socketTimeout);
+    this[kStream].setTimeout(options.socketTimeout);
   }
 
   // if command monitoring is enabled we need to modify the callback here

lib/cmap/connection_pool.js

@@ -42,6 +42,8 @@ const VALID_POOL_OPTIONS = new Set([
   'port',
   'bson',
   'connectionType',
+  'monitorCommands',
+  'socketTimeout',
 
   // spec options
   'maxPoolSize',

test/functional/cmap/connection.test.js

@@ -49,4 +49,19 @@ describe('Connection', function() {
       });
     });
   });
+
+  it('should support socket timeouts', function(done) {
+    const connectOptions = Object.assign({
+      host: '240.0.0.1',
+      connectionType: Connection,
+      bson: new BSON(),
+      connectionTimeout: 500
+    });
+
+    connect(connectOptions, err => {
+      expect(err).to.exist;
+      expect(err).to.match(/timed out/);
+      done();
+    });
+  });
 });

test/unit/cmap/connection_pool.test.js

@@ -87,6 +87,31 @@ describe('Connection Pool', function() {
     });
   });
 
+  it('should propagate socket timeouts to connections', function(done) {
+    server.setMessageHandler(request => {
+      const doc = request.document;
+      if (doc.ismaster) {
+        request.reply(mock.DEFAULT_ISMASTER_36);
+      } else {
+        // blackhole other requests
+      }
+    });
+
+    const pool = new ConnectionPool(
+      Object.assign({ bson: new BSON(), maxPoolSize: 1, socketTimeout: 500 }, server.address())
+    );
+
+    pool.withConnection((err, conn, cb) => {
+      conn.command('admin.$cmd', { ping: 1 }, (err, result) => {
+        expect(err).to.exist;
+        expect(result).to.not.exist;
+        expect(err).to.match(/timed out/);
+        cb();
+      });
+    }, () => pool.close(done));
+  });
+
+
   describe('withConnection', function() {
     it('should manage a connection for a successful operation', function(done) {
       server.setMessageHandler(request => {