fix multiple transform bug

Author: baileympearson

src/cursor/abstract_cursor.ts

@@ -361,7 +361,7 @@ export abstract class AbstractCursor<
       return true;
     }
 
-    const doc = await next<TSchema>(this, true);
+    const doc = await next<TSchema>(this, true, false);
 
     if (doc) {
       this[kDocuments].unshift(doc);
@@ -680,48 +680,47 @@ export abstract class AbstractCursor<
   }
 }
 
-function nextDocument<T>(cursor: AbstractCursor<T>): T | null {
-  const doc = cursor[kDocuments].shift();
-
-  if (doc && cursor[kTransform]) {
-    return cursor[kTransform](doc) as T;
-  }
-
-  return doc;
-}
-
 /**
  * @param cursor - the cursor on which to call `next`
  * @param blocking - a boolean indicating whether or not the cursor should `block` until data
  *     is available.  Generally, this flag is set to `false` because if the getMore returns no documents,
  *     the cursor has been exhausted.  In certain scenarios (ChangeStreams, tailable await cursors and
  *     `tryNext`, for example) blocking is necessary because a getMore returning no documents does
  *     not indicate the end of the cursor.
+ * @param transform - if true, the cursor's transform function is applied to the result document (if the transform exists)
  * @returns the next document in the cursor, or `null`.  When `blocking` is `true`, a `null` document means
  * the cursor has been exhausted.  Otherwise, it means that there is no document available in the cursor's buffer.
  */
