diff --git a/.changeset/fifty-kids-boil.md b/.changeset/fifty-kids-boil.md
new file mode 100644
index 0000000000..00392ced0a
--- /dev/null
+++ b/.changeset/fifty-kids-boil.md
@@ -0,0 +1,6 @@
+---
+'@envelop/core': minor
+'@envelop/types': minor
+---
+
+allow hooking into published subscribe values
diff --git a/.changeset/six-crabs-flash.md b/.changeset/six-crabs-flash.md
new file mode 100644
index 0000000000..c8c9f91be0
--- /dev/null
+++ b/.changeset/six-crabs-flash.md
@@ -0,0 +1,5 @@
+---
+'@envelop/execute-subscription-event': patch
+---
+
+initial release
diff --git a/benchmark/k6.js b/benchmark/k6.js
index be6922fdba..7a4d26c191 100644
--- a/benchmark/k6.js
+++ b/benchmark/k6.js
@@ -50,7 +50,7 @@ export const options = buildOptions({
   'graphql-js': {
     no_errors: ['rate=1.0'],
     expected_result: ['rate=1.0'],
-    http_req_duration: ['p(95)<=28'],
+    http_req_duration: ['p(95)<=35'],
     graphql_execute: ['p(95)<=2'],
     graphql_context: ['p(95)<=1'],
     graphql_validate: ['p(95)<=1'],
@@ -61,7 +61,7 @@ export const options = buildOptions({
   'envelop-just-cache': {
     no_errors: ['rate=1.0'],
     expected_result: ['rate=1.0'],
-    http_req_duration: ['p(95)<=20'],
+    http_req_duration: ['p(95)<=23'],
     graphql_execute: ['p(95)<=1'],
     graphql_context: ['p(95)<=1'],
     graphql_validate: ['p(95)<=1'],
@@ -72,23 +72,23 @@ export const options = buildOptions({
   'prom-tracing': {
     no_errors: ['rate=1.0'],
     expected_result: ['rate=1.0'],
-    http_req_duration: ['p(95)<=45'],
-    graphql_execute: ['p(95)<=4'],
+    http_req_duration: ['p(95)<=52'],
+    graphql_execute: ['p(95)<=6'],
     graphql_context: ['p(95)<=1'],
     graphql_validate: ['p(95)<=1'],
     graphql_parse: ['p(95)<=1'],
     envelop_init: ['p(95)<=1'],
-    envelop_total: ['p(95)<=4'],
+    envelop_total: ['p(95)<=6'],
   },
   'envelop-cache-and-no-internal-tracing': {
     no_errors: ['rate=1.0'],
     expected_result: ['rate=1.0'],
-    http_req_duration: ['p(95)<=15'],
+    http_req_duration: ['p(95)<=18'],
   },
   'envelop-cache-jit': {
     no_errors: ['rate=1.0'],
     expected_result: ['rate=1.0'],
-    http_req_duration: ['p(95)<=14'],
+    http_req_duration: ['p(95)<=16'],
     graphql_execute: ['p(95)<=1'],
     graphql_context: ['p(95)<=1'],
     graphql_validate: ['p(95)<=1'],
diff --git a/packages/core/src/graphql-typings.d.ts b/packages/core/src/graphql-typings.d.ts
index 474528ad13..2dafe88d31 100644
--- a/packages/core/src/graphql-typings.d.ts
+++ b/packages/core/src/graphql-typings.d.ts
@@ -1,4 +1,4 @@
 declare module 'graphql/jsutils/isAsyncIterable' {
-  function isAsyncIterable(input: unknown): input is AsyncIterable<any>;
+  function isAsyncIterable(input: unknown): input is AsyncIterableIterator<any>;
   export default isAsyncIterable;
 }
diff --git a/packages/core/src/orchestrator.ts b/packages/core/src/orchestrator.ts
index e02b75453b..2c44cccd30 100644
--- a/packages/core/src/orchestrator.ts
+++ b/packages/core/src/orchestrator.ts
@@ -16,25 +16,30 @@ import {
   SubscribeResultHook,
   TypedSubscriptionArgs,
   TypedExecutionArgs,
+  SubscribeFunction,
+  OnSubscribeResultResultOnNextHook,
+  OnSubscribeResultResultOnEndHook,
+  OnExecuteDoneHookResultOnNextHook,
+  OnExecuteDoneHookResultOnEndHook,
+  ExecuteFunction,
+  AsyncIterableIteratorOrValue,
 } from '@envelop/types';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
 import {
   DocumentNode,
   execute,
-  ExecutionArgs,
   ExecutionResult,
   GraphQLError,
-  GraphQLFieldResolver,
   GraphQLSchema,
-  GraphQLTypeResolver,
   parse,
   specifiedRules,
   subscribe,
-  SubscriptionArgs,
   validate,
   ValidationRule,
 } from 'graphql';
 import { Maybe } from 'graphql/jsutils/Maybe';
 import { prepareTracedSchema, resolversHooksSymbol } from './traced-schema';
+import { finalAsyncIterator, makeExecute, makeSubscribe, mapAsyncIterator } from './utils';
 
 export type EnvelopOrchestrator<
   InitialContext extends ArbitraryObject = ArbitraryObject,
@@ -272,32 +277,9 @@ export function createEnvelopOrchestrator<PluginsContext = any>(plugins: Plugin[
       }
     : initialContext => orchestratorCtx => orchestratorCtx ? { ...initialContext, ...orchestratorCtx } : initialContext;
 
-  const customSubscribe = async (
-    argsOrSchema: SubscriptionArgs | GraphQLSchema,
-    document?: DocumentNode,
-    rootValue?: any,
-    contextValue?: any,
-    variableValues?: Maybe<{ [key: string]: any }>,
-    operationName?: Maybe<string>,
-    fieldResolver?: Maybe<GraphQLFieldResolver<any, any>>,
-    subscribeFieldResolver?: Maybe<GraphQLFieldResolver<any, any>>
-  ) => {
-    const args: SubscriptionArgs =
-      argsOrSchema instanceof GraphQLSchema
-        ? {
-            schema: argsOrSchema,
-            document: document!,
-            rootValue,
-            contextValue,
-            variableValues,
-            operationName,
-            fieldResolver,
-            subscribeFieldResolver,
-          }
-        : argsOrSchema;
-
+  const customSubscribe = makeSubscribe(async args => {
     const onResolversHandlers: OnResolverCalledHook[] = [];
-    let subscribeFn: typeof subscribe = subscribe;
+    let subscribeFn = subscribe as SubscribeFunction;
 
     const afterCalls: SubscribeResultHook[] = [];
     let context = args.contextValue || {};
@@ -333,46 +315,49 @@ export function createEnvelopOrchestrator<PluginsContext = any>(plugins: Plugin[
       contextValue: context,
     });
 
+    const onNextHandler: OnSubscribeResultResultOnNextHook[] = [];
+    const onEndHandler: OnSubscribeResultResultOnEndHook[] = [];
+
     for (const afterCb of afterCalls) {
-      afterCb({
+      const hookResult = afterCb({
         result,
         setResult: newResult => {
           result = newResult;
         },
       });
+      if (hookResult) {
+        if (hookResult.onNext) {
+          onNextHandler.push(hookResult.onNext);
+        }
+        if (hookResult.onEnd) {
+          onEndHandler.push(hookResult.onEnd);
+        }
+      }
     }
 
+    if (onNextHandler.length && isAsyncIterable(result)) {
+      result = mapAsyncIterator(result, async result => {
+        for (const onNext of onNextHandler) {
+          await onNext({ result, setResult: newResult => (result = newResult) });
+        }
+        return result;
+      });
+    }
+    if (onEndHandler.length && isAsyncIterable(result)) {
+      result = finalAsyncIterator(result, () => {
+        for (const onEnd of onEndHandler) {
+          onEnd();
+        }
+      });
+    }
     return result;
-  };
+  });
 
   const customExecute = beforeCallbacks.execute.length
-    ? async (
-        argsOrSchema: ExecutionArgs | GraphQLSchema,
-        document?: DocumentNode,
-        rootValue?: any,
-        contextValue?: any,
-        variableValues?: Maybe<{ [key: string]: any }>,
-        operationName?: Maybe<string>,
-        fieldResolver?: Maybe<GraphQLFieldResolver<any, any>>,
-        typeResolver?: Maybe<GraphQLTypeResolver<any, any>>
-      ) => {
-        const args: ExecutionArgs =
-          argsOrSchema instanceof GraphQLSchema
-            ? {
-                schema: argsOrSchema,
-                document: document!,
-                rootValue,
-                contextValue,
-                variableValues,
-                operationName,
-                fieldResolver,
-                typeResolver,
-              }
-            : argsOrSchema;
-
+    ? makeExecute(async args => {
         const onResolversHandlers: OnResolverCalledHook[] = [];
-        let executeFn: typeof execute = execute;
-        let result: ExecutionResult;
+        let executeFn = execute as ExecuteFunction;
+        let result: AsyncIterableIteratorOrValue<ExecutionResult>;
 
         const afterCalls: OnExecuteDoneHook[] = [];
         let context = args.contextValue || {};
@@ -429,17 +414,44 @@ export function createEnvelopOrchestrator<PluginsContext = any>(plugins: Plugin[
           contextValue: context,
         });
 
+        const onNextHandler: OnExecuteDoneHookResultOnNextHook[] = [];
+        const onEndHandler: OnExecuteDoneHookResultOnEndHook[] = [];
+
         for (const afterCb of afterCalls) {
-          afterCb({
+          const hookResult = afterCb({
             result,
             setResult: newResult => {
               result = newResult;
             },
           });
+          if (hookResult) {
+            if (hookResult.onNext) {
+              onNextHandler.push(hookResult.onNext);
+            }
+            if (hookResult.onEnd) {
+              onEndHandler.push(hookResult.onEnd);
+            }
+          }
+        }
+
+        if (onNextHandler.length && isAsyncIterable(result)) {
+          result = mapAsyncIterator(result, async result => {
+            for (const onNext of onNextHandler) {
+              await onNext({ result, setResult: newResult => (result = newResult) });
+            }
+            return result;
+          });
+        }
+        if (onEndHandler.length && isAsyncIterable(result)) {
+          result = finalAsyncIterator(result, () => {
+            for (const onEnd of onEndHandler) {
+              onEnd();
+            }
+          });
         }
 
         return result;
-      }
+      })
     : execute;
 
   initDone = true;
@@ -463,7 +475,7 @@ export function createEnvelopOrchestrator<PluginsContext = any>(plugins: Plugin[
     init,
     parse: customParse,
     validate: customValidate,
-    execute: customExecute,
+    execute: customExecute as ExecuteFunction,
     subscribe: customSubscribe,
     contextFactory: customContextFactory,
   };
diff --git a/packages/core/src/plugins/use-error-handler.ts b/packages/core/src/plugins/use-error-handler.ts
index ca0f200daf..827b88912b 100644
--- a/packages/core/src/plugins/use-error-handler.ts
+++ b/packages/core/src/plugins/use-error-handler.ts
@@ -1,13 +1,31 @@
 import { Plugin } from '@envelop/types';
-import { GraphQLError } from 'graphql';
+import { ExecutionResult, GraphQLError } from 'graphql';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
 
-export const useErrorHandler = (errorHandler: (errors: readonly GraphQLError[]) => void): Plugin => ({
+export type ErrorHandler = (errors: readonly GraphQLError[]) => void;
+
+const makeHandleResult =
+  (errorHandler: ErrorHandler) =>
+  ({ result }: { result: ExecutionResult }) => {
+    if (result.errors?.length) {
+      errorHandler(result.errors);
+    }
+  };
+
+export const useErrorHandler = (errorHandler: ErrorHandler): Plugin => ({
   onExecute() {
+    const handleResult = makeHandleResult(errorHandler);
     return {
-      onExecuteDone: ({ result }) => {
-        if (result.errors?.length) {
-          errorHandler(result.errors);
+      onExecuteDone({ result }) {
+        if (isAsyncIterable(result)) {
+          return {
+            onNext({ result }) {
+              handleResult({ result });
+            },
+          };
         }
+        handleResult({ result });
+        return undefined;
       },
     };
   },
diff --git a/packages/core/src/plugins/use-masked-errors.ts b/packages/core/src/plugins/use-masked-errors.ts
index 22903396e6..233fc77d0b 100644
--- a/packages/core/src/plugins/use-masked-errors.ts
+++ b/packages/core/src/plugins/use-masked-errors.ts
@@ -1,5 +1,6 @@
 import { Plugin } from '@envelop/types';
-import { GraphQLError } from 'graphql';
+import { ExecutionResult, GraphQLError } from 'graphql';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
 
 export class EnvelopError extends GraphQLError {
   constructor(message: string, extensions?: Record<string, any>) {
@@ -21,16 +22,31 @@ export type UseMaskedErrorsOpts = {
   formatError?: FormatErrorHandler;
 };
 
+const makeHandleResult =
+  (format: FormatErrorHandler) =>
+  ({ result, setResult }: { result: ExecutionResult; setResult: (result: ExecutionResult) => void }) => {
+    if (result.errors != null) {
+      setResult({ ...result, errors: result.errors.map(error => format(error)) });
+    }
+  };
+
 export const useMaskedErrors = (opts?: UseMaskedErrorsOpts): Plugin => {
   const format = opts?.formatError ?? formatError;
+  const handleResult = makeHandleResult(format);
 
   return {
-    onExecute: () => {
+    onExecute() {
       return {
-        onExecuteDone: ({ result, setResult }) => {
-          if (result.errors != null) {
-            setResult({ ...result, errors: result.errors.map(error => format(error)) });
+        onExecuteDone({ result, setResult }) {
+          if (isAsyncIterable(result)) {
+            return {
+              onNext: ({ result, setResult }) => {
+                handleResult({ result, setResult });
+              },
+            };
           }
+          handleResult({ result, setResult });
+          return undefined;
         },
       };
     },
diff --git a/packages/core/src/plugins/use-payload-formatter.ts b/packages/core/src/plugins/use-payload-formatter.ts
index e15b5cc64d..7dd06da381 100644
--- a/packages/core/src/plugins/use-payload-formatter.ts
+++ b/packages/core/src/plugins/use-payload-formatter.ts
@@ -1,15 +1,33 @@
 import { Plugin } from '@envelop/types';
 import { ExecutionResult } from 'graphql';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
 
-export const usePayloadFormatter = (formatter: (result: ExecutionResult<any, any>) => false | ExecutionResult<any, any>): Plugin => ({
+export type FormatterFunction = (result: ExecutionResult<any, any>) => false | ExecutionResult<any, any>;
+
+const makeHandleResult =
+  (formatter: FormatterFunction) =>
+  ({ result, setResult }: { result: ExecutionResult; setResult: (result: ExecutionResult) => void }) => {
+    const modified = formatter(result);
+    if (modified !== false) {
+      setResult(modified);
+    }
+  };
+
+export const usePayloadFormatter = (formatter: FormatterFunction): Plugin => ({
   onExecute() {
+    const handleResult = makeHandleResult(formatter);
     return {
-      onExecuteDone: ({ result, setResult }) => {
-        const modified = formatter(result);
-
-        if (modified !== false) {
-          setResult(modified);
+      onExecuteDone({ result, setResult }) {
+        if (isAsyncIterable(result)) {
+          return {
+            onNext({ result, setResult }) {
+              handleResult({ result, setResult });
+            },
+          };
         }
+
+        handleResult({ result, setResult });
+        return undefined;
       },
     };
   },
diff --git a/packages/core/src/traced-orchestrator.ts b/packages/core/src/traced-orchestrator.ts
index 93ab5579c5..2ff92d7361 100644
--- a/packages/core/src/traced-orchestrator.ts
+++ b/packages/core/src/traced-orchestrator.ts
@@ -2,6 +2,7 @@ import { DocumentNode, ExecutionArgs, GraphQLFieldResolver, GraphQLSchema, Graph
 import { Maybe } from 'graphql/jsutils/Maybe';
 import { ArbitraryObject } from '@envelop/types';
 import { EnvelopOrchestrator } from './orchestrator';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
 
 const HR_TO_NS = 1e9;
 const NS_TO_MS = 1e6;
@@ -94,8 +95,16 @@ export function traceOrchestrator<TInitialContext extends ArbitraryObject, TPlug
       try {
         const result = await orchestrator.execute(args);
         done();
-        result.extensions = result.extensions || {};
-        result.extensions.envelopTracing = args.contextValue._envelopTracing;
+
+        if (!isAsyncIterable(result)) {
+          result.extensions = result.extensions || {};
+          result.extensions.envelopTracing = args.contextValue._envelopTracing;
+        } else {
+          // eslint-disable-next-line no-console
+          console.warn(
+            `"traceOrchestrator" encountered a AsyncIterator which is not supported yet, so tracing data is not available for the operation.`
+          );
+        }
 
         return result;
       } catch (e) {
diff --git a/packages/core/src/utils.ts b/packages/core/src/utils.ts
index f4f32be7a1..9f44e492dc 100644
--- a/packages/core/src/utils.ts
+++ b/packages/core/src/utils.ts
@@ -1,4 +1,17 @@
-import { ASTNode, DocumentNode, Kind, OperationDefinitionNode, visit, BREAK, Source } from 'graphql';
+import {
+  ASTNode,
+  DocumentNode,
+  Kind,
+  OperationDefinitionNode,
+  visit,
+  BREAK,
+  Source,
+  ExecutionResult,
+  SubscriptionArgs,
+  ExecutionArgs,
+} from 'graphql';
+import { AsyncIterableIteratorOrValue, PolymorphicExecuteArguments, PolymorphicSubscribeArguments } from '@envelop/types';
+import { PromiseOrValue } from 'graphql/jsutils/PromiseOrValue';
 
 export const envelopIsIntrospectionSymbol = Symbol('ENVELOP_IS_INTROSPECTION');
 
@@ -34,3 +47,69 @@ export function isIntrospectionDocument(document: DocumentNode): boolean {
 export function isIntrospectionOperationString(operation: string | Source): boolean {
   return (typeof operation === 'string' ? operation : operation.body).indexOf('__schema') !== -1;
 }
+
+export function getSubscribeArgs(args: PolymorphicSubscribeArguments): SubscriptionArgs {
+  return args.length === 1
+    ? args[0]
+    : {
+        schema: args[0],
+        document: args[1],
+        rootValue: args[2],
+        contextValue: args[3],
+        variableValues: args[4],
+        operationName: args[5],
+        fieldResolver: args[6],
+        subscribeFieldResolver: args[7],
+      };
+}
+
+/**
+ * Utility function for making a subscribe function that handles polymorphic arguments.
+ */
+export const makeSubscribe =
+  (subscribeFn: (args: SubscriptionArgs) => PromiseOrValue<AsyncIterableIterator<ExecutionResult> | ExecutionResult>) =>
+  (...polyArgs: PolymorphicSubscribeArguments): PromiseOrValue<AsyncIterableIterator<ExecutionResult> | ExecutionResult> =>
+    subscribeFn(getSubscribeArgs(polyArgs));
+
+export async function* mapAsyncIterator<TInput, TOutput = TInput>(
+  asyncIterable: AsyncIterableIterator<TInput>,
+  map: (input: TInput) => Promise<TOutput> | TOutput
+): AsyncIterableIterator<TOutput> {
+  for await (const value of asyncIterable) {
+    yield map(value);
+  }
+}
+
+export function getExecuteArgs(args: PolymorphicExecuteArguments): ExecutionArgs {
+  return args.length === 1
+    ? args[0]
+    : {
+        schema: args[0],
+        document: args[1],
+        rootValue: args[2],
+        contextValue: args[3],
+        variableValues: args[4],
+        operationName: args[5],
+        fieldResolver: args[6],
+        typeResolver: args[7],
+      };
+}
+
+/**
+ * Utility function for making a execute function that handles polymorphic arguments.
+ */
+export const makeExecute =
+  (executeFn: (args: ExecutionArgs) => PromiseOrValue<AsyncIterableIteratorOrValue<ExecutionResult>>) =>
+  (...polyArgs: PolymorphicExecuteArguments): PromiseOrValue<AsyncIterableIteratorOrValue<ExecutionResult>> =>
+    executeFn(getExecuteArgs(polyArgs));
+
+export async function* finalAsyncIterator<TInput>(
+  asyncIterable: AsyncIterableIterator<TInput>,
+  onFinal: () => void
+): AsyncIterableIterator<TInput> {
+  try {
+    yield* asyncIterable;
+  } finally {
+    onFinal();
+  }
+}
diff --git a/packages/core/test/common.ts b/packages/core/test/common.ts
index 261a7cc736..94e0555b25 100644
--- a/packages/core/test/common.ts
+++ b/packages/core/test/common.ts
@@ -4,11 +4,17 @@ export const schema = makeExecutableSchema({
   typeDefs: /* GraphQL */ `
     type Query {
       me: User!
+      alphabet: [String]!
     }
     type User {
       id: ID!
       name: String!
     }
+
+    type Subscription {
+      alphabet: String!
+      message: String!
+    }
   `,
   resolvers: {
     Query: {
@@ -16,6 +22,17 @@ export const schema = makeExecutableSchema({
         return { _id: 1, firstName: 'Dotan', lastName: 'Simha' };
       },
     },
+    Subscription: {
+      message: {
+        subscribe: (_, __, context) => {
+          if (!context || 'subscribeSource' in context === false) {
+            throw new Error('No subscribeSource provided for context :(');
+          }
+          return context.subscribeSource;
+        },
+        resolve: (_, __, context) => context.message,
+      },
+    },
     User: {
       id: u => u._id,
       name: u => `${u.firstName} ${u.lastName}`,
@@ -31,3 +48,9 @@ export const query = /* GraphQL */ `
     }
   }
 `;
+
+export const subscriptionOperationString = /* GraphQL */ `
+  subscription {
+    message
+  }
+`;
diff --git a/packages/core/test/execute.spec.ts b/packages/core/test/execute.spec.ts
index 60a1d4443e..ecc9ba36ff 100644
--- a/packages/core/test/execute.spec.ts
+++ b/packages/core/test/execute.spec.ts
@@ -1,5 +1,5 @@
-import { createSpiedPlugin, createTestkit } from '@envelop/testing';
-import { execute, GraphQLSchema } from 'graphql';
+import { assertStreamExecutionValue, collectAsyncIteratorValues, createSpiedPlugin, createTestkit } from '@envelop/testing';
+import { execute, ExecutionResult, GraphQLSchema } from 'graphql';
 import { schema, query } from './common';
 
 describe('execute', () => {
@@ -138,4 +138,134 @@ 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 = await teskit.execute(/* GraphQL */ `
+      query {
+        alphabet
+      }
+    `);
+    assertStreamExecutionValue(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 = await teskit.execute(/* GraphQL */ `
+      query {
+        alphabet
+      }
+    `);
+    assertStreamExecutionValue(result);
+    // run AsyncGenerator
+    await collectAsyncIteratorValues(result);
+  });
+
+  it('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({ data: { alphabet: 'a' } });
+                  },
+                };
+              },
+            };
+          },
+        },
+      ],
+      schema
+    );
+
+    const result = await teskit.execute(/* GraphQL */ `
+      query {
+        alphabet
+      }
+    `);
+    assertStreamExecutionValue(result);
+    const instance = result[Symbol.asyncIterator]();
+    await instance.next();
+    await instance.return!();
+  });
 });
diff --git a/packages/core/test/subscribe.spec.ts b/packages/core/test/subscribe.spec.ts
new file mode 100644
index 0000000000..53749c187a
--- /dev/null
+++ b/packages/core/test/subscribe.spec.ts
@@ -0,0 +1,90 @@
+import { assertStreamExecutionValue, collectAsyncIteratorValues, createTestkit } from '@envelop/testing';
+import { ExecutionResult } from 'graphql';
+import { schema } from './common';
+
+describe('subscribe', () => {
+  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(
+      [
+        {
+          onSubscribe({ setSubscribeFn }) {
+            setSubscribeFn(streamExecuteFn);
+
+            return {
+              onSubscribeResult: () => {
+                return {
+                  onNext: ({ setResult }) => {
+                    setResult({ data: { alphabet: 'x' } });
+                  },
+                };
+              },
+            };
+          },
+        },
+      ],
+      schema
+    );
+
+    const result = await teskit.execute(/* GraphQL */ `
+      subscription {
+        alphabet
+      }
+    `);
+    assertStreamExecutionValue(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(
+      [
+        {
+          onSubscribe({ setSubscribeFn }) {
+            setSubscribeFn(streamExecuteFn);
+
+            return {
+              onSubscribeResult: () => {
+                let latestResult: ExecutionResult;
+                return {
+                  onNext: ({ result }) => {
+                    latestResult = result;
+                  },
+                  onEnd: () => {
+                    expect(latestResult).toEqual({ data: { alphabet: 'd' } });
+                  },
+                };
+              },
+            };
+          },
+        },
+      ],
+      schema
+    );
+
+    const result = await teskit.execute(/* GraphQL */ `
+      subscription {
+        alphabet
+      }
+    `);
+    assertStreamExecutionValue(result);
+    await collectAsyncIteratorValues(result);
+  });
+});
diff --git a/packages/plugins/apollo-server-errors/src/index.ts b/packages/plugins/apollo-server-errors/src/index.ts
index fd5ceea0f7..3411b01371 100644
--- a/packages/plugins/apollo-server-errors/src/index.ts
+++ b/packages/plugins/apollo-server-errors/src/index.ts
@@ -1,20 +1,38 @@
 import { Plugin } from '@envelop/types';
 import { formatApolloErrors } from 'apollo-server-errors';
+import type { ExecutionResult } from 'graphql';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
+
+const makeHandleResult =
+  (options: Parameters<typeof formatApolloErrors>[1] = {}) =>
+  ({ result, setResult }: { result: ExecutionResult; setResult: (result: ExecutionResult) => void }) => {
+    if (result.errors && result.errors.length > 0) {
+      setResult({
+        ...result,
+        errors: formatApolloErrors(result.errors, {
+          debug: options.debug,
+          formatter: options.formatter,
+        }),
+      });
+    }
+  };
 
 export const useApolloServerErrors = (options: Parameters<typeof formatApolloErrors>[1] = {}): Plugin => {
   return {
     onExecute() {
+      const handleResult = makeHandleResult(options);
+
       return {
         onExecuteDone({ result, setResult }) {
-          if (result.errors && result.errors.length > 0) {
-            setResult({
-              ...result,
-              errors: formatApolloErrors(result.errors, {
-                debug: options.debug,
-                formatter: options.formatter,
-              }),
-            });
+          if (isAsyncIterable(result)) {
+            return {
+              onNext: ({ result, setResult }) => {
+                handleResult({ result, setResult });
+              },
+            };
           }
+          handleResult({ result, setResult });
+          return undefined;
         },
       };
     },
diff --git a/packages/plugins/apollo-server-errors/test/apollo-server-errors.spec.ts b/packages/plugins/apollo-server-errors/test/apollo-server-errors.spec.ts
index 74e6ad8591..107726c9c7 100644
--- a/packages/plugins/apollo-server-errors/test/apollo-server-errors.spec.ts
+++ b/packages/plugins/apollo-server-errors/test/apollo-server-errors.spec.ts
@@ -1,8 +1,9 @@
 import { makeExecutableSchema } from '@graphql-tools/schema';
 import { ApolloServerBase } from 'apollo-server-core';
-import { GraphQLError, GraphQLSchema } from 'graphql';
+import { GraphQLSchema } from 'graphql';
 import { envelop, useSchema } from '@envelop/core';
 import { useApolloServerErrors } from '../src';
+import { assertSingleExecutionValue } from '@envelop/testing';
 
 describe('useApolloServerErrors', () => {
   const executeBoth = async (schema: GraphQLSchema, query: string, debug: boolean) => {
@@ -32,6 +33,7 @@ describe('useApolloServerErrors', () => {
 
     const query = `query test { test }`;
     const results = await executeBoth(schema, query, false);
+    assertSingleExecutionValue(results.envelop);
     expect(results.apollo.data!.test).toBeNull();
     expect(results.envelop.data!.test).toBeNull();
     expect(results.envelop.errors![0].locations).toEqual(results.apollo.errors![0].locations);
@@ -55,6 +57,7 @@ describe('useApolloServerErrors', () => {
 
     const query = `query test { test }`;
     const results = await executeBoth(schema, query, true);
+    assertSingleExecutionValue(results.envelop);
     expect(results.apollo.data!.test).toBeNull();
     expect(results.envelop.data!.test).toBeNull();
     expect(results.envelop.errors![0].locations).toEqual(results.apollo.errors![0].locations);
diff --git a/packages/plugins/apollo-tracing/src/index.ts b/packages/plugins/apollo-tracing/src/index.ts
index 2f3d6c8145..1495cb57a6 100644
--- a/packages/plugins/apollo-tracing/src/index.ts
+++ b/packages/plugins/apollo-tracing/src/index.ts
@@ -1,6 +1,7 @@
 import { Plugin } from '@envelop/types';
 import { TracingFormat } from 'apollo-tracing';
 import { GraphQLType, ResponsePath, responsePathAsArray } from 'graphql';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
 
 const HR_TO_NS = 1e9;
 const NS_TO_MS = 1e6;
@@ -56,7 +57,6 @@ export const useApolloTracing = (): Plugin => {
         },
         onExecuteDone({ result }) {
           const endTime = new Date();
-          result.extensions = result.extensions || {};
 
           const tracing: TracingFormat = {
             version: 1,
@@ -80,7 +80,15 @@ export const useApolloTracing = (): Plugin => {
             },
           };
 
-          result.extensions.tracing = tracing;
+          if (!isAsyncIterable(result)) {
+            result.extensions = result.extensions || {};
+            result.extensions.tracing = tracing;
+          } else {
+            // eslint-disable-next-line no-console
+            console.warn(
+              `Plugin "apollo-tracing" encountered a AsyncIterator which is not supported yet, so tracing data is not available for the operation.`
+            );
+          }
         },
       };
     },
diff --git a/packages/plugins/execute-subscription-event/README.md b/packages/plugins/execute-subscription-event/README.md
new file mode 100644
index 0000000000..a1c15ea68f
--- /dev/null
+++ b/packages/plugins/execute-subscription-event/README.md
@@ -0,0 +1,51 @@
+## `@envelop/execute-subscription-event`
+
+Utilities for hooking into the [ExecuteSubscriptionEvent](<https://spec.graphql.org/draft/#ExecuteSubscriptionEvent()>) phase.
+
+### `useContextValuePerExecuteSubscriptionEvent`
+
+Create a new context object per `ExecuteSubscriptionEvent` phase, allowing to bypass common issues with context objects such as [`DataLoader`](https://github.com/dotansimha/envelop/issues/80) [caching](https://github.com/graphql/graphql-js/issues/894) [issues](https://github.com/apollographql/subscriptions-transport-ws/issues/330).
+
+```ts
+import { envelop } from '@envelop/core';
+import { useContextValuePerExecuteSubscriptionEvent } from '@envelop/execute-subscription-event';
+import { createContext, createDataLoaders } from './context';
+
+const getEnveloped = envelop({
+  plugins: [
+    useContext(() => createContext())
+    useContextValuePerExecuteSubscriptionEvent(() => ({
+      // Existing context is merged with this context partial
+      // By recreating the DataLoader we ensure no DataLoader caches from the previous event/initial field subscribe call are are hit
+      contextPartial: {
+        dataLoaders: createDataLoaders()
+      },
+    })),
+    // ... other plugins ...
+  ],
+});
+```
+
+Alternatively, you can also provide a callback that is invoked after each [`ExecuteSubscriptionEvent`](<https://spec.graphql.org/draft/#ExecuteSubscriptionEvent()>) phase.
+
+```ts
+import { envelop } from '@envelop/core';
+import { useContextValuePerExecuteSubscriptionEvent } from '@envelop/execute-subscription-event';
+import { createContext, createDataLoaders } from './context';
+
+const getEnveloped = envelop({
+  plugins: [
+    useContext(() => createContext())
+    useContextValuePerExecuteSubscriptionEvent(({ args }) => ({
+      onEnd: () => {
+        // Note that onEnd is invoked only after each ExecuteSubscriptionEvent phase
+        // This means the initial event will still use the cache from potential subscribe dataloader calls
+        // If you use this to clear DataLoader caches it is recommended to not do any DataLoader calls within your field subscribe function.
+        args.contextValue.dataLoaders.users.clearAll()
+        args.contextValue.dataLoaders.posts.clearAll()
+      }
+    })),
+    // ... other plugins ...
+  ],
+});
+```
diff --git a/packages/plugins/execute-subscription-event/package.json b/packages/plugins/execute-subscription-event/package.json
new file mode 100644
index 0000000000..c84d8f36c4
--- /dev/null
+++ b/packages/plugins/execute-subscription-event/package.json
@@ -0,0 +1,49 @@
+{
+  "name": "@envelop/execute-subscription-event",
+  "version": "0.0.0",
+  "author": "Laurin Quast <laurinquast@googlemail.com>",
+  "license": "MIT",
+  "sideEffects": false,
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/dotansimha/envelop.git",
+    "directory": "packages/plugins/disable-introspection"
+  },
+  "main": "dist/index.js",
+  "module": "dist/index.mjs",
+  "exports": {
+    ".": {
+      "require": "./dist/index.js",
+      "import": "./dist/index.mjs"
+    },
+    "./*": {
+      "require": "./dist/*.js",
+      "import": "./dist/*.mjs"
+    }
+  },
+  "typings": "dist/index.d.ts",
+  "typescript": {
+    "definition": "dist/index.d.ts"
+  },
+  "scripts": {
+    "test": "jest",
+    "prepack": "bob prepack"
+  },
+  "dependencies": {},
+  "devDependencies": {
+    "@n1ru4l/push-pull-async-iterable-iterator": "3.0.0",
+    "bob-the-bundler": "1.4.1",
+    "graphql": "15.5.1",
+    "typescript": "4.3.5"
+  },
+  "peerDependencies": {
+    "graphql": "^14.0.0 || ^15.0.0"
+  },
+  "buildOptions": {
+    "input": "./src/index.ts"
+  },
+  "publishConfig": {
+    "directory": "dist",
+    "access": "public"
+  }
+}
diff --git a/packages/plugins/execute-subscription-event/src/index.ts b/packages/plugins/execute-subscription-event/src/index.ts
new file mode 100644
index 0000000000..fe89e19253
--- /dev/null
+++ b/packages/plugins/execute-subscription-event/src/index.ts
@@ -0,0 +1,42 @@
+import { SubscriptionArgs, execute } from 'graphql';
+import { Plugin } from '@envelop/types';
+import { makeExecute, DefaultContext } from '@envelop/core';
+import { PromiseOrValue } from 'graphql/jsutils/PromiseOrValue';
+import { subscribe } from './subscribe';
+
+export type ContextFactoryOptions = {
+  /** The arguments with which the subscription was set up. */
+  args: SubscriptionArgs;
+};
+
+export type ContextFactoryHook<TContextValue> = {
+  /** Context that will be used for the "ExecuteSubscriptionEvent" phase. */
+  contextPartial: Partial<TContextValue>;
+  /** Optional callback that is invoked once the "ExecuteSubscriptionEvent" phase has ended. Useful for cleanup, such as tearing down database connections. */
+  onEnd?: () => void;
+};
+
+export type ContextFactoryType<TContextValue = DefaultContext> = (
+  options: ContextFactoryOptions
+) => PromiseOrValue<ContextFactoryHook<TContextValue> | void>;
+
+export const useExtendContextValuePerExecuteSubscriptionEvent = <TContextValue = unknown>(
+  createContext: ContextFactoryType<TContextValue>
+): Plugin<TContextValue> => {
+  return {
+    onSubscribe({ args, setSubscribeFn }) {
+      const executeNew = makeExecute(async executionArgs => {
+        const context = await createContext({ args });
+        try {
+          return execute({
+            ...executionArgs,
+            contextValue: { ...executionArgs.contextValue, ...context?.contextPartial },
+          });
+        } finally {
+          context?.onEnd?.();
+        }
+      });
+      setSubscribeFn(subscribe(executeNew));
+    },
+  };
+};
diff --git a/packages/plugins/execute-subscription-event/src/subscribe.ts b/packages/plugins/execute-subscription-event/src/subscribe.ts
new file mode 100644
index 0000000000..8bafdc1b9c
--- /dev/null
+++ b/packages/plugins/execute-subscription-event/src/subscribe.ts
@@ -0,0 +1,49 @@
+import { createSourceEventStream } from 'graphql';
+
+import { ExecuteFunction, makeSubscribe, SubscribeFunction } from '@envelop/core';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
+import mapAsyncIterator from 'graphql/subscription/mapAsyncIterator';
+
+/**
+ * This is a almost identical port from graphql-js subscribe.
+ * The only difference is that a custom `execute` function can be injected for customizing the behavior.
+ */
+export const subscribe = (execute: ExecuteFunction): SubscribeFunction =>
+  makeSubscribe(async args => {
+    const { schema, document, rootValue, contextValue, variableValues, operationName, fieldResolver, subscribeFieldResolver } =
+      args;
+
+    const resultOrStream = await createSourceEventStream(
+      schema,
+      document,
+      rootValue,
+      contextValue,
+      variableValues ?? undefined,
+      operationName,
+      subscribeFieldResolver
+    );
+
+    if (!isAsyncIterable(resultOrStream)) {
+      return resultOrStream;
+    }
+
+    // For each payload yielded from a subscription, map it over the normal
+    // GraphQL `execute` function, with `payload` as the rootValue.
+    // This implements the "MapSourceToResponseEvent" algorithm described in
+    // 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.
+    const mapSourceToResponse = async (payload: object) =>
+      execute({
+        schema,
+        document,
+        rootValue: payload,
+        contextValue,
+        variableValues,
+        operationName,
+        fieldResolver,
+      });
+
+    // Map every source value to a ExecutionResult value as described above.
+    return mapAsyncIterator(resultOrStream, mapSourceToResponse);
+  });
diff --git a/packages/plugins/execute-subscription-event/test/use-extend-context-value-per-subscription-event.spec.ts b/packages/plugins/execute-subscription-event/test/use-extend-context-value-per-subscription-event.spec.ts
new file mode 100644
index 0000000000..c93ef11899
--- /dev/null
+++ b/packages/plugins/execute-subscription-event/test/use-extend-context-value-per-subscription-event.spec.ts
@@ -0,0 +1,78 @@
+import { assertStreamExecutionValue, createTestkit } from '@envelop/testing';
+import { schema, subscriptionOperationString } from '../../../core/test/common';
+import { useExtendContextValuePerExecuteSubscriptionEvent } from '../src';
+import { useExtendContext } from '@envelop/core';
+import { makePushPullAsyncIterableIterator } from '@n1ru4l/push-pull-async-iterable-iterator';
+
+describe('useContextValuePerExecuteSubscriptionEvent', () => {
+  it('it can be used for injecting a context that is different from the subscription context', async () => {
+    expect.assertions(4);
+    const { pushValue, asyncIterableIterator } = makePushPullAsyncIterableIterator<unknown>();
+    const subscriptionContextValue: {
+      subscribeSource: AsyncIterableIterator<unknown>;
+      message: string;
+    } = { subscribeSource: asyncIterableIterator, message: 'this is only used during subscribe phase' };
+
+    let counter = 0;
+
+    const testInstance = createTestkit(
+      [
+        useExtendContext(() => subscriptionContextValue),
+        useExtendContextValuePerExecuteSubscriptionEvent(() => ({
+          contextPartial: {
+            message: `${counter}`,
+          },
+        })),
+      ],
+      schema
+    );
+
+    const result = await testInstance.execute(subscriptionOperationString);
+    assertStreamExecutionValue(result);
+
+    pushValue({});
+
+    for await (const value of result) {
+      expect(value.errors).toBeUndefined();
+      if (counter === 0) {
+        expect(value.data?.message).toEqual('0');
+        counter = 1;
+        pushValue({});
+      } else if (counter === 1) {
+        expect(value.data?.message).toEqual('1');
+        return;
+      }
+    }
+  });
+
+  it('invokes cleanup function after value is published', async () => {
+    expect.assertions(3);
+    const { pushValue, asyncIterableIterator } = makePushPullAsyncIterableIterator<unknown>();
+
+    let onEnd = jest.fn();
+    const testInstance = createTestkit(
+      [
+        useExtendContext(() => ({ subscribeSource: asyncIterableIterator })),
+        useExtendContextValuePerExecuteSubscriptionEvent(() => ({
+          contextPartial: {
+            message: `hi`,
+          },
+          onEnd,
+        })),
+      ],
+      schema
+    );
+
+    const result = await testInstance.execute(subscriptionOperationString);
+    assertStreamExecutionValue(result);
+
+    pushValue({});
+
+    for await (const value of result) {
+      expect(value.errors).toBeUndefined();
+      expect(value.data?.message).toEqual('hi');
+      expect(onEnd.mock.calls).toHaveLength(1);
+      return;
+    }
+  });
+});
diff --git a/packages/plugins/fragment-arguments/src/extended-parser.ts b/packages/plugins/fragment-arguments/src/extended-parser.ts
index cc8b78fb73..7be318c9f3 100644
--- a/packages/plugins/fragment-arguments/src/extended-parser.ts
+++ b/packages/plugins/fragment-arguments/src/extended-parser.ts
@@ -1,3 +1,4 @@
+/* eslint-disable @typescript-eslint/explicit-module-boundary-types */
 import { Parser } from 'graphql/language/parser';
 import { Lexer } from 'graphql/language/lexer';
 import { TokenKind, Kind, Source, DocumentNode, TokenKindEnum, Token } from 'graphql';
diff --git a/packages/plugins/newrelic/src/index.ts b/packages/plugins/newrelic/src/index.ts
index 9dd1b2ae24..35edd0ca6e 100644
--- a/packages/plugins/newrelic/src/index.ts
+++ b/packages/plugins/newrelic/src/index.ts
@@ -2,6 +2,7 @@ import { shim as instrumentationApi } from 'newrelic';
 import { Plugin, OnResolverCalledHook } from '@envelop/types';
 import { print, FieldNode, Kind, OperationDefinitionNode } from 'graphql';
 import { Path } from 'graphql/jsutils/Path';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
 
 enum AttributeName {
   COMPONENT_NAME = 'Envelop_NewRelic_Plugin',
@@ -158,6 +159,16 @@ export const useNewRelic = (rawOptions?: UseNewRelicOptions): Plugin => {
       return {
         onResolverCalled,
         onExecuteDone({ result }) {
+          if (isAsyncIterable(result)) {
+            operationSegment.end();
+            // eslint-disable-next-line no-console
+            console.warn(
+              `Plugin "newrelic" encountered a AsyncIterator which is not supported yet, so tracing data is not available for the operation.`
+            );
+
+            return;
+          }
+
           if (result.data && options.includeRawResult) {
             spanContext.addCustomAttribute(AttributeName.EXECUTION_RESULT, JSON.stringify(result));
           }
diff --git a/packages/plugins/opentelemetry/src/index.ts b/packages/plugins/opentelemetry/src/index.ts
index 0e8c28f595..6b5bfbbd66 100644
--- a/packages/plugins/opentelemetry/src/index.ts
+++ b/packages/plugins/opentelemetry/src/index.ts
@@ -3,6 +3,7 @@ import { SpanAttributes, SpanKind } from '@opentelemetry/api';
 import * as opentelemetry from '@opentelemetry/api';
 import { BasicTracerProvider, ConsoleSpanExporter, SimpleSpanProcessor } from '@opentelemetry/tracing';
 import { print } from 'graphql';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
 
 export enum AttributeName {
   EXECUTION_ERROR = 'graphql.execute.error',
@@ -58,6 +59,15 @@ export const useOpenTelemetry = (
 
       const resultCbs: OnExecuteHookResult<PluginContext> = {
         onExecuteDone({ result }) {
+          if (isAsyncIterable(result)) {
+            executionSpan.end();
+            // eslint-disable-next-line no-console
+            console.warn(
+              `Plugin "newrelic" encountered a AsyncIterator which is not supported yet, so tracing data is not available for the operation.`
+            );
+            return;
+          }
+
           if (result.data && options.result) {
             executionSpan.setAttribute(AttributeName.EXECUTION_RESULT, JSON.stringify(result));
           }
diff --git a/packages/plugins/preload-assets/src/index.ts b/packages/plugins/preload-assets/src/index.ts
index d1133a10c3..e7c206a41a 100644
--- a/packages/plugins/preload-assets/src/index.ts
+++ b/packages/plugins/preload-assets/src/index.ts
@@ -1,4 +1,5 @@
 import { Plugin } from '@envelop/types';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
 
 export type UsePreloadAssetsOpts = {
   shouldPreloadAssets?: (context: unknown) => boolean;
@@ -16,14 +17,30 @@ export const usePreloadAssets = (opts?: UsePreloadAssetsOpts): Plugin => ({
       return {
         onExecuteDone: ({ result, setResult }) => {
           if (assets.size) {
-            setResult({
-              ...result,
-              extensions: {
-                ...result.extensions,
-                preloadAssets: Array.from(assets),
-              },
-            });
+            if (isAsyncIterable(result)) {
+              return {
+                onNext(asyncIterableApi) {
+                  asyncIterableApi.setResult({
+                    ...asyncIterableApi.result,
+                    extensions: {
+                      ...asyncIterableApi.result.extensions,
+                      preloadAssets: Array.from(assets),
+                    },
+                  });
+                },
+              };
+            } else {
+              setResult({
+                ...result,
+                extensions: {
+                  ...result.extensions,
+                  preloadAssets: Array.from(assets),
+                },
+              });
+            }
           }
+
+          return undefined;
         },
       };
     }
diff --git a/packages/plugins/prometheus/src/index.ts b/packages/plugins/prometheus/src/index.ts
index 3d04e2edf0..9a0cdec55a 100644
--- a/packages/plugins/prometheus/src/index.ts
+++ b/packages/plugins/prometheus/src/index.ts
@@ -14,6 +14,7 @@ import {
 import { PrometheusTracingPluginConfig } from './config';
 import { TypeInfo } from 'graphql';
 import { isIntrospectionOperationString } from '@envelop/core';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
 
 export { PrometheusTracingPluginConfig, createCounter, createHistogram, createSummary, FillLabelsFnParams };
 
@@ -290,7 +291,7 @@ export const usePrometheus = (config: PrometheusTracingPluginConfig = {}): Plugi
               );
             }
 
-            if (errorsCounter && result.errors && result.errors.length > 0) {
+            if (errorsCounter && !isAsyncIterable(result) && result.errors && result.errors.length > 0) {
               for (const error of result.errors) {
                 errorsCounter.counter
                   .labels(
diff --git a/packages/plugins/sentry/src/index.ts b/packages/plugins/sentry/src/index.ts
index b7a8151bde..4ce9d3987e 100644
--- a/packages/plugins/sentry/src/index.ts
+++ b/packages/plugins/sentry/src/index.ts
@@ -1,10 +1,11 @@
 /* eslint-disable @typescript-eslint/no-empty-function */
 /* eslint-disable no-console */
 /* eslint-disable dot-notation */
-import { Plugin, OnExecuteHookResult, OnResolverCalledHook } from '@envelop/types';
+import { Plugin, OnResolverCalledHook } from '@envelop/types';
 import * as Sentry from '@sentry/node';
 import { Span } from '@sentry/types';
 import { ExecutionArgs, Kind, OperationDefinitionNode, print, responsePathAsArray } from 'graphql';
+import isAsyncIterable from 'graphql/jsutils/isAsyncIterable';
 
 export type SentryPluginOptions = {
   startTransaction?: boolean;
@@ -126,6 +127,15 @@ export const useSentry = (options: SentryPluginOptions = {}): Plugin => {
       return {
         onResolverCalled,
         onExecuteDone({ result }) {
+          if (isAsyncIterable(result)) {
+            rootSpan.finish();
+            // eslint-disable-next-line no-console
+            console.warn(
+              `Plugin "sentry" encountered a AsyncIterator which is not supported yet, so tracing data is not available for the operation.`
+            );
+            return;
+          }
+
           if (includeRawResult) {
             rootSpan.setData('result', result);
           }
diff --git a/packages/testing/src/index.ts b/packages/testing/src/index.ts
index e49b97c8ac..4e51be3be0 100644
--- a/packages/testing/src/index.ts
+++ b/packages/testing/src/index.ts
@@ -177,3 +177,11 @@ export function assertStreamExecutionValue(input: ExecutionReturn): asserts inpu
     throw new Error('Received single result but expected stream.');
   }
 }
+
+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;
+};
diff --git a/packages/types/src/graphql.ts b/packages/types/src/graphql.ts
new file mode 100644
index 0000000000..1a41766561
--- /dev/null
+++ b/packages/types/src/graphql.ts
@@ -0,0 +1,44 @@
+import type {
+  DocumentNode,
+  GraphQLFieldResolver,
+  GraphQLSchema,
+  SubscriptionArgs,
+  ExecutionResult,
+  ExecutionArgs,
+  GraphQLTypeResolver,
+} from 'graphql';
+import type { Maybe, PromiseOrValue, AsyncIterableIteratorOrValue } from './utils';
+
+export type PolymorphicExecuteArguments =
+  | [ExecutionArgs]
+  | [
+      GraphQLSchema,
+      DocumentNode,
+      any,
+      any,
+      Maybe<{ [key: string]: any }>,
+      Maybe<string>,
+      Maybe<GraphQLFieldResolver<any, any>>,
+      Maybe<GraphQLTypeResolver<any, any>>
+    ];
+
+export type ExecuteFunction = (
+  ...args: PolymorphicExecuteArguments
+) => PromiseOrValue<AsyncIterableIteratorOrValue<ExecutionResult>>;
+
+export type PolymorphicSubscribeArguments =
+  | [SubscriptionArgs]
+  | [
+      GraphQLSchema,
+      DocumentNode,
+      any?,
+      any?,
+      Maybe<{ [key: string]: any }>?,
+      Maybe<string>?,
+      Maybe<GraphQLFieldResolver<any, any>>?,
+      Maybe<GraphQLFieldResolver<any, any>>?
+    ];
+
+export type SubscribeFunction = (
+  ...args: PolymorphicSubscribeArguments
+) => PromiseOrValue<AsyncIterableIteratorOrValue<ExecutionResult>>;
diff --git a/packages/types/src/hooks.ts b/packages/types/src/hooks.ts
index 89eac11e36..de9118dc9c 100644
--- a/packages/types/src/hooks.ts
+++ b/packages/types/src/hooks.ts
@@ -1,6 +1,5 @@
-import {
+import type {
   DocumentNode,
-  execute,
   ExecutionArgs,
   ExecutionResult,
   GraphQLError,
@@ -9,7 +8,6 @@ import {
   parse,
   ParseOptions,
   Source,
-  subscribe,
   SubscriptionArgs,
   TypeInfo,
   validate,
@@ -18,6 +16,8 @@ import {
 import { Maybe } from 'graphql/jsutils/Maybe';
 import { PromiseOrValue } from 'graphql/jsutils/PromiseOrValue';
 import { DefaultContext } from './context-types';
+import { AsyncIterableIteratorOrValue, ExecuteFunction } from '@envelop/core';
+import { SubscribeFunction } from './graphql';
 import { Plugin } from './plugin';
 
 export type DefaultArgs = Record<string, unknown>;
@@ -137,7 +137,7 @@ export type OnResolverCalledHook<
 
 /** onExecute */
 export type TypedExecutionArgs<ContextType> = Omit<ExecutionArgs, 'contextValue'> & { contextValue: ContextType };
-export type OriginalExecuteFn = typeof execute;
+export type OriginalExecuteFn = ExecuteFunction;
 export type OnExecuteEventPayload<ContextType> = {
   executeFn: OriginalExecuteFn;
   args: TypedExecutionArgs<ContextType>;
@@ -145,8 +145,21 @@ export type OnExecuteEventPayload<ContextType> = {
   setResultAndStopExecution: (newResult: ExecutionResult) => void;
   extendContext: (contextExtension: Partial<ContextType>) => void;
 };
-export type OnExecuteDoneEventPayload = { result: ExecutionResult; setResult: (newResult: ExecutionResult) => void };
-export type OnExecuteDoneHook = (options: OnExecuteDoneEventPayload) => void;
+export type OnExecuteDoneHookResultOnNextHookPayload = {
+  result: ExecutionResult;
+  setResult: (newResult: ExecutionResult) => void;
+};
+export type OnExecuteDoneHookResultOnNextHook = (payload: OnExecuteDoneHookResultOnNextHookPayload) => void | Promise<void>;
+export type OnExecuteDoneHookResultOnEndHook = () => void;
+export type OnExecuteDoneHookResult = {
+  onNext?: OnExecuteDoneHookResultOnNextHook;
+  onEnd?: OnExecuteDoneHookResultOnEndHook;
+};
+export type OnExecuteDoneEventPayload = {
+  result: AsyncIterableIteratorOrValue<ExecutionResult>;
+  setResult: (newResult: AsyncIterableIteratorOrValue<ExecutionResult>) => void;
+};
+export type OnExecuteDoneHook = (options: OnExecuteDoneEventPayload) => void | OnExecuteDoneHookResult;
 export type OnExecuteHookResult<ContextType> = {
   onExecuteDone?: OnExecuteDoneHook;
   onResolverCalled?: OnResolverCalledHook<any, DefaultArgs, ContextType>;
@@ -158,7 +171,7 @@ export type OnExecuteHook<ContextType> = (
 /** onSubscribe */
 export type TypedSubscriptionArgs<ContextType> = Omit<SubscriptionArgs, 'contextValue'> & { contextValue: ContextType };
 
-export type OriginalSubscribeFn = typeof subscribe;
+export type OriginalSubscribeFn = SubscribeFunction;
 export type OnSubscribeEventPayload<ContextType> = {
   subscribeFn: OriginalSubscribeFn;
   args: TypedSubscriptionArgs<ContextType>;
@@ -169,7 +182,17 @@ export type OnSubscribeResultEventPayload = {
   result: AsyncIterableIterator<ExecutionResult> | ExecutionResult;
   setResult: (newResult: AsyncIterableIterator<ExecutionResult> | ExecutionResult) => void;
 };
-export type SubscribeResultHook = (options: OnSubscribeResultEventPayload) => void;
+export type OnSubscribeResultResultOnNextHookPayload = {
+  result: ExecutionResult;
+  setResult: (newResult: ExecutionResult) => void;
+};
+export type OnSubscribeResultResultOnNextHook = (payload: OnSubscribeResultResultOnNextHookPayload) => void | Promise<void>;
+export type OnSubscribeResultResultOnEndHook = () => void;
+export type OnSubscribeResultResult = {
+  onNext?: OnSubscribeResultResultOnNextHook;
+  onEnd?: OnSubscribeResultResultOnEndHook;
+};
+export type SubscribeResultHook = (options: OnSubscribeResultEventPayload) => void | OnSubscribeResultResult;
 export type OnSubscribeHookResult<ContextType> = {
   onSubscribeResult?: SubscribeResultHook;
   onResolverCalled?: OnResolverCalledHook<ContextType>;
diff --git a/packages/types/src/index.ts b/packages/types/src/index.ts
index aa199b7e4d..f84d34be86 100644
--- a/packages/types/src/index.ts
+++ b/packages/types/src/index.ts
@@ -2,4 +2,5 @@ export * from './context-types';
 export * from './hooks';
 export * from './plugin';
 export * from './get-enveloped';
+export * from './graphql';
 export * from './utils';
diff --git a/packages/types/src/utils.ts b/packages/types/src/utils.ts
index bd82a7b611..57920b749e 100644
--- a/packages/types/src/utils.ts
+++ b/packages/types/src/utils.ts
@@ -27,4 +27,5 @@ export type Unarray<T> = T extends Array<infer U> ? U : T;
 
 export type ArbitraryObject = Record<string | number | symbol, any>;
 export type PromiseOrValue<T> = T | Promise<T>;
+export type AsyncIterableIteratorOrValue<T> = T | AsyncIterableIterator<T>;
 export type Maybe<T> = T | null | undefined;
diff --git a/tsconfig.json b/tsconfig.json
index 9f71e1c874..b996b8b0a9 100644
--- a/tsconfig.json
+++ b/tsconfig.json
@@ -24,15 +24,12 @@
     "resolveJsonModule": true,
     "skipLibCheck": true,
     "paths": {
-      "@envelop/*": [
-        "packages/*/src/index.ts",
-        "packages/plugins/*/src/index.ts"
-      ]
+      "@envelop/core": ["packages/core/src/index.ts"],
+      "@envelop/testing": ["packages/testing/src/index.ts"],
+      "@envelop/types": ["packages/types/src/index.ts"],
+      "@envelop/*": ["packages/plugins/*/src/index.ts"]
     }
   },
   "include": ["packages"],
-  "exclude": [
-    "**/dist",
-    "**/temp"
-  ]
+  "exclude": ["**/dist", "**/temp"]
 }
diff --git a/yarn.lock b/yarn.lock
index 05230ea5aa..a4e7d85507 100644
--- a/yarn.lock
+++ b/yarn.lock
@@ -1941,6 +1941,11 @@
     "@n1ru4l/graphql-live-query" "0.7.1"
     "@n1ru4l/push-pull-async-iterable-iterator" "^2.1.4"
 
+"@n1ru4l/push-pull-async-iterable-iterator@3.0.0":
+  version "3.0.0"
+  resolved "https://registry.yarnpkg.com/@n1ru4l/push-pull-async-iterable-iterator/-/push-pull-async-iterable-iterator-3.0.0.tgz#22dc34094c2de5f21b9a798d0ffab16b45de0eb7"
+  integrity sha512-gwoIwo/Dt1GOI+lbcG1G7IeRM2K+Fo0op3OGyFJ4tXUCf2a3Q8lUCm81aoevrXC0nu4gbAXeOWy7wWxjpSvZUw==
+
 "@n1ru4l/push-pull-async-iterable-iterator@^2.1.4":
   version "2.1.4"
   resolved "https://registry.yarnpkg.com/@n1ru4l/push-pull-async-iterable-iterator/-/push-pull-async-iterable-iterator-2.1.4.tgz#a90225474352f9f159bff979905f707b9c6bcf04"