test: add integration test

Author: nbbeeken

test/integration/connection-monitoring-and-pooling/connection.test.ts

@@ -22,7 +22,7 @@ import {
 } from '../../mongodb';
 import * as mock from '../../tools/mongodb-mock/index';
 import { skipBrokenAuthTestBeforeEachHook } from '../../tools/runner/hooks/configuration';
-import { sleep } from '../../tools/utils';
+import { processTick, sleep } from '../../tools/utils';
 import { assert as test, setupDatabase } from '../shared';
 
 const commonConnectOptions = {
@@ -249,6 +249,55 @@ describe('Connection', function () {
       client.connect();
     });
 
+    describe(
+      'when a monitoring Connection receives many hellos in one chunk',
+      { requires: { topology: 'replicaset' } },
+      function () {
+        let client: MongoClient;
+        let hbSuccess = 0;
+
+        beforeEach(async function () {
+          client = this.configuration.newClient({}, { heartbeatFrequencyMS: 100 }); // just so we don't have to wait so long for a hello
+          hbSuccess = 0;
+          client.on('serverHeartbeatSucceeded', () => (hbSuccess += 1));
+        });
+
+        afterEach(async function () {
+          hbSuccess = 0;
+          await client.close();
+        });
+
+        // In the future we may want to skip processing concatenated heartbeats.
+        // This test exists to prevent regression of processing many messages inside one chunk.
+        it(
+          'processes all of them and emits heartbeats',
+          { requires: { topology: 'replicaset' } },
+          async function () {
+            expect(hbSuccess).to.equal(0);
+
+            await client.db().command({ ping: 1 }); // start monitoring.
+            const monitor = [...client.topology.s.servers.values()][0].monitor;
+
+            // @ts-expect-error: accessing private property
+            const messageStream = monitor.connection.messageStream;
+            // @ts-expect-error: accessing private property
+            const socket = monitor.connection.socket;
+
+            const [hello] = (await once(messageStream, 'data')) as [Buffer];
+
+            const thousandHellos = Array.from({ length: 1000 }, () => [...hello]).flat(1);
+
+            // pretend this came from the server
+            socket.emit('data', Buffer.from(thousandHellos));
+
+            // All of the hb will be emitted synchronously in the next tick as the entire chunk is processed.
+            await processTick();
+            expect(hbSuccess).to.be.greaterThan(100);
+          }
+        );
+      }
+    );
+
     context(
       'when a large message is written to the socket',
       { requires: { topology: 'single', auth: 'disabled' } },