consolidate payloads

Author: yaacovCR

src/execution/IncrementalPublisher.ts

@@ -109,6 +109,12 @@ export interface FormattedCompletedResult {
   errors?: ReadonlyArray<GraphQLError>;
 }
 
+interface IncrementalAggregate {
+  newPendingSources: Set<DeferredFragmentRecord | StreamRecord>;
+  incrementalResults: Array<IncrementalResult>;
+  completedResults: Array<CompletedResult>;
+}
+
 /**
  * This class is used to publish incremental results to the client, enabling semi-concurrent
  * execution while preserving result order.
@@ -179,20 +185,28 @@ export class IncrementalPublisher {
           return { value: undefined, done: true };
         }
 
-        for (const item of this._released) {
-          this._pending.delete(item);
-        }
-        const released = this._released;
-        this._released = new Set();
+        if (this._released.size > 0) {
+          let aggregate = this._incrementalInitializer();
+          do {
+            for (const item of this._released) {
+              this._pending.delete(item);
+            }
+            const released = this._released;
+            this._released = new Set();
 
-        const result = this._getIncrementalResult(released);
+            aggregate = this._incrementalReducer(aggregate, released);
+          } while (this._released.size > 0);
 
-        if (!this.hasNext()) {
-          isDone = true;
-        }
+          const hasNext = this.hasNext();
+
+          if (!hasNext) {
+            isDone = true;
+          }
 
-        if (result !== undefined) {
-          return { value: result, done: false };
+          return {
+            value: this._incrementalFinalizer(aggregate),
+            done: false,
+          };
         }
 
         // eslint-disable-next-line no-await-in-loop
@@ -532,37 +546,20 @@ export class IncrementalPublisher {
     this._trigger();
   }
 
-  private _getIncrementalResult(
-    completedRecords: ReadonlySet<SubsequentResultRecord>,
-  ): SubsequentIncrementalExecutionResult | undefined {
-    const { pending, incremental, completed } =
-      this._processPending(completedRecords);
-
-    const hasNext = this.hasNext();
-    if (incremental.length === 0 && completed.length === 0 && hasNext) {
-      return undefined;
-    }
-
-    const result: SubsequentIncrementalExecutionResult = { hasNext };
-    if (pending.length) {
-      result.pending = pending;
-    }
-    if (incremental.length) {
-      result.incremental = incremental;
-    }
-    if (completed.length) {
-      result.completed = completed;
-    }
-
-    return result;
+  private _incrementalInitializer(): IncrementalAggregate {
+    return {
+      newPendingSources: new Set<DeferredFragmentRecord | StreamRecord>(),
+      incrementalResults: [],
+      completedResults: [],
+    };
   }
 
-  private _processPending(
+  private _incrementalReducer(
+    aggregate: IncrementalAggregate,
     completedRecords: ReadonlySet<SubsequentResultRecord>,
-  ): IncrementalUpdate {
-    const newPendingSources = new Set<DeferredFragmentRecord | StreamRecord>();
-    const incrementalResults: Array<IncrementalResult> = [];
-    const completedResults: Array<CompletedResult> = [];
+  ): IncrementalAggregate {
+    const { newPendingSources, incrementalResults, completedResults } =
+      aggregate;
     for (const subsequentResultRecord of completedRecords) {
       for (const child of subsequentResultRecord.children) {
         const pendingSource = isStreamItemsRecord(child)
@@ -625,11 +622,30 @@ export class IncrementalPublisher {
       }
     }
 
-    return {
-      pending: this.pendingSourcesToResults(newPendingSources),
-      incremental: incrementalResults,
-      completed: completedResults,
+    return aggregate;
+  }
+
+  private _incrementalFinalizer(
+    aggregate: IncrementalAggregate,
+  ): SubsequentIncrementalExecutionResult {
+    const { newPendingSources, incrementalResults, completedResults } =
+      aggregate;
+    const pendingResults = this.pendingSourcesToResults(newPendingSources);
+
+    const result: SubsequentIncrementalExecutionResult = {
+      hasNext: this.hasNext(),
     };
+    if (pendingResults.length) {
+      result.pending = pendingResults;
+    }
+    if (incrementalResults.length) {
+      result.incremental = incrementalResults;
+    }
+    if (completedResults.length) {
+      result.completed = completedResults;
+    }
+
+    return result;
   }
 
   private _completedRecordToResult(

src/execution/__tests__/defer-test.ts

@@ -1000,35 +1000,25 @@ describe('Execute: defer directive', () => {
         hasNext: true,
       },
       {
-        pending: [{ path: ['hero', 'nestedObject'] }],
         incremental: [
           {
             data: { bar: 'bar' },
             path: ['hero', 'nestedObject', 'deeperObject'],
           },
-        ],
-        completed: [{ path: ['hero'] }],
-        hasNext: true,
-      },
-      {
-        pending: [{ path: ['hero', 'nestedObject', 'deeperObject'] }],
-        incremental: [
           {
             data: { baz: 'baz' },
             path: ['hero', 'nestedObject', 'deeperObject'],
           },
-        ],
-        hasNext: true,
-        completed: [{ path: ['hero', 'nestedObject'] }],
-      },
-      {
-        incremental: [
           {
             data: { bak: 'bak' },
             path: ['hero', 'nestedObject', 'deeperObject'],
           },
         ],
-        completed: [{ path: ['hero', 'nestedObject', 'deeperObject'] }],
+        completed: [
+          { path: ['hero'] },
+          { path: ['hero', 'nestedObject'] },
+          { path: ['hero', 'nestedObject', 'deeperObject'] },
+        ],
         hasNext: false,
       },
     ]);
@@ -1075,31 +1065,25 @@ describe('Execute: defer directive', () => {
         hasNext: true,
       },
       {
-        pending: [{ path: ['hero', 'nestedObject', 'deeperObject'] }],
         incremental: [
           {
             data: {
               foo: 'foo',
             },
             path: ['hero', 'nestedObject', 'deeperObject'],
           },
-        ],
-        completed: [
-          { path: ['hero'] },
-          { path: ['hero', 'nestedObject', 'deeperObject'] },
-        ],
-        hasNext: true,
-      },
-      {
-        incremental: [
           {
             data: {
               bar: 'bar',
             },
             path: ['hero', 'nestedObject', 'deeperObject'],
           },
         ],
-        completed: [{ path: ['hero', 'nestedObject', 'deeperObject'] }],
+        completed: [
+          { path: ['hero'] },
+          { path: ['hero', 'nestedObject', 'deeperObject'] },
+          { path: ['hero', 'nestedObject', 'deeperObject'] },
+        ],
         hasNext: false,
       },
     ]);
@@ -1895,27 +1879,17 @@ describe('Execute: defer directive', () => {
         hasNext: true,
       },
       {
-        pending: [
-          { path: ['hero', 'friends', 0] },
-          { path: ['hero', 'friends', 1] },
-          { path: ['hero', 'friends', 2] },
-        ],
         incremental: [
           {
             data: { name: 'slow', friends: [{}, {}, {}] },
             path: ['hero'],
           },
-        ],
-        completed: [{ path: ['hero'] }],
-        hasNext: true,
-      },
-      {
-        incremental: [
           { data: { name: 'Han' }, path: ['hero', 'friends', 0] },
           { data: { name: 'Leia' }, path: ['hero', 'friends', 1] },
           { data: { name: 'C-3PO' }, path: ['hero', 'friends', 2] },
         ],
         completed: [
+          { path: ['hero'] },
           { path: ['hero', 'friends', 0] },
           { path: ['hero', 'friends', 1] },
           { path: ['hero', 'friends', 2] },