don't await at all if there are ready payloads

Author: yaacovCR

src/execution/__tests__/stream-test.ts

@@ -555,6 +555,9 @@ describe('Execute: stream directive', () => {
             path: ['friendList', 2],
           },
         ],
+        hasNext: true,
+      },
+      {
         hasNext: false,
       },
     ]);
@@ -594,7 +597,7 @@ describe('Execute: stream directive', () => {
         }
       }
     `);
-    const result = await completeAsync(document, 2, {
+    const result = await completeAsync(document, 3, {
       async *friendList() {
         yield await Promise.resolve(friends[0]);
         yield await Promise.resolve(friends[1]);
@@ -623,6 +626,12 @@ describe('Execute: stream directive', () => {
               path: ['friendList', 2],
             },
           ],
+          hasNext: true,
+        },
+      },
+      {
+        done: false,
+        value: {
           hasNext: false,
         },
       },

src/execution/publisher.ts

@@ -168,21 +168,39 @@ export class Publisher<TSource extends Source, TIncremental, TPayload> {
         return { value: undefined, done: true };
       }
 
+      const incremental = this._getCompletedIncrementalResults();
+      if (!incremental.length) {
+        return onSignal();
+      }
+
+      const hasNext = this.hasNext();
+
+      if (!hasNext) {
+        isDone = true;
+      }
+
+      return {
+        value: this.toPayload(incremental, hasNext),
+        done: false,
+      };
+    };
+
+    const onSignal = async (): Promise<IteratorResult<TPayload, void>> => {
       await this.signal;
 
       if (isDone) {
         return { value: undefined, done: true };
       }
 
       const incremental = this._getCompletedIncrementalResults();
-      const hasNext = this.hasNext();
 
       this.signal = new Promise((resolve) => {
         this.trigger = resolve;
       });
 
+      const hasNext = this.hasNext();
       if (!incremental.length && hasNext) {
-        return next();
+        return onSignal();
       }
 
       if (!hasNext) {