add generic ToIncrementalResult and ToPayload methods to Publisher

Author: yaacovCR

src/execution/execute.ts

@@ -506,7 +506,7 @@ export function buildExecutionContext(
     fieldResolver: fieldResolver ?? defaultFieldResolver,
     typeResolver: typeResolver ?? defaultTypeResolver,
     subscribeFieldResolver: subscribeFieldResolver ?? defaultFieldResolver,
-    publisher: new Publisher(),
+    publisher: new Publisher(toIncrementalResult, toPayload),
     errors: [],
   };
 }
@@ -518,7 +518,7 @@ function buildPerEventExecutionContext(
   return {
     ...exeContext,
     rootValue: payload,
-    publisher: new Publisher(),
+    publisher: new Publisher(toIncrementalResult, toPayload),
     errors: [],
   };
 }
@@ -2035,6 +2035,35 @@ async function executeStreamIterator(
   }
 }
 
+function toIncrementalResult(
+  asyncPayloadRecord: AsyncPayloadRecord,
+): IncrementalResult {
+  const incrementalResult: IncrementalResult = {};
+  if (isStreamPayload(asyncPayloadRecord)) {
+    const items = asyncPayloadRecord.items;
+    (incrementalResult as IncrementalStreamResult).items = items;
+  } else {
+    const data = asyncPayloadRecord.data;
+    (incrementalResult as IncrementalDeferResult).data = data ?? null;
+  }
+
+  incrementalResult.path = asyncPayloadRecord.path;
+  if (asyncPayloadRecord.label) {
+    incrementalResult.label = asyncPayloadRecord.label;
+  }
+  if (asyncPayloadRecord.errors.length > 0) {
+    incrementalResult.errors = asyncPayloadRecord.errors;
+  }
+  return incrementalResult;
+}
+
+function toPayload(
+  incremental: ReadonlyArray<IncrementalResult>,
+  hasNext: boolean,
+): SubsequentIncrementalExecutionResult {
+  return incremental.length ? { incremental, hasNext } : { hasNext };
+}
+
 function filterSubsequentPayloads(
   exeContext: ExecutionContext,
   nullPath: Path,
@@ -2161,3 +2190,9 @@ class StreamRecord {
 export type { StreamRecord };
 
 export type AsyncPayloadRecord = DeferredFragmentRecord | StreamRecord;
+
+function isStreamPayload(
+  asyncPayload: AsyncPayloadRecord,
+): asyncPayload is StreamRecord {
+  return asyncPayload.type === 'stream';
+}

src/execution/publisher.ts

@@ -1,20 +1,31 @@
 import type {
   AsyncPayloadRecord,
   StreamRecord,
-  IncrementalDeferResult,
   IncrementalResult,
-  IncrementalStreamResult,
   SubsequentIncrementalExecutionResult,
 } from './execute.js';
 
+type ToIncrementalResult = (
+  asyncPayloadRecord: AsyncPayloadRecord,
+) => IncrementalResult;
+
+type ToPayload = (
+  incremental: ReadonlyArray<IncrementalResult>,
+  hasNext: boolean,
+) => SubsequentIncrementalExecutionResult;
+
 /**
  * @internal
  */
 export class Publisher {
   subsequentPayloads: Set<AsyncPayloadRecord>;
+  toIncrementalResult: ToIncrementalResult;
+  toPayload: ToPayload;
 
-  constructor() {
+  constructor(toIncrementalResult: ToIncrementalResult, toPayload: ToPayload) {
     this.subsequentPayloads = new Set();
+    this.toIncrementalResult = toIncrementalResult;
+    this.toPayload = toPayload;
   }
 
   add(payload: AsyncPayloadRecord) {
@@ -39,31 +50,17 @@ export class Publisher {
   getCompletedIncrementalResults(): Array<IncrementalResult> {
     const incrementalResults: Array<IncrementalResult> = [];
     for (const asyncPayloadRecord of this.subsequentPayloads) {
-      const incrementalResult: IncrementalResult = {};
       if (!asyncPayloadRecord.isCompleted) {
         continue;
       }
       this.subsequentPayloads.delete(asyncPayloadRecord);
       if (isStreamPayload(asyncPayloadRecord)) {
-        const items = asyncPayloadRecord.items;
         if (asyncPayloadRecord.isCompletedIterator) {
           // async iterable resolver just finished but there may be pending payloads
           continue;
         }
-        (incrementalResult as IncrementalStreamResult).items = items;
-      } else {
-        const data = asyncPayloadRecord.data;
-        (incrementalResult as IncrementalDeferResult).data = data ?? null;
-      }
-
-      incrementalResult.path = asyncPayloadRecord.path;
-      if (asyncPayloadRecord.label) {
-        incrementalResult.label = asyncPayloadRecord.label;
-      }
-      if (asyncPayloadRecord.errors.length > 0) {
-        incrementalResult.errors = asyncPayloadRecord.errors;
       }
-      incrementalResults.push(incrementalResult);
+      incrementalResults.push(this.toIncrementalResult(asyncPayloadRecord));
     }
     return incrementalResults;
   }
@@ -104,7 +101,7 @@ export class Publisher {
       }
 
       return {
-        value: incremental.length ? { incremental, hasNext } : { hasNext },
+        value: publisher.toPayload(incremental, hasNext),
         done: false,
       };
     }