Experiment with simplifying new defer/stream execute types

In this experiment, either execute returns a single value (with no
hasNext/incremental), or it returns an async iterable of values where
most of the data (including for the first payload!) is on an
`incremental` field.

This does affect the form of subscriptions in a perhaps surprising way.

I did not make tests pass.

Author: glasser

src/execution/__tests__/flattenAsyncIterable-test.ts

@@ -139,9 +139,7 @@ export interface ExecutionResult<
 > {
   errors?: ReadonlyArray<GraphQLError>;
   data?: TData | null;
-  hasNext?: boolean;
   extensions?: TExtensions;
-  incremental?: ReadonlyArray<IncrementalResult>;
 }
 
 export interface FormattedExecutionResult<
@@ -155,23 +153,17 @@ export interface FormattedExecutionResult<
   incremental?: ReadonlyArray<IncrementalResult>;
 }
 
-export interface SubsequentExecutionResult<TExtensions = ObjMap<unknown>> {
+export interface AsyncExecutionResult {
   hasNext?: boolean;
-  extensions?: TExtensions;
   incremental?: ReadonlyArray<IncrementalResult>;
 }
 
-export type AsyncExecutionResult = ExecutionResult | SubsequentExecutionResult;
-
 export interface IncrementalDeferResult<
   TData = ObjMap<unknown>,
   TExtensions = ObjMap<unknown>,
-> {
-  errors?: ReadonlyArray<GraphQLError>;
-  data?: TData | null;
+> extends ExecutionResult<TData, TExtensions> {
   path?: ReadonlyArray<string | number>;
   label?: string;
-  extensions?: TExtensions;
 }
 
 export interface IncrementalStreamResult<
@@ -1362,7 +1354,8 @@ export const defaultFieldResolver: GraphQLFieldResolver<unknown, unknown> =
 export function subscribe(
   args: ExecutionArgs,
 ): PromiseOrValue<
-  AsyncGenerator<ExecutionResult, void, void> | ExecutionResult
+  | AsyncGenerator<ExecutionResult | AsyncExecutionResult, void, void>
+  | ExecutionResult
 > {
   // If a valid execution context cannot be created due to incorrect arguments,
   // a "Response" with only errors is returned.
@@ -1384,11 +1377,17 @@ export function subscribe(
   return mapSourceToResponse(exeContext, resultOrStream);
 }
 
+// eslint-disable-next-line @typescript-eslint/require-await
+async function* singleElementAsyncIterable<T>(element: T) {
+  yield element;
+}
+
 function mapSourceToResponse(
   exeContext: ExecutionContext,
   resultOrStream: ExecutionResult | AsyncIterable<unknown>,
 ): PromiseOrValue<
-  AsyncGenerator<ExecutionResult, void, void> | ExecutionResult
+  | AsyncGenerator<ExecutionResult | AsyncExecutionResult, void, void>
+  | ExecutionResult
 > {
   if (!isAsyncIterable(resultOrStream)) {
     return resultOrStream;
@@ -1400,10 +1399,16 @@ function mapSourceToResponse(
   // the GraphQL specification. The `execute` function provides the
   // "ExecuteSubscriptionEvent" algorithm, as it is nearly identical to the
   // "ExecuteQuery" algorithm, for which `execute` is also used.
-  return flattenAsyncIterable<ExecutionResult, AsyncExecutionResult>(
-    mapAsyncIterable(resultOrStream, (payload: unknown) =>
-      executeImpl(buildPerEventExecutionContext(exeContext, payload)),
-    ),
+  return flattenAsyncIterable<ExecutionResult | AsyncExecutionResult>(
+    mapAsyncIterable(resultOrStream, async (payload: unknown) => {
+      const result = await executeImpl(
+        buildPerEventExecutionContext(exeContext, payload),
+      );
+      if (isAsyncIterable(result)) {
+        return result;
+      }
+      return singleElementAsyncIterable(result);
+    }),
   );
 }
 
@@ -1874,8 +1879,8 @@ function yieldSubsequentPayloads(
         _hasReturnedInitialResult = true;
         return Promise.resolve({
           value: {
-            ...initialResult,
             hasNext: true,
+            incremental: [initialResult],
           },
           done: false,
         });

src/execution/execute.ts

@@ -1,50 +1,17 @@
-import { isAsyncIterable } from '../jsutils/isAsyncIterable';
-
 type AsyncIterableOrGenerator<T> =
   | AsyncGenerator<T, void, void>
   | AsyncIterable<T>;
 
 /**
- * Given an AsyncIterable that could potentially yield other async iterators,
- * flatten all yielded results into a single AsyncIterable
+ * Given an AsyncIterable of async iterators, flatten all yielded results into a
+ * single AsyncIterable
  */
-export function flattenAsyncIterable<T, AT>(
-  iterable: AsyncIterableOrGenerator<T | AsyncIterableOrGenerator<AT>>,
-): AsyncGenerator<T | AT, void, void> {
-  const iteratorMethod = iterable[Symbol.asyncIterator];
-  const iterator: any = iteratorMethod.call(iterable);
-  let iteratorStack: Array<AsyncIterator<T>> = [iterator];
-
-  async function next(): Promise<IteratorResult<T | AT, void>> {
-    const currentIterator = iteratorStack[0];
-    if (!currentIterator) {
-      return { value: undefined, done: true };
-    }
-    const result = await currentIterator.next();
-    if (result.done) {
-      iteratorStack.shift();
-      return next();
-    } else if (isAsyncIterable(result.value)) {
-      const childIterator = result.value[
-        Symbol.asyncIterator
-      ]() as AsyncIterator<T>;
-      iteratorStack.unshift(childIterator);
-      return next();
+export async function* flattenAsyncIterable<T>(
+  iterable: AsyncIterableOrGenerator<AsyncIterableOrGenerator<T>>,
+): AsyncGenerator<T> {
+  for await (const subIterator of iterable) {
+    for await (const value of subIterator) {
+      yield value;
     }
-    return result;
   }
-  return {
-    next,
-    return() {
-      iteratorStack = [];
-      return iterator.return();
-    },
-    throw(error?: unknown): Promise<IteratorResult<T | AT>> {
-      iteratorStack = [];
-      return iterator.throw(error);
-    },
-    [Symbol.asyncIterator]() {
-      return this;
-    },
-  };
 }

src/execution/flattenAsyncIterable.ts

@@ -13,7 +13,6 @@ export type {
   ExecutionArgs,
   ExecutionResult,
   FormattedExecutionResult,
-  SubsequentExecutionResult,
   IncrementalDeferResult,
   IncrementalStreamResult,
   IncrementalResult,

src/execution/index.ts

@@ -334,7 +334,6 @@ export type {
   ExecutionArgs,
   ExecutionResult,
   FormattedExecutionResult,
-  SubsequentExecutionResult,
   IncrementalDeferResult,
   IncrementalStreamResult,
   IncrementalResult,