refactor(cursor): move all wire protocol calls to Server

The primitive wire protocol methods are not meant to be used
directly outside of the `Server` class(es). The fact that the
cursors are doing so makes it difficult to provide consistent
error handling behavior required by SDAM/sessions/etc. This patch
moves these methods to the `Server`, with an eye towards
eventually moving them to `Connection`.

NODE-2066

Author: mbroadst

lib/core/cursor.js

@@ -7,7 +7,6 @@ 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;
 
@@ -195,7 +194,7 @@ Cursor.prototype._getMore = function(callback) {
     batchSize = this.cursorState.limit - this.cursorState.currentLimit;
   }
 
-  wireProtocol.getMore(this.server, this.ns, this.cursorState, batchSize, this.options, callback);
+  this.server.getMore(this.ns, this.cursorState, batchSize, this.options, callback);
 };
 
 /**
@@ -304,7 +303,7 @@ Cursor.prototype.kill = function(callback) {
     return;
   }
 
-  wireProtocol.killCursors(this.server, this.ns, this.cursorState, callback);
+  this.server.killCursors(this.ns, this.cursorState, callback);
 };
 
 /**
@@ -716,25 +715,11 @@ Cursor.prototype._initializeCursor = function(callback) {
     }
 
     if (cursor.cmd.find != null) {
-      wireProtocol.query(
-        cursor.server,
-        cursor.ns,
-        cursor.cmd,
-        cursor.cursorState,
-        cursor.options,
-        queryCallback
-      );
-
+      server.query(cursor.ns, cursor.cmd, cursor.cursorState, cursor.options, queryCallback);
       return;
     }
 
-    cursor.query = wireProtocol.command(
-      cursor.server,
-      cursor.ns,
-      cursor.cmd,
-      cursor.options,
-      queryCallback
-    );
+    server.command(cursor.ns, cursor.cmd, cursor.options, queryCallback);
   });
 };
 

lib/core/sdam/server.js

@@ -248,6 +248,61 @@ class Server extends EventEmitter {
     });
   }
 
+  /**
+   * Execute a query against the server
+   *
+   * @param {string} ns The MongoDB fully qualified namespace (ex: db1.collection1)
+   * @param {object} cmd The command document for the query
+   * @param {object} options Optional settings
+   * @param {function} callback
+   */
+  query(ns, cmd, cursorState, options, callback) {
+    wireProtocol.query(this, ns, cmd, cursorState, options, (err, result) => {
+      if (err && isSDAMUnrecoverableError(err)) {
+        this.emit('error', err);
+      }
+
+      callback(err, result);
+    });
+  }
+
+  /**
+   * Execute a `getMore` against the server
+   *
+   * @param {string} ns The MongoDB fully qualified namespace (ex: db1.collection1)
+   * @param {object} cursorState State data associated with the cursor calling this method
+   * @param {object} options Optional settings
+   * @param {function} callback
+   */
+  getMore(ns, cursorState, batchSize, options, callback) {
+    wireProtocol.getMore(this, ns, cursorState, batchSize, options, (err, result) => {
+      if (err && isSDAMUnrecoverableError(err)) {
+        this.emit('error', err);
+      }
+
+      callback(err, result);
+    });
+  }
+
+  /**
+   * Execute a `killCursors` command against the server
+   *
+   * @param {string} ns The MongoDB fully qualified namespace (ex: db1.collection1)
+   * @param {object} cursorState State data associated with the cursor calling this method
+   * @param {function} callback
+   */
+  killCursors(ns, cursorState, callback) {
+    wireProtocol.killCursors(this, ns, cursorState, (err, result) => {
+      if (err && isSDAMUnrecoverableError(err)) {
+        this.emit('error', err);
+      }
+
+      if (typeof callback === 'function') {
+        callback(err, result);
+      }
+    });
+  }
+
   /**
    * Insert one or more documents
    * @method

lib/core/topologies/server.js

@@ -628,6 +628,41 @@ Server.prototype.command = function(ns, cmd, options, callback) {
   wireProtocol.command(self, ns, cmd, options, callback);
 };
 
+/**
+ * Execute a query against the server
+ *
+ * @param {string} ns The MongoDB fully qualified namespace (ex: db1.collection1)
+ * @param {object} cmd The command document for the query
+ * @param {object} options Optional settings
+ * @param {function} callback
+ */
+Server.prototype.query = function(ns, cmd, cursorState, options, callback) {
+  wireProtocol.query(this, ns, cmd, cursorState, options, callback);
+};
+
+/**
+ * Execute a `getMore` against the server
+ *
+ * @param {string} ns The MongoDB fully qualified namespace (ex: db1.collection1)
+ * @param {object} cursorState State data associated with the cursor calling this method
+ * @param {object} options Optional settings
+ * @param {function} callback
+ */
+Server.prototype.getMore = function(ns, cursorState, batchSize, options, callback) {
+  wireProtocol.getMore(this, ns, cursorState, batchSize, options, callback);
+};
+
+/**
+ * Execute a `killCursors` command against the server
+ *
+ * @param {string} ns The MongoDB fully qualified namespace (ex: db1.collection1)
+ * @param {object} cursorState State data associated with the cursor calling this method
+ * @param {function} callback
+ */
+Server.prototype.killCursors = function(ns, cursorState, callback) {
+  wireProtocol.killCursors(this, ns, cursorState, callback);
+};
+
 /**
  * Insert one or more documents
  * @method