refactor(wire-protocol): flatten wire protocol handler hierarchy

See the ticket for more details, but this generally removes the
class-based module design for wire protocol handling, removing
needless allocations, and focusing more on the meat of the methods
hung off those classes.

NODE-1835

Author: mbroadst

lib/cursor.js

@@ -7,7 +7,7 @@ const MongoNetworkError = require('./error').MongoNetworkError;
 const mongoErrorContextSymbol = require('./error').mongoErrorContextSymbol;
 const f = require('util').format;
 const collationNotSupported = require('./utils').collationNotSupported;
-
+const wireProtocol = require('./wireprotocol');
 const BSON = retrieveBSON();
 const Long = BSON.Long;
 
@@ -223,14 +223,7 @@ Cursor.prototype._getmore = function(callback) {
     batchSize = this.cursorState.limit - this.cursorState.currentLimit;
   }
 
-  this.server.wireProtocolHandler.getMore(
-    this.server,
-    this.ns,
-    this.cursorState,
-    batchSize,
-    this.options,
-    callback
-  );
+  wireProtocol.getMore(this.server, this.ns, this.cursorState, batchSize, this.options, callback);
 };
 
 /**
@@ -339,7 +332,7 @@ Cursor.prototype.kill = function(callback) {
     return;
   }
 
-  this.server.wireProtocolHandler.killCursor(this.server, this.ns, this.cursorState, callback);
+  wireProtocol.killCursors(this.server, this.ns, this.cursorState, callback);
 };
 
 /**
@@ -732,7 +725,7 @@ function initializeCursor(cursor, callback) {
     }
 
     if (cursor.cmd.find != null) {
-      cursor.server.wireProtocolHandler.query(
+      wireProtocol.query(
         cursor.server,
         cursor.ns,
         cursor.cmd,
@@ -744,7 +737,7 @@ function initializeCursor(cursor, callback) {
       return;
     }
 
-    cursor.query = cursor.server.wireProtocolHandler.command(
+    cursor.query = wireProtocol.command(
       cursor.server,
       cursor.ns,
       cursor.cmd,

lib/sdam/server.js

@@ -4,8 +4,7 @@ const MongoError = require('../error').MongoError;
 const Pool = require('../connection/pool');
 const relayEvents = require('../utils').relayEvents;
 const calculateDurationInMs = require('../utils').calculateDurationInMs;
-const TwoSixWireProtocolSupport = require('../wireprotocol/2_6_support');
-const ThreeTwoWireProtocolSupport = require('../wireprotocol/3_2_support');
+const wireProtocol = require('../wireprotocol');
 const BSON = require('../connection/utils').retrieveBSON();
 const createClientInfo = require('../topologies/shared').createClientInfo;
 const Logger = require('../connection/logger');
@@ -155,27 +154,7 @@ class Server extends EventEmitter {
       return;
     }
 
-    // Create the query object
-    const query = this.s.wireProtocolHandler.command(this, ns, cmd, {}, options);
-    // Set slave OK of the query
-    query.slaveOk = options.readPreference ? options.readPreference.slaveOk() : false;
-
-    // write options
-    const writeOptions = {
-      raw: typeof options.raw === 'boolean' ? options.raw : false,
-      promoteLongs: typeof options.promoteLongs === 'boolean' ? options.promoteLongs : true,
-      promoteValues: typeof options.promoteValues === 'boolean' ? options.promoteValues : true,
-      promoteBuffers: typeof options.promoteBuffers === 'boolean' ? options.promoteBuffers : false,
-      command: true,
-      monitoring: typeof options.monitoring === 'boolean' ? options.monitoring : false,
-      fullResult: typeof options.fullResult === 'boolean' ? options.fullResult : false,
-      requestId: query.requestId,
-      socketTimeout: typeof options.socketTimeout === 'number' ? options.socketTimeout : null,
-      session: options.session || null
-    };
-
-    // write the operation to the pool
-    this.s.pool.write(query, writeOptions, callback);
+    wireProtocol.command(this, ns, cmd, options, callback);
   }
 
   /**
@@ -273,7 +252,7 @@ function executeWriteOperation(args, options, callback) {
   }
 
   // Execute write
-  return server.s.wireProtocolHandler[op](server.s.pool, ns, server.s.bson, ops, options, callback);
+  return wireProtocol[op](server, ns, ops, options, callback);
 }
 
 function saslSupportedMechs(options) {
@@ -321,16 +300,6 @@ function executeServerHandshake(server, callback) {
   );
 }
 
-function configureWireProtocolHandler(ismaster) {
-  // 3.2 wire protocol handler
-  if (ismaster.maxWireVersion >= 4) {
-    return new ThreeTwoWireProtocolSupport();
-  }
-
-  // default to 2.6 wire protocol handler
-  return new TwoSixWireProtocolSupport();
-}
-
 function connectEventHandler(server) {
   return function() {
     // log information of received information if in info mode
@@ -370,9 +339,6 @@ function connectEventHandler(server) {
         }
       }
 
-      // configure the wire protocol handler
-      server.s.wireProtocolHandler = configureWireProtocolHandler(isMaster);
-
       // log the connection event if requested
       if (server.s.logger.isInfo()) {
         server.s.logger.info(

lib/topologies/server.js

@@ -10,8 +10,7 @@ var inherits = require('util').inherits,
   Pool = require('../connection/pool'),
   MongoError = require('../error').MongoError,
   MongoNetworkError = require('../error').MongoNetworkError,
-  TwoSixWireProtocolSupport = require('../wireprotocol/2_6_support'),
-  ThreeTwoWireProtocolSupport = require('../wireprotocol/3_2_support'),
+  wireProtocol = require('../wireprotocol'),
   BasicCursor = require('../cursor'),
   sdam = require('./shared'),
   createClientInfo = require('./shared').createClientInfo,
@@ -224,9 +223,6 @@ var Server = function(options) {
   this.monitoringProcessId = null;
   // Initial connection
   this.initialConnect = true;
-  // Wire protocol handler, default to oldest known protocol handler
-  // this gets changed when the first ismaster is called.
-  this.wireProtocolHandler = new TwoSixWireProtocolSupport();
   // Default type
   this._type = 'server';
   // Set the client info
@@ -306,16 +302,6 @@ function isSupportedServer(response) {
   return response && typeof response.maxWireVersion === 'number' && response.maxWireVersion >= 2;
 }
 
-function configureWireProtocolHandler(self, ismaster) {
-  // 3.2 wire protocol handler
-  if (ismaster.maxWireVersion >= 4) {
-    return new ThreeTwoWireProtocolSupport();
-  }
-
-  // default to 2.6 wire protocol handler
-  return new TwoSixWireProtocolSupport();
-}
-
 function disconnectHandler(self, type, ns, cmd, options, callback) {
   // Topology is not connected, save the call in the provided store to be
   // Executed at some point when the handler deems it's reconnected
@@ -451,8 +437,7 @@ var eventHandler = function(self, event) {
           if (self.ismaster.msg === 'isdbgrid') {
             self._type = 'mongos';
           }
-          // Add the correct wire protocol handler
-          self.wireProtocolHandler = configureWireProtocolHandler(self, self.ismaster);
+
           // Have we defined self monitoring
           if (self.s.monitoring) {
             self.monitoringProcessId = setTimeout(
@@ -744,7 +729,7 @@ Server.prototype.command = function(ns, cmd, options, callback) {
     return callback(new MongoError(`server ${this.name} does not support collation`));
   }
 
-  self.wireProtocolHandler.command(self, ns, cmd, options, callback);
+  wireProtocol.command(self, ns, cmd, options, callback);
 };
 
 /**
@@ -775,7 +760,7 @@ Server.prototype.insert = function(ns, ops, options, callback) {
   ops = Array.isArray(ops) ? ops : [ops];
 
   // Execute write
-  return self.wireProtocolHandler.insert(self, ns, ops, options, callback);
+  return wireProtocol.insert(self, ns, ops, options, callback);
 };
 
 /**
@@ -810,7 +795,7 @@ Server.prototype.update = function(ns, ops, options, callback) {
   // Setup the docs as an array
   ops = Array.isArray(ops) ? ops : [ops];
   // Execute write
-  return self.wireProtocolHandler.update(self, ns, ops, options, callback);
+  return wireProtocol.update(self, ns, ops, options, callback);
 };
 
 /**
@@ -845,7 +830,7 @@ Server.prototype.remove = function(ns, ops, options, callback) {
   // Setup the docs as an array
   ops = Array.isArray(ops) ? ops : [ops];
   // Execute write
-  return self.wireProtocolHandler.remove(self, ns, ops, options, callback);
+  return wireProtocol.remove(self, ns, ops, options, callback);
 };
 
 /**

lib/utils.js

@@ -75,6 +75,25 @@ function retrieveEJSON() {
   return EJSON;
 }
 
+/**
+ * A helper function for determining `maxWireVersion` between legacy and new topology
+ * instances
+ *
+ * @private
+ * @param {(Topology|Server)} topologyOrServer
+ */
+function maxWireVersion(topologyOrServer) {
+  if (topologyOrServer.ismaster) {
+    return topologyOrServer.ismaster.maxWireVersion;
+  }
+
+  if (topologyOrServer.description) {
+    return topologyOrServer.description.maxWireVersion;
+  }
+
+  return null;
+}
+
 /*
  * Checks that collation is supported by server.
  *
@@ -84,7 +103,7 @@ function retrieveEJSON() {
  * @return true if server does not support collation
  */
 function collationNotSupported(server, cmd) {
-  return cmd && cmd.collation && server.ismaster && server.ismaster.maxWireVersion < 5;
+  return cmd && cmd.collation && maxWireVersion(server) < 5;
 }
 
 module.exports = {
@@ -93,5 +112,6 @@ module.exports = {
   relayEvents,
   collationNotSupported,
   retrieveEJSON,
-  retrieveKerberos
+  retrieveKerberos,
+  maxWireVersion
 };