feat(mongo-client): implement MongoClient.prototype.startSession

NODE-1088

Author: mbroadst

lib/mongo_client.js

@@ -15,7 +15,9 @@ var parse = require('./url_parser'),
   f = require('util').format,
   assign = require('./utils').assign,
   shallowClone = require('./utils').shallowClone,
-  authenticate = require('./authenticate');
+  authenticate = require('./authenticate'),
+  ServerSessionPool = require('mongodb-core').Sessions.ServerSessionPool,
+  ClientSession = require('mongodb-core').Sessions.ClientSession;
 
 /**
  * @fileOverview The **MongoClient** class is a class that allows for making Connections to MongoDB.
@@ -180,7 +182,8 @@ function MongoClient(url, options) {
     url: url,
     options: options || {},
     promiseLibrary: null,
-    dbCache: {}
+    dbCache: {},
+    sessionPool: null
   };
 
   // Get the promiseLibrary
@@ -228,9 +231,8 @@ MongoClient.prototype.connect = function(callback) {
       // Did we have a validation error
       if (err) return reject(err);
       // Attempt to connect
-      connect(self, self.s.url, self.s.options, function(err, topology) {
+      connect(self, self.s.url, self.s.options, function(err) {
         if (err) return reject(err);
-        self.topology = topology;
         resolve(self);
       });
     });
@@ -239,9 +241,8 @@ MongoClient.prototype.connect = function(callback) {
   // Did we have a validation error
   if (err) return callback(err);
   // Fallback to callback based connect
-  connect(self, self.s.url, self.s.options, function(err, topology) {
+  connect(self, self.s.url, self.s.options, function(err) {
     if (err) return callback(err);
-    self.topology = topology;
     callback(null, self);
   });
 };
@@ -454,6 +455,28 @@ MongoClient.connect = function(url, options, callback) {
 
 define.staticMethod('connect', { callback: true, promise: true });
 
+/**
+ * Starts a new session on the server
+ *
+ * @param {object} [options] optional settings for a driver session
+ * @param {MongoClient~sessionCallback} [callback] The callback called with a newly establish session, or an error if one occurred
+ * @return {Promise} if no callback is specified, a promise will be returned for the newly established session
+ */
+MongoClient.prototype.startSession = function(options) {
+  options = options || {};
+  if (!this.topology) {
+    throw new MongoError('Must connect to a server before calling this method');
+  }
+
+  const capabilities = this.topology.capabilities();
+  if (capabilities && !capabilities.hasSessionSupport) {
+    throw new MongoError('Current topology does not support sessions');
+  }
+
+  const session = new ClientSession(this.topology.s.coreTopology, this.s.sessionPool, options);
+  return session;
+};
+
 var mergeOptions = function(target, source, flatten) {
   for (var name in source) {
     if (source[name] && typeof source[name] === 'object' && flatten) {
@@ -611,6 +634,11 @@ function relayEvents(self, topology) {
   });
 }
 
+function assignTopology(client, topology) {
+  client.topology = topology;
+  client.s.sessionPool = new ServerSessionPool(topology.s.coreTopology);
+}
+
 function createServer(self, options, callback) {
   // Set default options
   var servers = translateOptions(options);
@@ -629,8 +657,9 @@ function createServer(self, options, callback) {
     addListeners(self, servers[0]);
     // Check if we are really speaking to a mongos
     var ismaster = topology.lastIsMaster();
+
     // Set the topology
-    self.topology = topology;
+    assignTopology(self, topology);
 
     // Do we actually have a mongos
     if (ismaster && ismaster.msg === 'isdbgrid') {
@@ -659,7 +688,8 @@ function createReplicaset(self, options, callback) {
   // Open the connection
   topology.connect(options, function(err, topology) {
     if (err) return callback(err);
-    self.topology = topology;
+
+    assignTopology(self, topology);
     callback(null, topology);
   });
 }
@@ -676,7 +706,8 @@ function createMongos(self, options, callback) {
   // Open the connection
   topology.connect(options, function(err, topology) {
     if (err) return callback(err);
-    self.topology = topology;
+
+    assignTopology(self, topology);
     callback(null, topology);
   });
 }
@@ -772,7 +803,8 @@ var connect = function(self, url, options, callback) {
   // Did we pass in a Server/ReplSet/Mongos
   if (url instanceof Server || url instanceof ReplSet || url instanceof Mongos) {
     // Set the topology
-    self.topology = url;
+    assignTopology(self, url);
+
     // Add listeners
     addListeners(self, url);
     // Connect

package.json

@@ -13,7 +13,7 @@
     "official"
   ],
   "dependencies": {
-    "mongodb-core": "mongodb-js/mongodb-core#3.0.0"
+    "mongodb-core": "mongodb-js/mongodb-core#driver-sessions"
   },
   "devDependencies": {
     "betterbenchmarks": "^0.1.0",

test/functional/sessions_tests.js

@@ -0,0 +1,66 @@
+'use strict';
+
+const MongoClient = require('../..').MongoClient;
+const expect = require('chai').expect;
+const assign = require('../../lib/utils').assign;
+const mock = require('../mock');
+
+const test = {};
+describe('Sessions', function() {
+  describe('unit', function() {
+    afterEach(() => mock.cleanup());
+    beforeEach(() => {
+      return mock.createServer().then(server => {
+        test.server = server;
+      });
+    });
+
+    it('should throw an exception if sessions are not supported', {
+      metadata: { requires: { topology: 'single' } },
+      test: function(done) {
+        test.server.setMessageHandler(request => {
+          var doc = request.document;
+          if (doc.ismaster) {
+            request.reply(assign({}, mock.DEFAULT_ISMASTER));
+          }
+        });
+
+        MongoClient.connect(`mongodb://${test.server.uri()}/test`, function(err, client) {
+          expect(err).to.not.exist;
+          expect(() => {
+            client.startSession();
+          }).to.throw(/Current topology does not support sessions/);
+
+          client.close();
+          done();
+        });
+      }
+    });
+
+    it('should return a client session when requested if the topology supports it', {
+      metadata: { requires: { topology: 'single' } },
+
+      test: function(done) {
+        test.server.setMessageHandler(request => {
+          var doc = request.document;
+          if (doc.ismaster) {
+            request.reply(
+              assign({}, mock.DEFAULT_ISMASTER, {
+                logicalSessionTimeoutMinutes: 10
+              })
+            );
+          }
+        });
+
+        MongoClient.connect(`mongodb://${test.server.uri()}/test`, function(err, client) {
+          expect(err).to.not.exist;
+          let session = client.startSession();
+          expect(session).to.exist;
+
+          client.close();
+          done();
+        });
+      }
+    });
+  });
+});