-async function next<T>(cursor: AbstractCursor<T>, blocking: boolean): Promise<T | null> {
+async function next<T>(
+  cursor: AbstractCursor<T>,
+  blocking: boolean,
+  transform = true
+): Promise<T | null> {
   const cursorId = cursor[kId];
   if (cursor.closed) {
     return null;
   }
 
   if (cursor[kDocuments].length !== 0) {
-    return nextDocument<T>(cursor);
+    const doc = cursor[kDocuments].shift();
+
+    if (doc != null && transform && cursor[kTransform]) {
+      return cursor[kTransform](doc);
+    }
+
+    return doc;
   }
 
   if (cursorId == null) {
     // All cursors must operate within a session, one must be made implicitly if not explicitly provided
     const init = promisify(cb => cursor[kInit](cb));
     await init();
-    return next(cursor, blocking);
+    return next(cursor, blocking, transform);
   }
 
   if (cursorIsDead(cursor)) {
-    try {
-      await cleanupCursorAsync(cursor, undefined);
-      // eslint-disable-next-line no-empty
-    } catch {}
+    // if the cursor is dead, we clean it up
+    await cleanupCursorAsync(cursor);
     return null;
   }
 
@@ -735,11 +734,8 @@ async function next<T>(cursor: AbstractCursor<T>, blocking: boolean): Promise<T
   try {
     response = await getMore(batchSize);
   } catch (error) {
-    if (error || cursorIsDead(cursor)) {
-      try {
-        await cleanupCursorAsync(cursor, { error });
-        // eslint-disable-next-line no-empty
-      } catch {}
+    if (error) {
+      await cleanupCursorAsync(cursor, { error });
       throw error;
     }
   }
@@ -756,19 +752,40 @@ async function next<T>(cursor: AbstractCursor<T>, blocking: boolean): Promise<T
     cursor[kId] = cursorId;
   }
 
+  if (cursorIsDead(cursor)) {
+    // If we successfully received a response from a cursor BUT the cursor indicates that it is exhausted,
+    // we intentionally clean up the cursor to release its session back into the pool before the cursor
+    // is iterated.  This prevents a cursor that is exhausted on the server from holding
+    // onto a session indefinitely until the AbstractCursor is iterated.
+    await cleanupCursorAsync(cursor);
+  }
+
   if (cursor[kDocuments].length === 0 && blocking === false) {
     return null;
   }
 
-  return next(cursor, blocking);
+  return next(cursor, blocking, transform);
 }
 
 function cursorIsDead(cursor: AbstractCursor): boolean {
   const cursorId = cursor[kId];
   return !!cursorId && cursorId.isZero();
 }
 
-const cleanupCursorAsync = promisify(cleanupCursor);
+const cleanupCursorAsyncInternal = promisify(cleanupCursor);
+
+async function cleanupCursorAsync<T>(
+  cursor: AbstractCursor<T>,
+  options: { needsToEmitClosed?: boolean; error?: AnyError } = {}
+): Promise<void> {
+  try {
+    await cleanupCursorAsyncInternal(cursor, options);
+  } catch {
+    // `cleanupCursor` never throws but we can't really test that.
+    // so this is a hack to ensure that any upstream consumers
+    // can safely guarantee on this wrapper never throwing.
+  }
+}
 
 function cleanupCursor(
   cursor: AbstractCursor,

test/integration/crud/misc_cursors.test.js

@@ -1708,7 +1708,7 @@ describe('Cursor', function () {
     expect(cursor).property('closed', false);
 
     const willClose = once(cursor, 'close');
-    const willEnd = once(stream, 'end');
+    const willEnd = once(stream, 'close');
 
     const dataEvents = on(stream, 'data');
 
@@ -1722,16 +1722,16 @@ describe('Cursor', function () {
     // After 5 successful data events, destroy stream
     stream.destroy();
 
-    // We should get an end event on the stream and a close event on the cursor
+    // We should get a a close event on the stream and a close event on the cursor
     // We should **not** get an 'error' event,
     // the following will throw if either stream or cursor emitted an 'error' event
     await Promise.race([
       willEnd,
-      sleep(100).then(() => Promise.reject(new Error('end event never emitted')))
+      sleep(100, { ref: false }).then(() => Promise.reject(new Error('end event never emitted')))
     ]);
     await Promise.race([
       willClose,
-      sleep(100).then(() => Promise.reject(new Error('close event never emitted')))
+      sleep(100, { ref: false }).then(() => Promise.reject(new Error('close event never emitted')))
     ]);
   });
 
@@ -3589,7 +3589,7 @@ describe('Cursor', function () {
     await client.close();
   });
 
-  it('should return implicit session to pool when client-side cursor exhausts results after a getMore', function (done) {
+  it('should return implicit session to pool when client-side cursor exhausts results after a getMore', async function () {
     const configuration = this.configuration;
     const client = configuration.newClient({ w: 1 }, { maxPoolSize: 1 });
 
@@ -3604,25 +3604,22 @@ describe('Cursor', function () {
       { a: 9, b: 10 }
     ];
 
-    collection.insertMany(docs, err => {
-      expect(err).to.not.exist;
-      const cursor = collection.find({}, { batchSize: 3 });
-      cursor.next(function () {
-        expect(client.s.activeSessions.size).to.equal(1);
-        cursor.next(function () {
-          expect(client.s.activeSessions.size).to.equal(1);
-          cursor.next(function () {
-            expect(client.s.activeSessions.size).to.equal(1);
-            cursor.next(function () {
-              expect(client.s.activeSessions.size).to.equal(0);
-              cursor.close(() => {
-                client.close(done);
-              });
-            });
-          });
-        });
-      });
-    });
+    await collection.insertMany(docs);
+
+    // TODO - talk to Neal about this test
+    const cursor = await collection.find({}, { batchSize: 3 });
+    for (let i = 0; i < 3; ++i) {
+      await cursor.next();
+      expect(client.s.activeSessions.size).to.equal(1);
+    }
+
+    await cursor.next();
+    expect(client.s.activeSessions.size, 'session not checked in after cursor exhausted').to.equal(
+      0
+    );
+
+    await cursor.close();
+    await client.close();
   });
 
   describe('#clone', function () {

test/integration/node-specific/cursor_stream.test.js

@@ -297,10 +297,10 @@ describe('Cursor Streams', function () {
         stream.on('error', err => (error = err));
         cursor.on('close', function () {
           // NOTE: use `setImmediate` here because the stream implementation uses `nextTick` to emit the error
-          setImmediate(() => {
+          setTimeout(() => {
             expect(error).to.exist;
             client.close(done);
-          });
+          }, 50);
         });
 
         stream.pipe(process.stdout);