feat: execute/subscribe AsyncIterable API

packages/core/test/common.ts

@@ -1,12 +1,13 @@
 import { EventEmitter, on } from 'events';
 import { GraphQLID, GraphQLNonNull, GraphQLObjectType, GraphQLSchema, GraphQLString } from 'graphql';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
 
 const createPubSub = <TTopicPayload extends { [key: string]: unknown }>(emitter: EventEmitter) => {
   return {
     publish: <TTopic extends Extract<keyof TTopicPayload, string>>(topic: TTopic, payload: TTopicPayload[TTopic]) => {
       emitter.emit(topic as string, payload);
     },
-    subscribe: async function*<TTopic extends Extract<keyof TTopicPayload, string>>(
+    subscribe: async function* <TTopic extends Extract<keyof TTopicPayload, string>>(
       topic: TTopic
     ): AsyncIterableIterator<TTopicPayload[TTopic]> {
       const asyncIterator = on(emitter, topic);
@@ -52,7 +53,7 @@ const GraphQLSubscription = new GraphQLObjectType({
   fields: {
     ping: {
       type: GraphQLString,
-      subscribe: async function*() {
+      subscribe: async function* () {
         const stream = pubSub.subscribe('ping');
         return yield* stream;
       },
@@ -79,3 +80,17 @@ export const subscription = /* GraphQL */ `
     ping
   }
 `;
+
+export const collectAsyncIteratorValues = async <TType>(asyncIterable: AsyncIterableIterator<TType>): Promise<Array<TType>> => {
+  const values: Array<TType> = [];
+  for await (const value of asyncIterable) {
+    values.push(value);
+  }
+  return values;
+};
+
+export function assertAsyncIterator(input: unknown): asserts input is AsyncIterableIterator<unknown> {
+  if (!isAsyncIterable(input)) {
+    throw new Error('Expected AsyncIterable iterator.');
+  }
+}

packages/core/test/execute.spec.ts

@@ -1,6 +1,7 @@
 import { createSpiedPlugin, createTestkit } from '@envelop/testing';
-import { execute, GraphQLSchema } from 'graphql';
-import { schema, query } from './common';
+import { execute, ExecutionResult } from 'graphql';
+import { ExecuteFunction } from 'packages/types/src';
+import { schema, query, assertAsyncIterator, collectAsyncIteratorValues } from './common';
 
 describe('execute', () => {
   it('Should wrap and trigger events correctly', async () => {
@@ -133,4 +134,120 @@ describe('execute', () => {
       setResult: expect.any(Function),
     });
   });
+
+  it('Should be able to manipulate streams', async () => {
+    const streamExecuteFn = async function* () {
+      for (const value of ['a', 'b', 'c', 'd']) {
+        yield { data: { alphabet: value } };
+      }
+    };
+
+    const teskit = createTestkit(
+      [
+        {
+          onExecute({ setExecuteFn }) {
+            setExecuteFn(streamExecuteFn);
+
+            return {
+              onExecuteDone: () => {
+                return {
+                  onNext: ({ setResult }) => {
+                    setResult({ data: { alphabet: 'x' } });
+                  },
+                };
+              },
+            };
+          },
+        },
+      ],
+      schema
+    );
+
+    const result: ReturnType<ExecuteFunction> = await teskit.executeRaw({} as any);
+    assertAsyncIterator(result);
+    const values = await collectAsyncIteratorValues(result);
+    expect(values).toEqual([
+      { data: { alphabet: 'x' } },
+      { data: { alphabet: 'x' } },
+      { data: { alphabet: 'x' } },
+      { data: { alphabet: 'x' } },
+    ]);
+  });
+
+  it('Should be able to invoke something after the stream has ended.', async () => {
+    expect.assertions(1);
+    const streamExecuteFn = async function* () {
+      for (const value of ['a', 'b', 'c', 'd']) {
+        yield { data: { alphabet: value } };
+      }
+    };
+
+    const teskit = createTestkit(
+      [
+        {
+          onExecute({ setExecuteFn }) {
+            setExecuteFn(streamExecuteFn);
+
+            return {
+              onExecuteDone: () => {
+                let latestResult: ExecutionResult;
+                return {
+                  onNext: ({ result }) => {
+                    latestResult = result;
+                  },
+                  onEnd: () => {
+                    expect(latestResult).toEqual({ data: { alphabet: 'd' } });
+                  },
+                };
+              },
+            };
+          },
+        },
+      ],
+      schema
+    );
+
+    const result: ReturnType<ExecuteFunction> = await teskit.executeRaw({} as any);
+    assertAsyncIterator(result);
+    // run AsyncGenerator
+    await collectAsyncIteratorValues(result);
+  });
+
+  it.skip('Should be able to invoke something after the stream has ended (manual return).', async () => {
+    expect.assertions(1);
+    const streamExecuteFn = async function* () {
+      for (const value of ['a', 'b', 'c', 'd']) {
+        yield { data: { alphabet: value } };
+      }
+    };
+
+    const teskit = createTestkit(
+      [
+        {
+          onExecute({ setExecuteFn }) {
+            setExecuteFn(streamExecuteFn);
+
+            return {
+              onExecuteDone: () => {
+                let latestResult: ExecutionResult;
+                return {
+                  onNext: ({ result }) => {
+                    latestResult = result;
+                  },
+                  onEnd: () => {
+                    expect(latestResult).toEqual(undefined);
+                  },
+                };
+              },
+            };
+          },
+        },
+      ],
+      schema
+    );
+
+    const result: ReturnType<ExecuteFunction> = await teskit.executeRaw({} as any);
+    assertAsyncIterator(result);
+    result[Symbol.asyncIterator]().return!();
+  });
 });

packages/testing/src/index.ts

@@ -1,7 +1,7 @@
-import { DocumentNode, ExecutionResult, GraphQLSchema, print } from 'graphql';
+import { DocumentNode, ExecutionArgs, ExecutionResult, GraphQLSchema, print } from 'graphql';
 import { getGraphQLParameters, processRequest, Push } from 'graphql-helix';
-import { envelop, useSchema } from '@envelop/core';
-import { Envelop, Plugin } from '@envelop/types';
+import { envelop, makeExecute, useSchema } from '@envelop/core';
+import { Envelop, ExecuteFunction, Plugin } from '@envelop/types';
 
 // eslint-disable-next-line @typescript-eslint/explicit-module-boundary-types
 export function createSpiedPlugin() {
@@ -57,6 +57,7 @@ export function createTestkit(
   subscribe: (operation: DocumentNode | string, variables?: Record<string, any>, initialContext?: any) => Promise<Push<any, any>>;
   replaceSchema: (schema: GraphQLSchema) => void;
   wait: (ms: number) => Promise<void>;
+  executeRaw: ExecuteFunction;
 } {
   let replaceSchema: (s: GraphQLSchema) => void = () => {};
 
@@ -99,7 +100,6 @@ export function createTestkit(
         contextFactory: initialContext ? () => proxy.contextFactory(initialContext) : proxy.contextFactory,
         schema: proxy.schema,
       });
-
       return (r as any).payload as ExecutionResult;
     },
     subscribe: async (operation, rawVariables = {}, initialContext = null) => {
@@ -134,6 +134,13 @@ export function createTestkit(
 
       return r;
     },
+    executeRaw: makeExecute(async (args: ExecutionArgs) => {
+      const proxy = initRequest();
+      return await proxy.execute({
+        ...args,
+        contextValue: await proxy.contextFactory(args.contextValue),
+      });
+    }),
   };
 }