incremental publisher should handle all response building (graphql#3930)

extracted from graphql#3929 the publisher itself can determine whether to return a single result or the initial result + stream the only desired change is to replace the following code block with the below: [FROM:](https://github.com/graphql/graphql-js/blob/fae5da500bad94c39a7ecd77a4c4361b58d6d2da/src/execution/execute.ts#L293-L340) ```ts const incrementalPublisher = exeContext.incrementalPublisher; const initialResultRecord = incrementalPublisher.prepareInitialResultRecord(); try { const result = executeOperation(exeContext, initialResultRecord); if (isPromise(result)) { return result.then( (data) => { const errors = incrementalPublisher.getInitialErrors(initialResultRecord); const initialResult = buildResponse(data, errors); incrementalPublisher.publishInitial(initialResultRecord); if (incrementalPublisher.hasNext()) { return { initialResult: { ...initialResult, hasNext: true, }, subsequentResults: incrementalPublisher.subscribe(), }; } return initialResult; }, (error) => { incrementalPublisher.addFieldError(initialResultRecord, error); const errors = incrementalPublisher.getInitialErrors(initialResultRecord); return buildResponse(null, errors); }, ); } const initialResult = buildResponse(result, initialResultRecord.errors); incrementalPublisher.publishInitial(initialResultRecord); if (incrementalPublisher.hasNext()) { return { initialResult: { ...initialResult, hasNext: true, }, subsequentResults: incrementalPublisher.subscribe(), }; } return initialResult; } catch (error) { incrementalPublisher.addFieldError(initialResultRecord, error); const errors = incrementalPublisher.getInitialErrors(initialResultRecord); return buildResponse(null, errors); } } ``` [TO:](https://github.com/yaacovCR/graphql-executor/blob/598608e8d8b23bc527dd73283b477997305afd58/src/execution/execute.ts#L234-L250): ```ts const incrementalPublisher = exeContext.incrementalPublisher; const initialResultRecord = incrementalPublisher.prepareInitialResultRecord(); try { const data = executeOperation(exeContext, initialResultRecord); if (isPromise(data)) { return data.then( (resolved) => incrementalPublisher.buildDataResponse(initialResultRecord, resolved), (error) => incrementalPublisher.buildErrorResponse(initialResultRecord, error), ); } return incrementalPublisher.buildDataResponse(initialResultRecord, data); } catch (error) { return incrementalPublisher.buildErrorResponse(initialResultRecord, error); } ``` Supporting changes are required: 1. some existing public methods no longer are required to be public and so are made private (or removed entirely!), with lint rules forcing the reordering of existing methods 2. to prevent cyclic type dependencies (not strictly necessary, but still!), types are moved from `execute.ts` to `IncrementalPublisher.ts`
sakesun · Sep 1, 2023 · 4973be2 · 4973be2
1 parent 30214e6
commit 4973be2
Show file tree

Hide file tree

Showing 10 changed files with 190 additions and 215 deletions.
diff --git a/src/execution/IncrementalPublisher.ts b/src/execution/IncrementalPublisher.ts
@@ -8,6 +8,63 @@ import type {
   GraphQLFormattedError,
 } from '../error/GraphQLError.js';
 
+/**
+ * The result of GraphQL execution.
+ *
+ *   - `errors` is included when any errors occurred as a non-empty array.
+ *   - `data` is the result of a successful execution of the query.
+ *   - `hasNext` is true if a future payload is expected.
+ *   - `extensions` is reserved for adding non-standard properties.
+ *   - `incremental` is a list of the results from defer/stream directives.
+ */
+export interface ExecutionResult<
+  TData = ObjMap<unknown>,
+  TExtensions = ObjMap<unknown>,
+> {
+  errors?: ReadonlyArray<GraphQLError>;
+  data?: TData | null;
+  extensions?: TExtensions;
+}
+
+export interface FormattedExecutionResult<
+  TData = ObjMap<unknown>,
+  TExtensions = ObjMap<unknown>,
+> {
+  errors?: ReadonlyArray<GraphQLFormattedError>;
+  data?: TData | null;
+  extensions?: TExtensions;
+}
+
+export interface ExperimentalIncrementalExecutionResults<
+  TData = ObjMap<unknown>,
+  TExtensions = ObjMap<unknown>,
+> {
+  initialResult: InitialIncrementalExecutionResult<TData, TExtensions>;
+  subsequentResults: AsyncGenerator<
+    SubsequentIncrementalExecutionResult<TData, TExtensions>,
+    void,
+    void
+  >;
+}
+
+export interface InitialIncrementalExecutionResult<
+  TData = ObjMap<unknown>,
+  TExtensions = ObjMap<unknown>,
+> extends ExecutionResult<TData, TExtensions> {
+  hasNext: boolean;
+  incremental?: ReadonlyArray<IncrementalResult<TData, TExtensions>>;
+  extensions?: TExtensions;
+}
+
+export interface FormattedInitialIncrementalExecutionResult<
+  TData = ObjMap<unknown>,
+  TExtensions = ObjMap<unknown>,
+> extends FormattedExecutionResult<TData, TExtensions> {
+  hasNext: boolean;
+  incremental?: ReadonlyArray<FormattedIncrementalResult<TData, TExtensions>>;
+  extensions?: TExtensions;
+}
+
 export interface SubsequentIncrementalExecutionResult<
   TData = ObjMap<unknown>,
   TExtensions = ObjMap<unknown>,
@@ -113,86 +170,6 @@ export class IncrementalPublisher {
     this._reset();
   }
 
-  hasNext(): boolean {
-    return this._pending.size > 0;
-  }
-
-  subscribe(): AsyncGenerator<
-    SubsequentIncrementalExecutionResult,
-    void,
-    void
-  > {
-    let isDone = false;
-
-    const _next = async (): Promise<
-      IteratorResult<SubsequentIncrementalExecutionResult, void>
-    > => {
-      // eslint-disable-next-line no-constant-condition
-      while (true) {
-        if (isDone) {
-          return { value: undefined, done: true };
-        }
-
-        for (const item of this._released) {
-          this._pending.delete(item);
-        }
-        const released = this._released;
-        this._released = new Set();
-
-        const result = this._getIncrementalResult(released);
-
-        if (!this.hasNext()) {
-          isDone = true;
-        }
-
-        if (result !== undefined) {
-          return { value: result, done: false };
-        }
-
-        // eslint-disable-next-line no-await-in-loop
-        await this._signalled;
-      }
-    };
-
-    const returnStreamIterators = async (): Promise<void> => {
-      const promises: Array<Promise<IteratorResult<unknown>>> = [];
-      this._pending.forEach((incrementalDataRecord) => {
-        if (
-          isStreamItemsRecord(incrementalDataRecord) &&
-          incrementalDataRecord.asyncIterator?.return
-        ) {
-          promises.push(incrementalDataRecord.asyncIterator.return());
-        }
-      });
-      await Promise.all(promises);
-    };
-
-    const _return = async (): Promise<
-      IteratorResult<SubsequentIncrementalExecutionResult, void>
-    > => {
-      isDone = true;
-      await returnStreamIterators();
-      return { value: undefined, done: true };
-    };
-
-    const _throw = async (
-      error?: unknown,
-    ): Promise<IteratorResult<SubsequentIncrementalExecutionResult, void>> => {
-      isDone = true;
-      await returnStreamIterators();
-      return Promise.reject(error);
-    };
-
-    return {
-      [Symbol.asyncIterator]() {
-        return this;
-      },
-      next: _next,
-      return: _return,
-      throw: _throw,
-    };
-  }
-
   prepareInitialResultRecord(): InitialResultRecord {
     return {
       errors: [],
@@ -256,19 +233,38 @@ export class IncrementalPublisher {
     incrementalDataRecord.errors.push(error);
   }
 
-  publishInitial(initialResult: InitialResultRecord) {
-    for (const child of initialResult.children) {
+  buildDataResponse(
+    initialResultRecord: InitialResultRecord,
+    data: ObjMap<unknown> | null,
+  ): ExecutionResult | ExperimentalIncrementalExecutionResults {
+    for (const child of initialResultRecord.children) {
       if (child.filtered) {
         continue;
       }
       this._publish(child);
     }
+
+    const errors = initialResultRecord.errors;
+    const initialResult = errors.length === 0 ? { data } : { errors, data };
+    if (this._pending.size > 0) {
+      return {
+        initialResult: {
+          ...initialResult,
+          hasNext: true,
+        },
+        subsequentResults: this._subscribe(),
+      };
+    }
+    return initialResult;
   }
 
-  getInitialErrors(
-    initialResult: InitialResultRecord,
-  ): ReadonlyArray<GraphQLError> {
-    return initialResult.errors;
+  buildErrorResponse(
+    initialResultRecord: InitialResultRecord,
+    error: GraphQLError,
+  ): ExecutionResult {
+    const errors = initialResultRecord.errors;
+    errors.push(error);
+    return { data: null, errors };
   }
 
   filter(nullPath: Path, erroringIncrementalDataRecord: IncrementalDataRecord) {
@@ -301,6 +297,82 @@ export class IncrementalPublisher {
     });
   }
 
+  private _subscribe(): AsyncGenerator<
+    SubsequentIncrementalExecutionResult,
+    void,
+    void
+  > {
+    let isDone = false;
+
+    const _next = async (): Promise<
+      IteratorResult<SubsequentIncrementalExecutionResult, void>
+    > => {
+      // eslint-disable-next-line no-constant-condition
+      while (true) {
+        if (isDone) {
+          return { value: undefined, done: true };
+        }
+
+        for (const item of this._released) {
+          this._pending.delete(item);
+        }
+        const released = this._released;
+        this._released = new Set();
+
+        const result = this._getIncrementalResult(released);
+
+        if (this._pending.size === 0) {
+          isDone = true;
+        }
+
+        if (result !== undefined) {
+          return { value: result, done: false };
+        }
+
+        // eslint-disable-next-line no-await-in-loop
+        await this._signalled;
+      }
+    };
+
+    const returnStreamIterators = async (): Promise<void> => {
+      const promises: Array<Promise<IteratorResult<unknown>>> = [];
+      this._pending.forEach((incrementalDataRecord) => {
+        if (
+          isStreamItemsRecord(incrementalDataRecord) &&
+          incrementalDataRecord.asyncIterator?.return
+        ) {
+          promises.push(incrementalDataRecord.asyncIterator.return());
+        }
+      });
+      await Promise.all(promises);
+    };
+
+    const _return = async (): Promise<
+      IteratorResult<SubsequentIncrementalExecutionResult, void>
+    > => {
+      isDone = true;
+      await returnStreamIterators();
+      return { value: undefined, done: true };
+    };
+
+    const _throw = async (
+      error?: unknown,
+    ): Promise<IteratorResult<SubsequentIncrementalExecutionResult, void>> => {
+      isDone = true;
+      await returnStreamIterators();
+      return Promise.reject(error);
+    };
+
+    return {
+      [Symbol.asyncIterator]() {
+        return this;
+      },
+      next: _next,
+      return: _return,
+      throw: _throw,
+    };
+  }
+
   private _trigger() {
     this._resolve();
     this._reset();
@@ -368,9 +440,10 @@ export class IncrementalPublisher {
       incrementalResults.push(incrementalResult);
     }
 
+    const hasNext = this._pending.size > 0;
     return incrementalResults.length
-      ? { incremental: incrementalResults, hasNext: this.hasNext() }
-      : encounteredCompletedAsyncIterator && !this.hasNext()
+      ? { incremental: incrementalResults, hasNext }
+      : encounteredCompletedAsyncIterator && !hasNext
       ? { hasNext: false }
       : undefined;
   }

diff --git a/src/execution/__tests__/defer-test.ts b/src/execution/__tests__/defer-test.ts
@@ -16,9 +16,11 @@ import {
 import { GraphQLID, GraphQLString } from '../../type/scalars.js';
 import { GraphQLSchema } from '../../type/schema.js';
 
-import type { InitialIncrementalExecutionResult } from '../execute.js';
 import { execute, experimentalExecuteIncrementally } from '../execute.js';
-import type { SubsequentIncrementalExecutionResult } from '../IncrementalPublisher.js';
+import type {
+  InitialIncrementalExecutionResult,
+  SubsequentIncrementalExecutionResult,
+} from '../IncrementalPublisher.js';
 
 const friendType = new GraphQLObjectType({
   fields: {

diff --git a/src/execution/__tests__/lists-test.ts b/src/execution/__tests__/lists-test.ts
@@ -18,8 +18,8 @@ import { GraphQLSchema } from '../../type/schema.js';
 
 import { buildSchema } from '../../utilities/buildASTSchema.js';
 
-import type { ExecutionResult } from '../execute.js';
 import { execute, executeSync } from '../execute.js';
+import type { ExecutionResult } from '../IncrementalPublisher.js';
 
 describe('Execute: Accepts any iterable as list value', () => {
   function complete(rootValue: unknown) {

diff --git a/src/execution/__tests__/nonnull-test.ts b/src/execution/__tests__/nonnull-test.ts
@@ -13,8 +13,8 @@ import { GraphQLSchema } from '../../type/schema.js';
 
 import { buildSchema } from '../../utilities/buildASTSchema.js';
 
-import type { ExecutionResult } from '../execute.js';
 import { execute, executeSync } from '../execute.js';
+import type { ExecutionResult } from '../IncrementalPublisher.js';
 
 const syncError = new Error('sync');
 const syncNonNullError = new Error('syncNonNull');

diff --git a/src/execution/__tests__/oneof-test.ts b/src/execution/__tests__/oneof-test.ts
@@ -6,8 +6,8 @@ import { parse } from '../../language/parser.js';
 
 import { buildSchema } from '../../utilities/buildASTSchema.js';
 
-import type { ExecutionResult } from '../execute.js';
 import { execute } from '../execute.js';
+import type { ExecutionResult } from '../IncrementalPublisher.js';
 
 const schema = buildSchema(`
   type Query {

diff --git a/src/execution/__tests__/stream-test.ts b/src/execution/__tests__/stream-test.ts
@@ -17,9 +17,11 @@ import {
 import { GraphQLID, GraphQLString } from '../../type/scalars.js';
 import { GraphQLSchema } from '../../type/schema.js';
 
-import type { InitialIncrementalExecutionResult } from '../execute.js';
 import { experimentalExecuteIncrementally } from '../execute.js';
-import type { SubsequentIncrementalExecutionResult } from '../IncrementalPublisher.js';
+import type {
+  InitialIncrementalExecutionResult,
+  SubsequentIncrementalExecutionResult,
+} from '../IncrementalPublisher.js';
 
 const friendType = new GraphQLObjectType({
   fields: {

diff --git a/src/execution/__tests__/subscribe-test.ts b/src/execution/__tests__/subscribe-test.ts
@@ -20,8 +20,9 @@ import {
 } from '../../type/scalars.js';
 import { GraphQLSchema } from '../../type/schema.js';
 
-import type { ExecutionArgs, ExecutionResult } from '../execute.js';
+import type { ExecutionArgs } from '../execute.js';
 import { createSourceEventStream, subscribe } from '../execute.js';
+import type { ExecutionResult } from '../IncrementalPublisher.js';
 
 import { SimplePubSub } from './simplePubSub.js';