feat: add asyncIterator

Author: andymina

src/change_stream.ts

@@ -727,6 +727,28 @@ export class ChangeStream<
     }, callback);
   }
 
+  [Symbol.asyncIterator](): AsyncIterator<TChange, void> {
+    async function* nativeAsyncIterator(this: ChangeStream<TSchema, TChange>) {
+      if (this.closed) {
+        return;
+      }
+
+      while (true) {
+        if (!(await this.hasNext())) {
+          break;
+        }
+
+        yield await this.next();
+      }
+    }
+
+    const iterator = nativeAsyncIterator.call(this);
+
+    return {
+      next: () => iterator.next()
+    };
+  }
+
   /** Is the cursor closed */
   get closed(): boolean {
     return this[kClosed] || this.cursor.closed;

test/integration/change-streams/change_stream.test.ts

@@ -10,6 +10,9 @@ import { promisify } from 'util';
 import {
   AbstractCursor,
   ChangeStream,
+  ChangeStreamCursor,
+  ChangeStreamDocument,
+  ChangeStreamInsertDocument,
   ChangeStreamOptions,
   Collection,
   CommandStartedEvent,
@@ -1737,6 +1740,7 @@ describe('ChangeStream resumability', function () {
     aggregateEvents = [];
   });
 
+  // TODO(andymina): resumable error tests here
   context('iterator api', function () {
     context('#next', function () {
       for (const { error, code, message } of resumableErrorCodes) {
@@ -2138,6 +2142,35 @@ describe('ChangeStream resumability', function () {
         );
       });
     });
+
+    context.only('#asyncIterator', function () {
+      /**
+       * TODO(andymina): three test cases to cover
+       * 
+       * happy path - asyncIterable works
+       * unhappy path - it errors out
+       * resumable error - continues but also throws the error out
+       */
+      // happy path
+      it('happy path', async function () {
+        changeStream = collection.watch([]);
+        await initIteratorMode(changeStream);
+
+        const docs = [{ city: 'New York City' }, { city: 'Seattle' }, { city: 'Boston' }];
+        await collection.insertMany(docs);
+
+        let count = 0;
+        for await (const change of changeStream) {
+          const { fullDocument } = change;
+          expect(fullDocument.city).to.equal(docs[count].city);
+
+          count++;
+          if (count === 3) {
+            changeStream.close();
+          }
+        }
+      });
+    });
   });
 
   describe('event emitter based iteration', function () {