condense loop

Author: yaacovCR

src/execution/Queue.ts

@@ -50,42 +50,35 @@ export class Queue<T> {
   private async *subscribeImpl<U>(
     mapFn: (generator: Generator<T, void, void>) => U | undefined,
   ): AsyncGenerator<U> {
-    if (this._stopped) {
-      return;
-    }
-
-    while (this._items.length > 0) {
-      const maybe = mapFn(this.batch());
-      if (maybe === undefined) {
-        continue;
-      }
-      yield maybe;
+    while (true) {
       if (this._stopped) {
         return;
       }
-    }
 
-    let batch: Generator<T> | undefined;
-    // eslint-disable-next-line no-await-in-loop
-    while ((batch = await this._nextBatch()) !== undefined) {
-      let maybe = mapFn(batch);
-      if (maybe === undefined) {
-        continue;
-      }
-      yield maybe;
-      if (this._stopped) {
-        return;
-      }
+      // drain any items pushed prior to or between .next() calls
       while (this._items.length > 0) {
-        maybe = mapFn(this.batch());
-        if (maybe === undefined) {
-          continue;
+        const mapped = mapFn(this.batch());
+        if (mapped === undefined) {
+          break;
         }
-        yield maybe;
+        yield mapped;
         if (this._stopped) {
           return;
         }
       }
+
+      // wait for a yield-able batch
+      let nextMapped;
+      do {
+        // eslint-disable-next-line no-await-in-loop
+        const nextBatch = await this._nextBatch();
+        if (nextBatch === undefined || this._stopped) {
+          return;
+        }
+        nextMapped = mapFn(nextBatch);
+      } while (nextMapped === undefined);
+
+      yield nextMapped;
     }
   }
 

src/execution/__tests__/Queue-test.ts

@@ -135,33 +135,6 @@ describe('Queue', () => {
     expect(await sub.next()).to.deep.equal({ done: false, value: [6] });
   });
 
-  it('should skip payloads when mapped to undefined, skipping second async payload', async () => {
-    const queue = new Queue<number>(async (push) => {
-      await resolveOnNextTick();
-      push(0);
-      await resolveOnNextTick();
-      push(1);
-      await resolveOnNextTick();
-      push(2);
-      await resolveOnNextTick();
-      push(3);
-      await resolveOnNextTick();
-      push(4);
-    });
-
-    const sub = queue.subscribe((batch) => {
-      const arr = Array.from(batch);
-      if (arr[0] % 2 === 0) {
-        return arr;
-      }
-    });
-    expect(await sub.next()).to.deep.equal({ done: false, value: [0] });
-    // [1, 2, 3] are batched as we await 0
-    // - one tick for the [AsyncGeneratorResumeNext] job
-    // - one tick for the await within the withCleanUp next()
-    expect(await sub.next()).to.deep.equal({ done: false, value: [4] });
-  });
-
   it('should condense pushes during map into the same batch', async () => {
     let push!: (item: number) => void;
     const queue = new Queue<number>((_push) => {