refactor(apm): correct apm event assumptions for OP_MSG messages

Author: mbroadst

lib/connection/msg.js

@@ -28,6 +28,7 @@
 // };
 
 const opcodes = require('../wireprotocol/shared').opcodes;
+const databaseNamespace = require('../wireprotocol/shared').databaseNamespace;
 
 // Incrementing request id
 let _requestId = 0;
@@ -38,13 +39,15 @@ const OPTS_MORE_TO_COME = 2;
 const OPTS_EXHAUST_ALLOWED = 1 >> 16;
 
 class Msg {
- constructor(bson, query, options) {
+ constructor(bson, ns, command, options) {
  // Basic options needed to be passed in
- if (query == null) throw new Error('query must be specified for query');
+ if (command == null) throw new Error('query must be specified for query');
 
  // Basic options
  this.bson = bson;
- this.query = Array.isArray(query) ? query : [query];
+ this.ns = ns;
+ this.command = command;
+ this.command.$db = databaseNamespace(ns);
 
  // Ensure empty options
  this.options = options || {};
@@ -91,24 +94,21 @@ class Msg {
 
  let totalLength = header.length;
 
- for (let i = 0; i < this.query.length; ++i) {
- const query = this.query[i];
-
- const nameArgumentPair = getValidSegmentListNamePairs(query);
- if (nameArgumentPair) {
- // TODO: Add support for payload type 1
- const argument = nameArgumentPair.argument;
-
- // Add initial type 0 segment with arguments pulled up
- const clonedQuery = Object.assign({}, query);
- delete clonedQuery[argument];
- totalLength += this.makeDocumentSegment(buffers, clonedQuery);
-
- // Create type 1 query
- totalLength += this.makeSequenceSegment(buffers, argument, query[argument]);
- } else {
- totalLength += this.makeDocumentSegment(buffers, query);
- }
+ const command = this.command;
+ const nameArgumentPair = validSegmentListNamePairs(command);
+ if (nameArgumentPair) {
+ // TODO: Add support for payload type 1
+ const argument = nameArgumentPair.argument;
+
+ // Add initial type 0 segment with arguments pulled up
+ const clonedQuery = Object.assign({}, command);
+ delete clonedQuery[argument];
+ totalLength += this.makeDocumentSegment(buffers, clonedQuery);
+
+ // Create type 1 query
+ totalLength += this.makeSequenceSegment(buffers, argument, command[argument]);
+ } else {
+ totalLength += this.makeDocumentSegment(buffers, command);
  }
 
  writeInt32ListToUint8Buffer(header, [totalLength, this.requestId, 0, opcodes.OP_MSG, flags]);
@@ -183,6 +183,7 @@ class BinMsg {
  this.requestId = msgHeader.requestId;
  this.responseTo = msgHeader.responseTo;
  this.opCode = msgHeader.opCode;
+ this.fromCompressed = msgHeader.fromCompressed;
 
  // Read response flags
  this.responseFlags = msgBody.readInt32LE(0);
@@ -266,7 +267,7 @@ const VALID_NAME_ARGUMENT_MAPS = {
  delete: 'deletes'
 };
 
-function getValidSegmentListNamePairs(query) {
+function validSegmentListNamePairs(query) {
  for (let name in VALID_NAME_ARGUMENT_MAPS) {
  if (name in query) {
  const argument = VALID_NAME_ARGUMENT_MAPS[name];

lib/connection/pool.js

@@ -9,6 +9,7 @@ const MongoWriteConcernError = require('../error').MongoWriteConcernError;
 const Logger = require('./logger');
 const f = require('util').format;
 const Query = require('./commands').Query;
+const Msg = require('./msg').Msg;
 const CommandResult = require('./command_result');
 const MESSAGE_HEADER_SIZE = require('../wireprotocol/shared').MESSAGE_HEADER_SIZE;
 const COMPRESSION_DETAILS_SIZE = require('../wireprotocol/shared').COMPRESSION_DETAILS_SIZE;
@@ -1273,7 +1274,8 @@ Pool.prototype.write = function(command, options, cb) {
 // Return whether a command contains an uncompressible command term
 // Will return true if command contains no uncompressible command terms
 function canCompress(command) {
- const commandName = Object.keys(command.query)[0];
+ const commandDoc = command instanceof Msg ? command.command : command.query;
+ const commandName = Object.keys(commandDoc)[0];
  return uncompressibleCommands.indexOf(commandName) === -1;
 }
 

lib/sdam/server_description.js

@@ -31,7 +31,8 @@ const ISMASTER_FIELDS = [
  'setVersion',
  'electionId',
  'primary',
- 'logicalSessionTimeoutMinutes'
+ 'logicalSessionTimeoutMinutes',
+ '__nodejs_mock_server__'
 ];
 
 /**

lib/wireprotocol/command.js

@@ -1,6 +1,7 @@
 'use strict';
 
 const Query = require('../connection/commands').Query;
+const Msg = require('../connection/msg').Msg;
 const retrieveBSON = require('../connection/utils').retrieveBSON;
 const MongoError = require('../error').MongoError;
 const getReadPreference = require('./shared').getReadPreference;
@@ -48,14 +49,30 @@ function command(server, ns, cmd, options, callback) {
  // This value is not overridable
  commandOptions.slaveOk = readPreference.slaveOk();
 
+ const cmdNs = `${databaseNamespace(ns)}.$cmd`;
+ const message = supportsOpMsg(server)
+ ? new Msg(bson, cmdNs, finalCmd, commandOptions)
+ : new Query(bson, cmdNs, finalCmd, commandOptions);
+
  try {
- const query = new Query(bson, `${databaseNamespace(ns)}.$cmd`, finalCmd, commandOptions);
- pool.write(query, commandOptions, callback);
+ pool.write(message, commandOptions, callback);
  } catch (err) {
  callback(err);
  }
 }
 
+function supportsOpMsg(topologyOrServer) {
+ const description = topologyOrServer.ismaster
+ ? topologyOrServer.ismaster
+ : topologyOrServer.description;
+
+ if (description == null) {
+ return false;
+ }
+
+ return description.maxWireVersion >= 6 && description.__nodejs_mock_server__ == null;
+}
+
 function isTransactionCommand(command) {
  return !!(command.commitTransaction || command.abortTransaction);
 }