feat(cluster-time): ensure clusterTime makes it to outgoing commands

NODE-1088

Author: mbroadst

lib/connection/pool.js

@@ -31,6 +31,10 @@ var DESTROYED = 'destroyed';
 
 var _id = 0;
 
+function supportsClusterTime(topology) {
+  return topology.ismaster == null ? false : topology.ismaster.maxWireVersion >= 6;
+}
+
 /**
  * Creates a new Pool instance
  * @class
@@ -1137,6 +1141,16 @@ Pool.prototype.write = function(commands, options, cb) {
   // Get the requestId
   operation.requestId = commands[commands.length - 1].requestId;
 
+  if (supportsClusterTime(this.topology) && this.topology.clusterTime) {
+    commands.forEach(command => {
+      if (command instanceof Query) {
+        command.query.$clusterTime = this.topology.clusterTime;
+      } else {
+        command.$clusterTime = this.topology.clusterTime;
+      }
+    });
+  }
+
   // Prepare the operation buffer
   serializeCommands(self, commands, [], function(err, serializedCommands) {
     if (err) throw err;

test/tests/unit/single/sessions_tests.js

@@ -83,6 +83,40 @@ describe('Sessions (Single)', function() {
     }
   });
 
+  it('should send `clusterTime` on outgoing messages', {
+    metadata: { requires: { topology: 'single' } },
+    test: function(done) {
+      const clusterTime = genClusterTime(Date.now());
+      let sentIsMaster = false;
+      test.server.setMessageHandler(request => {
+        if (sentIsMaster) {
+          expect(request.document.$clusterTime).to.eql(clusterTime);
+          request.reply({ ok: 1 });
+          return;
+        }
+
+        sentIsMaster = true;
+        request.reply(
+          assign({}, mock.DEFAULT_ISMASTER, {
+            maxWireVersion: 6,
+            $clusterTime: clusterTime
+          })
+        );
+      });
+
+      const client = new Server(test.server.address());
+      client.on('error', done);
+      client.once('connect', () => {
+        client.command('admin.$cmd', { ping: 1 }, err => {
+          expect(err).to.not.exist;
+          done();
+        });
+      });
+
+      client.connect();
+    }
+  });
+
   it('should default `logicalSessionTimeoutMinutes` to `null`', {
     metadata: { requires: { topology: 'single' } },
     test: function() {