feat(unified-sdam): backport unified SDAM to master for v3.2.0

Author: mbroadst

lib/operations/mongo_client_ops.js

@@ -11,6 +11,7 @@ const ReadPreference = require('mongodb-core').ReadPreference;
 const ReplSet = require('../topologies/replset');
 const Server = require('../topologies/server');
 const ServerSessionPool = require('mongodb-core').Sessions.ServerSessionPool;
+const NativeTopology = require('../topologies/native_topology');
 
 let client;
 function loadClient() {
@@ -109,7 +110,9 @@ const validOptionNames = [
   'minSize',
   'monitorCommands',
   'retryWrites',
-  'useNewUrlParser'
+  'useNewUrlParser',
+  'useUnifiedTopology',
+  'serverSelectionTimeoutMS'
 ];
 
 function addListeners(mongoClient, topology) {
@@ -126,7 +129,10 @@ function addListeners(mongoClient, topology) {
 
 function assignTopology(client, topology) {
   client.topology = topology;
-  topology.s.sessionPool = new ServerSessionPool(topology.s.coreTopology);
+  topology.s.sessionPool =
+    topology instanceof NativeTopology
+      ? new ServerSessionPool(topology)
+      : new ServerSessionPool(topology.s.coreTopology);
 }
 
 // Clear out all events
@@ -225,6 +231,15 @@ function connect(mongoClient, url, options, callback) {
       }
     }
 
+    if (_finalOptions.useUnifiedTopology) {
+      return createTopology(
+        mongoClient,
+        'unified',
+        _finalOptions,
+        connectHandler(mongoClient, _finalOptions, connectCallback)
+      );
+    }
+
     // Do we have a replicaset then skip discovery and go straight to connectivity
     if (_finalOptions.replicaSet || _finalOptions.rs_name) {
       return createTopology(
@@ -434,6 +449,8 @@ function createTopology(mongoClient, topologyType, options, callback) {
     topology = new Mongos(servers, options);
   } else if (topologyType === 'replicaset') {
     topology = new ReplSet(servers, options);
+  } else if (topologyType === 'unified') {
+    topology = new NativeTopology(options.servers, options);
   }
 
   // Add listeners

lib/topologies/native_topology.js

@@ -0,0 +1,51 @@
+'use strict';
+
+const Topology = require('mongodb-core').Topology;
+const ServerCapabilities = require('./topology_base').ServerCapabilities;
+const Cursor = require('../cursor');
+const translateOptions = require('../utils').translateOptions;
+
+class NativeTopology extends Topology {
+  constructor(servers, options) {
+    options = options || {};
+
+    let clonedOptions = Object.assign(
+      {},
+      {
+        cursorFactory: Cursor,
+        reconnect: false,
+        emitError: options.emitError,
+        size: options.poolSize,
+        monitorCommands: options.monitorCommands
+      }
+    );
+
+    // Translate any SSL options and other connectivity options
+    clonedOptions = translateOptions(clonedOptions, options);
+
+    // Socket options
+    var socketOptions =
+      options.socketOptions && Object.keys(options.socketOptions).length > 0
+        ? options.socketOptions
+        : options;
+
+    // Translate all the options to the mongodb-core ones
+    clonedOptions = translateOptions(clonedOptions, socketOptions);
+
+    super(servers, clonedOptions);
+
+    // Do we have an application specific string
+    if (options.appname) {
+      this.s.clientInfo.application = { name: options.appname };
+    }
+  }
+
+  capabilities() {
+    if (this.s.sCapabilities) return this.s.sCapabilities;
+    if (this.lastIsMaster() == null) return null;
+    this.s.sCapabilities = new ServerCapabilities(this.lastIsMaster());
+    return this.s.sCapabilities;
+  }
+}
+
+module.exports = NativeTopology;

lib/utils.js

@@ -588,12 +588,12 @@ function decorateWithCollation(command, target, options) {
     throw new TypeError('parameter "target" is missing a topology');
   }
 
-  const capabilities = target.s.topology.capabilities();
+  const capabilities = topology.capabilities();
   if (options.collation && typeof options.collation === 'object') {
     if (capabilities && capabilities.commandsTakeCollation) {
       command.collation = options.collation;
     } else {
-      throw new MongoError(`server ${topology.s.coreTopology.name} does not support collation`);
+      throw new MongoError(`Current topology does not support collation`);
     }
   }
 }

test/config.js

@@ -4,18 +4,26 @@ const f = require('util').format;
 const url = require('url');
 const qs = require('querystring');
 class NativeConfiguration extends ConfigurationBase {
-  constructor(options) {
-    super(options);
+  constructor(environment) {
+    super(environment);
 
     this.type = 'native';
-    this.topology = options.topology || this.defaultTopology;
-    this.replicasetName = options.replicasetName || 'rs';
+    this.topology = environment.topology || this.defaultTopology;
+    this.environment = environment;
+
+    if (environment.setName) {
+      this.replicasetName = environment.setName || 'rs';
+    }
   }
 
   defaultTopology(serverHost, serverPort, serverOpts, _mongo) {
     return new _mongo.Server(serverHost, serverPort, serverOpts || {});
   }
 
+  usingUnifiedTopology() {
+    return !!process.env.MONGODB_UNIFIED_TOPOLOGY;
+  }
+
   start(callback) {
     const self = this;
     if (this.skipStart) return callback();
@@ -50,11 +58,20 @@ class NativeConfiguration extends ConfigurationBase {
   newClient(dbOptions, serverOptions) {
     // support MongoClient contructor form (url, options) for `newClient`
     if (typeof dbOptions === 'string') {
-      return new this.mongo.MongoClient(dbOptions, serverOptions);
+      return new this.mongo.MongoClient(
+        dbOptions,
+        this.usingUnifiedTopology()
+          ? Object.assign(
+              { useUnifiedTopology: true, serverSelectionTimeoutMS: 1000 },
+              serverOptions
+            )
+          : serverOptions
+      );
     }
 
     dbOptions = dbOptions || {};
     serverOptions = Object.assign({}, { haInterval: 100 }, serverOptions);
+    if (this.usingUnifiedTopology()) serverOptions.useUnifiedTopology = true;
 
     // Override implementation
     if (this.options.newDbInstance) {

test/examples/aggregate.js

@@ -2,7 +2,6 @@
 /* eslint-disable no-unused-vars */
 
 const setupDatabase = require('../functional/shared').setupDatabase;
-const MongoClient = require('../../lib/mongo_client');
 
 describe('examples.aggregaton:', function() {
   let client;
@@ -13,7 +12,7 @@ describe('examples.aggregaton:', function() {
   });
 
   beforeEach(async function() {
-    client = await MongoClient.connect(this.configuration.url());
+    client = await this.configuration.newClient().connect();
     collection = client.db(this.configuration.db).collection('aggregateExample');
   });
 

test/examples/array_filters.js

@@ -1,7 +1,6 @@
 'use strict';
 
 const setupDatabase = require('../functional/shared').setupDatabase;
-const MongoClient = require('../../lib/mongo_client');
 
 describe('examples(project-fields-from-query):', function() {
   let client;
@@ -12,7 +11,7 @@ describe('examples(project-fields-from-query):', function() {
   });
 
   beforeEach(async function() {
-    client = await MongoClient.connect(this.configuration.url());
+    client = await this.configuration.newClient().connect();
     collection = client.db(this.configuration.db).collection('arrayFilterUpdateExample');
   });
 

test/examples/causal_consistency.js

@@ -2,7 +2,6 @@
 
 const setupDatabase = require('../functional/shared').setupDatabase;
 const expect = require('chai').expect;
-const MongoClient = require('../../lib/mongo_client');
 
 describe('examples(causal-consistency):', function() {
   let client;
@@ -14,7 +13,7 @@ describe('examples(causal-consistency):', function() {
   });
 
   beforeEach(async function() {
-    client = await MongoClient.connect(this.configuration.url());
+    client = await this.configuration.newClient().connect();
     collection = client.db(this.configuration.db).collection('arrayFilterUpdateExample');
   });
 

test/examples/change_streams.js

@@ -3,7 +3,6 @@
 
 const setupDatabase = require('../functional/shared').setupDatabase;
 const expect = require('chai').expect;
-const MongoClient = require('../../lib/mongo_client');
 
 describe('examples(change-stream):', function() {
   let client;
@@ -14,7 +13,7 @@ describe('examples(change-stream):', function() {
   });
 
   beforeEach(async function() {
-    client = await MongoClient.connect(this.configuration.url());
+    client = await this.configuration.newClient().connect();
     db = client.db(this.configuration.db);
 
     await db.collection('inventory').deleteMany({});

test/examples/create_index.js

@@ -1,7 +1,6 @@
 'use strict';
 
 const setupDatabase = require('../functional/shared').setupDatabase;
-const MongoClient = require('../../lib/mongo_client');
 
 describe('examples.createIndex:', function() {
   let client;
@@ -12,7 +11,7 @@ describe('examples.createIndex:', function() {
   });
 
   beforeEach(async function() {
-    client = await MongoClient.connect(this.configuration.url());
+    client = await this.configuration.newClient().connect();
     collection = client.db(this.configuration.db).collection('createIndexExample');
   });
 

test/examples/insert.js

@@ -2,7 +2,6 @@
 
 const setupDatabase = require('../functional/shared').setupDatabase;
 const expect = require('chai').expect;
-const MongoClient = require('../../lib/mongo_client');
 
 describe('examples(insert):', function() {
   let client;
@@ -13,7 +12,7 @@ describe('examples(insert):', function() {
   });
 
   beforeEach(async function() {
-    client = await MongoClient.connect(this.configuration.url());
+    client = await this.configuration.newClient().connect();
     db = client.db(this.configuration.db);
 
     await db.collection('inventory').deleteMany({});