refactor(sessions): move increment methods onto ClientSession

lib/sessions.js

@@ -34,6 +34,7 @@ class ClientSession extends EventEmitter {
    * @param {Boolean} [options.causalConsistency] Whether causal consistency should be enabled on this session
    * @param {Boolean} [options.autoStartTransaction=false] When enabled this session automatically starts a transaction with the provided defaultTransactionOptions.
    * @param {Object} [options.defaultTransactionOptions] The default TransactionOptions to use for transactions started on this session.
+   * @param {Object} [clientOptions] Optional settings provided when creating a client in the porcelain driver
    */
   constructor(topology, sessionPool, options, clientOptions) {
     super();
@@ -136,6 +137,23 @@ class ClientSession extends EventEmitter {
     return this.id.id.buffer.equals(session.id.id.buffer);
   }
 
+  /**
+   * Increment the transaction number on the internal ServerSession
+   */
+  incrementTransactionNumber() {
+    this.serverSession.txnNumber++;
+  }
+
+  /**
+   * Increment the statement id on the internal ServerSession
+   *
+   * @param {Number} [operationCount] the number of operations performed
+   */
+  incrementStatementId(operationCount) {
+    operationCount = operationCount || 1;
+    this.serverSession.stmtId += operationCount;
+  }
+
   /**
    * @returns whether this session is current in a transaction or not
    */
@@ -265,7 +283,7 @@ function endTransaction(clientSession, commandName, callback) {
   const command = { [commandName]: 1 };
   if (clientSession.transactionOptions.writeConcern) {
     Object.assign(command, { writeConcern: clientSession.transactionOptions.writeConcern });
-  } else if (clientSession.clientOptions.w) {
+  } else if (clientSession.clientOptions && clientSession.clientOptions.w) {
     Object.assign(command, { writeConcern: { w: clientSession.clientOptions.w } });
   }
 

lib/topologies/replset.js

@@ -17,8 +17,6 @@ const Interval = require('./shared').Interval;
 const createClientInfo = require('./shared').createClientInfo;
 const SessionMixins = require('./shared').SessionMixins;
 const isRetryableWritesSupported = require('./shared').isRetryableWritesSupported;
-const incrementTransactionNumber = require('./shared').incrementTransactionNumber;
-const incrementStatementId = require('./shared').incrementStatementId;
 const relayEvents = require('./shared').relayEvents;
 
 var MongoCR = require('../auth/mongocr'),
@@ -1244,7 +1242,7 @@ function executeWriteOperation(args, options, callback) {
 
   // increment and assign txnNumber
   if (willRetryWrite) {
-    incrementTransactionNumber(options.session);
+    options.session.incrementTransactionNumber();
   }
 
   // optionally autostart transaction if requested
@@ -1254,7 +1252,7 @@ function executeWriteOperation(args, options, callback) {
 
   // We need to increment the statement id if we're in a transaction
   if (options.session && options.session.inTransaction()) {
-    incrementStatementId(options.session, ops.length);
+    options.session.incrementStatementId(ops.length);
   }
 }
 

lib/topologies/shared.js

@@ -417,26 +417,6 @@ const isRetryableWritesSupported = function(topology) {
   return true;
 };
 
-/**
- * Increment the transaction number on the ServerSession contained by the provided ClientSession
- *
- * @param {ClientSession} session
- */
-function incrementTransactionNumber(session) {
-  session.serverSession.txnNumber++;
-}
-
-/**
- * Increment the statement id on the ServerSession contained by the provided ClientSession
- *
- * @param {ClientSession} session the client sessions
- * @param {Number} [operationCount] the number of operations performed
- */
-function incrementStatementId(session, operationCount) {
-  operationCount = operationCount || 1;
-  session.serverSession.stmtId += operationCount;
-}
-
 /**
  * Relays events for a given listener and emitter
  *
@@ -461,6 +441,4 @@ module.exports.diff = diff;
 module.exports.Interval = Interval;
 module.exports.Timeout = Timeout;
 module.exports.isRetryableWritesSupported = isRetryableWritesSupported;
-module.exports.incrementTransactionNumber = incrementTransactionNumber;
-module.exports.incrementStatementId = incrementStatementId;
 module.exports.relayEvents = relayEvents;

lib/wireprotocol/3_2_support.js

@@ -6,7 +6,6 @@ const f = require('util').format;
 const MongoError = require('../error').MongoError;
 const MongoNetworkError = require('../error').MongoNetworkError;
 const getReadPreference = require('./shared').getReadPreference;
-const incrementStatementId = require('../topologies/shared').incrementStatementId;
 const BSON = retrieveBSON();
 const Long = BSON.Long;
 
@@ -41,15 +40,14 @@ function decorateWithTransactionsData(command, session) {
 
   if (serverSession.stmtId === 0) {
     command.startTransaction = true;
-    command.readConcern = { level: 'snapshot' };
 
-    if (session.transactionOptions.readConcern) {
-      Object.assign(command, { readConcern: session.transactionOptions.readConcern });
-    } else if (session.clientOptions.readConcern) {
-      Object.assign(command, { readConcern: session.clientOptions.readConcern });
+    const readConcern = session.transactionOptions.readConcern || session.clientOptions.readConcern;
+    if (readConcern) {
+      command.readConcern = readConcern;
     }
 
     if (session.supports.causalConsistency && session.operationTime) {
+      command.readConcern = command.readConcern || {};
       Object.assign(command.readConcern, { afterClusterTime: session.operationTime });
     }
   } else {
@@ -318,7 +316,7 @@ WireProtocol.prototype.getMore = function(
 
   // We need to increment the statement id if we're in a transaction
   if (options.session && options.session.inTransaction()) {
-    incrementStatementId(options.session);
+    options.session.incrementStatementId();
   }
 
   // Write out the getMore command
@@ -353,7 +351,7 @@ WireProtocol.prototype.command = function(bson, ns, cmd, cursorState, topology,
 
   // We need to increment the statement id if we're in a transaction
   if (options.session && options.session.inTransaction()) {
-    incrementStatementId(options.session);
+    options.session.incrementStatementId();
   }
 
   return query;