feat(NODE-4139): streaming protocol message changes

src/cmap/connection.ts

@@ -293,6 +293,16 @@ export class Connection extends TypedEventEmitter<ConnectionEvents> {
     this[kHello] = response;
   }
 
+  // Set the whether the message stream is for a monitoring connection using the
+  // streaming protocol.
+  set isStreamingProtocol(value: boolean) {
+    this[kMessageStream].isStreamingProtocol = value;
+  }
+
+  get isStreamingProtocol(): boolean {
+    return this[kMessageStream].isStreamingProtocol;
+  }
+
   get serviceId(): ObjectId | undefined {
     return this.hello?.serviceId;
   }

src/cmap/message_stream.ts

@@ -53,13 +53,19 @@ export class MessageStream extends Duplex {
   maxBsonMessageSize: number;
   /** @internal */
   [kBuffer]: BufferPool;
+  /** @internal */
+  isStreamingProtocol = false;
 
   constructor(options: MessageStreamOptions = {}) {
     super(options);
     this.maxBsonMessageSize = options.maxBsonMessageSize || kDefaultMaxBsonMessageSize;
     this[kBuffer] = new BufferPool();
   }
 
+  get buffer(): BufferPool {
+    return this[kBuffer];
+  }
+
   override _write(chunk: Buffer, _: unknown, callback: Callback<Buffer>): void {
     this[kBuffer].append(chunk);
     processIncomingData(this, callback);
@@ -165,12 +171,20 @@ function processIncomingData(stream: MessageStream, callback: Callback<Buffer>)
   let ResponseType = messageHeader.opCode === OP_MSG ? BinMsg : Response;
   if (messageHeader.opCode !== OP_COMPRESSED) {
     const messageBody = message.slice(MESSAGE_HEADER_SIZE);
-    stream.emit('message', new ResponseType(message, messageHeader, messageBody));
 
-    if (buffer.length >= 4) {
+    // If we are a monitoring message stream using the streaming protocol and
+    // there is more in the buffer that can be read, skip processing since we
+    // want the last hello command response that is in the buffer.
+    if (stream.isStreamingProtocol && buffer.length >= 4) {
       processIncomingData(stream, callback);
     } else {
-      callback();
+      stream.emit('message', new ResponseType(message, messageHeader, messageBody));
+
+      if (buffer.length >= 4) {
+        processIncomingData(stream, callback);
+      } else {
+        callback();
+      }
     }
 
     return;
@@ -198,12 +212,19 @@ function processIncomingData(stream: MessageStream, callback: Callback<Buffer>)
       return;
     }
 
-    stream.emit('message', new ResponseType(message, messageHeader, messageBody));
-
-    if (buffer.length >= 4) {
+    // If we are a monitoring message stream using the streaming protocol and
+    // there is more in the buffer that can be read, skip processing since we
+    // want the last hello command response that is in the buffer.
+    if (stream.isStreamingProtocol && buffer.length >= 4) {
       processIncomingData(stream, callback);
     } else {
-      callback();
+      stream.emit('message', new ResponseType(message, messageHeader, messageBody));
+
+      if (buffer.length >= 4) {
+        processIncomingData(stream, callback);
+      } else {
+        callback();
+      }
     }
   });
 }

src/sdam/monitor.ts

@@ -279,6 +279,10 @@ function checkServer(monitor: Monitor, callback: Callback<Document | null>) {
       // if we are using the streaming protocol then we immediately issue another `started`
       // event, otherwise the "check" is complete and return to the main monitor loop
       if (isAwaitable && hello.topologyVersion) {
+        // Tell the connection that we are using the streaming protocol so that the
+        // connection's message stream will only read the last hello on the buffer.
+        connection.isStreamingProtocol = true;
+
         monitor.emit(
           Server.SERVER_HEARTBEAT_STARTED,
           new ServerHeartbeatStartedEvent(monitor.address)

test/unit/cmap/message_stream.test.js

@@ -19,6 +19,34 @@ function bufferToStream(buffer) {
 }
 
 describe('Message Stream', function () {
+  context('when the stream uses the streaming protocol', function () {
+    const data = Buffer.from(
+      '370000000100000001000000010000000000000000000000000000000000000001000000130000001069736d6173746572000100000000' +
+        '370000000100000001000000010000000000000000000000000000000000000001000000130000001069736d6173746572000100000000' +
+        '370000000100000001000000010000000000000000000000000000000000000001000000130000001069736d6173746572000100000000' +
+        '370000000100000001000000010000000000000000000000000000000000000001000000130000001069736d6173746572000100000000',
+      'hex'
+    );
+
+    it('only reads the last message in the buffer', function (done) {
+      const inputStream = bufferToStream(data);
+      const messageStream = new MessageStream();
+      messageStream.isStreamingProtocol = true;
+
+      messageStream.once('message', msg => {
+        msg.parse();
+        expect(msg)
+          .to.have.property('documents')
+          .that.deep.equals([{ [LEGACY_HELLO_COMMAND]: 1 }]);
+        // Make sure there is nothing left in the buffer.
+        expect(messageStream.buffer.length).to.equal(0);
+        done();
+      });
+
+      inputStream.pipe(messageStream);
+    });
+  });
+
   describe('reading', function () {
     [
       {