add pending notifications

Author: yaacovCR

src/execution/IncrementalPublisher.ts

@@ -17,6 +17,7 @@ import type {
 import type { StreamUsage } from './execute.js';
 
 interface IncrementalUpdate<TData = unknown, TExtensions = ObjMap<unknown>> {
+  pending: ReadonlyArray<PendingResult>;
   incremental: ReadonlyArray<IncrementalResult<TData, TExtensions>>;
   completed: ReadonlyArray<CompletedResult>;
 }
@@ -34,6 +35,7 @@ export interface FormattedSubsequentIncrementalExecutionResult<
   TExtensions = ObjMap<unknown>,
 > {
   hasNext: boolean;
+  pending?: ReadonlyArray<PendingResult>;
   incremental?: ReadonlyArray<FormattedIncrementalResult<TData, TExtensions>>;
   completed?: ReadonlyArray<FormattedCompletedResult>;
   extensions?: TExtensions;
@@ -90,6 +92,11 @@ export type FormattedIncrementalResult<
   | FormattedIncrementalDeferResult<TData, TExtensions>
   | FormattedIncrementalStreamResult<TData, TExtensions>;
 
+export interface PendingResult {
+  path: ReadonlyArray<string | number>;
+  label?: string;
+}
+
 export interface CompletedResult {
   path: ReadonlyArray<string | number>;
   label?: string;
@@ -148,6 +155,10 @@ export class IncrementalPublisher {
     this._reset();
   }
 
+  getPending(): ReadonlySet<SubsequentResultRecord> {
+    return this._pending;
+  }
+
   hasNext(): boolean {
     return this._pending.size > 0;
   }
@@ -419,6 +430,23 @@ export class IncrementalPublisher {
     incrementalDataRecord.errors.push(error);
   }
 
+  pendingSourcesToResults(
+    pendingSources: ReadonlySet<DeferredFragmentRecord | StreamRecord>,
+  ): Array<PendingResult> {
+    const pendingResults: Array<PendingResult> = [];
+    for (const pendingSource of pendingSources) {
+      pendingSource.pendingSent = true;
+      const pendingResult: PendingResult = {
+        path: pendingSource.path,
+      };
+      if (pendingSource.label !== undefined) {
+        pendingResult.label = pendingSource.label;
+      }
+      pendingResults.push(pendingResult);
+    }
+    return pendingResults;
+  }
+
   publishInitial(): void {
     this._initialResult.isCompleted = true;
     for (const child of this._initialResult.children) {
@@ -507,14 +535,18 @@ export class IncrementalPublisher {
   private _getIncrementalResult(
     completedRecords: ReadonlySet<SubsequentResultRecord>,
   ): SubsequentIncrementalExecutionResult | undefined {
-    const { incremental, completed } = this._processPending(completedRecords);
+    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;
     }
@@ -528,16 +560,24 @@ export class IncrementalPublisher {
   private _processPending(
     completedRecords: ReadonlySet<SubsequentResultRecord>,
   ): IncrementalUpdate {
+    const newPendingSources = new Set<DeferredFragmentRecord | StreamRecord>();
     const incrementalResults: Array<IncrementalResult> = [];
     const completedResults: Array<CompletedResult> = [];
     for (const subsequentResultRecord of completedRecords) {
       for (const child of subsequentResultRecord.children) {
+        const pendingSource = isStreamItemsRecord(child)
+          ? child.streamRecord
+          : child;
+        if (!pendingSource.pendingSent) {
+          newPendingSources.add(pendingSource);
+        }
         this._publish(child);
       }
       if (isStreamItemsRecord(subsequentResultRecord)) {
         if (!subsequentResultRecord.sent) {
           subsequentResultRecord.sent = true;
           if (subsequentResultRecord.isFinalRecord) {
+            newPendingSources.delete(subsequentResultRecord.streamRecord);
             completedResults.push(
               this._completedRecordToResult(
                 subsequentResultRecord.streamRecord,
@@ -561,6 +601,7 @@ export class IncrementalPublisher {
           incrementalResults.push(incrementalResult);
         }
       } else {
+        newPendingSources.delete(subsequentResultRecord);
         completedResults.push(
           this._completedRecordToResult(subsequentResultRecord),
         );
@@ -585,6 +626,7 @@ export class IncrementalPublisher {
     }
 
     return {
+      pending: this.pendingSourcesToResults(newPendingSources),
       incremental: incrementalResults,
       completed: completedResults,
     };
@@ -738,6 +780,7 @@ export class DeferredFragmentRecord {
   deferredGroupedFieldSetRecords: Set<DeferredGroupedFieldSetRecord>;
   errors: Array<GraphQLError>;
   isCompleted: boolean;
+  pendingSent?: boolean;
   _pending: Set<DeferredGroupedFieldSetRecord>;
 
   constructor(opts: {
@@ -762,6 +805,7 @@ export class StreamRecord {
   path: ReadonlyArray<string | number>;
   errors: Array<GraphQLError>;
   asyncIterator?: AsyncIterator<unknown> | undefined;
+  pendingSent?: boolean;
   constructor(opts: {
     label: string | undefined;
     path: Path;
@@ -807,7 +851,7 @@ export type IncrementalDataRecord =
   | DeferredGroupedFieldSetRecord
   | StreamItemsRecord;
 
-type SubsequentResultRecord = DeferredFragmentRecord | StreamItemsRecord;
+export type SubsequentResultRecord = DeferredFragmentRecord | StreamItemsRecord;
 
 function getSubsequentResultRecords(
   incrementalDataRecord: IncrementalDataRecord | undefined,
@@ -823,7 +867,7 @@ function getSubsequentResultRecords(
   return incrementalDataRecord.deferredFragmentRecords;
 }
 
-function isStreamItemsRecord(
+export function isStreamItemsRecord(
   subsequentResultRecord: unknown,
 ): subsequentResultRecord is StreamItemsRecord {
   return subsequentResultRecord instanceof StreamItemsRecord;