wip

Author: nbbeeken

.eslintrc.json

@@ -131,10 +131,6 @@
       {
         "selector": "CallExpression[callee.name='clearTimeout']",
         "message": "clearTimeout must remove abort listener"
-      },
-      {
-        "selector": "CallExpression[callee.property.name='removeAllListeners'][arguments.length=0]",
-        "message": "removeAllListeners can remove error listeners leading to uncaught errors"
       }
     ],
     "@typescript-eslint/no-unused-vars": "error",

src/client-side-encryption/state_machine.ts

@@ -19,7 +19,6 @@ import { type Abortable } from '../mongo_types';
 import { Timeout, type TimeoutContext, TimeoutError } from '../timeout';
 import {
   addAbortListener,
-  addAbortSignalToStream,
   BufferPool,
   kDispose,
   MongoDBCollectionNamespace,
@@ -186,15 +185,10 @@ export type StateMachineOptions = {
  */
 // TODO(DRIVERS-2671): clarify CSOT behavior for FLE APIs
 export class StateMachine {
-  closeSignal: AbortSignal;
-
   constructor(
     private options: StateMachineOptions,
-    private bsonOptions = pluckBSONSerializeOptions(options),
-    closeSignal: AbortSignal
-  ) {
-    this.closeSignal = closeSignal;
-  }
+    private bsonOptions = pluckBSONSerializeOptions(options)
+  ) {}
 
   /**
    * Executes the state machine according to the specification
@@ -456,7 +450,7 @@ export class StateMachine {
       await (options?.timeoutContext?.csotEnabled()
         ? Promise.all([
             willResolveKmsRequest,
-            Timeout.expires(options.timeoutContext.remainingTimeMS, this.closeSignal)
+            Timeout.expires(options.timeoutContext?.remainingTimeMS)
           ])
         : willResolveKmsRequest);
     } catch (error) {

src/cmap/connect.ts

@@ -16,7 +16,7 @@ import {
   MongoRuntimeError,
   needsRetryableWriteLabel
 } from '../error';
-import { addAbortSignalToStream, HostAddress, ns, promiseWithResolvers } from '../utils';
+import { HostAddress, ns, promiseWithResolvers } from '../utils';
 import { AuthContext } from './auth/auth_provider';
 import { AuthMechanism } from './auth/providers';
 import {
@@ -387,8 +387,6 @@ export async function makeSocket(
     socket = net.createConnection(parseConnectOptions(options));
   }
 
-  addAbortSignalToStream(closeSignal, socket);
-
   socket.unref();
   socket.setKeepAlive(true, 300000);
   socket.setTimeout(connectTimeoutMS);

src/mongo_client.ts

@@ -704,6 +704,7 @@ export class MongoClient extends TypedEventEmitter<MongoClientEvents> implements
         // If maxPoolSize is 1 we won't be able to run anything
         // unless we interrupt whatever is using the one connection.
         this.closeController.abort(new MongoClientClosedError());
+        this.closeController = new AbortController();
       }
 
       const activeCursorCloses = Array.from(this.s.activeCursors, cursor => cursor.close());

src/utils.ts

@@ -1555,24 +1555,3 @@ export async function abortable<T>(
     abortListener?.[kDispose]();
   }
 }
-
-export function addAbortSignalToStream(
-  signal: AbortSignal | undefined,
-  stream: Writable | Readable
-) {
-  if (signal == null) {
-    return;
-  }
-
-  if (signal.aborted) {
-    stream.destroy(signal.reason);
-    return;
-  }
-
-  const abortListener = addAbortListener(signal, function () {
-    stream.off('close', abortListener[kDispose]).off('error', abortListener[kDispose]);
-    if (!stream.destroyed) stream.destroy(this.reason);
-  });
-  // not nearly as complex as node's eos() but... do we need all that?? sobbing emoji.
-  stream.once('close', abortListener[kDispose]).once('error', abortListener[kDispose]);
-}

test/integration/sessions/sessions.test.ts

@@ -70,19 +70,15 @@ describe('Sessions Spec', function () {
         await test.setup(this.configuration);
       });
 
-      it('should send endSessions for multiple sessions', function (done) {
+      it('should send endSessions for multiple sessions', async function () {
         const client = test.client;
         const sessions = [client.startSession(), client.startSession()].map(s => s.id);
 
-        client.close(err => {
-          expect(err).to.not.exist;
-          expect(test.commands.started).to.have.length(1);
-          expect(test.commands.started[0].commandName).to.equal('endSessions');
-          expect(test.commands.started[0].command.endSessions).to.include.deep.members(sessions);
-          expect(client.s.activeSessions.size).to.equal(0);
-
-          done();
-        });
+        await client.close();
+        expect(test.commands.started).to.have.lengthOf(1);
+        expect(test.commands.started[0].commandName).to.equal('endSessions');
+        expect(test.commands.started[0].command.endSessions).to.include.deep.members(sessions);
+        expect(client.s.activeSessions.size).to.equal(0);
       });
     });