Cache variables with the operations when transforms exist on the root level even if variables change in the further requests with the same operation

@@ -0,0 +1,5 @@
+---
+'@graphql-mesh/transform-federation': patch
+---
+
+Remove other directives in scalars just like it is done for objects and other types

@@ -0,0 +1,5 @@
+---
+'@graphql-mesh/runtime': patch
+---
+
+Simplify the logic and use GraphQL Tools executor

@@ -0,0 +1,5 @@
+---
+'@graphql-mesh/grpc': patch
+---
+
+Add response streams as subscriptions

@@ -0,0 +1,5 @@
+---
+'@graphql-mesh/runtime': patch
+---
+
+Do not cache entire request but only DocumentNode

@@ -0,0 +1,8 @@
+subscription SearchMoviesByCast {
+  exampleSearchMoviesByCast(input: { castName: "Tom Cruise" }) {
+    name
+    year
+    rating
+    cast
+  }
+}

@@ -36,7 +36,18 @@ describe('gRPC Example', () => {
     const result = await mesh.execute(MoviesByCastStream, undefined);
     let i = 0;
     for await (const item of result as AsyncIterable<any>) {
-      expect(item).toMatchSnapshot(`movies-by-cast-grpc-example-result-${i++}`);
+      expect(item).toMatchSnapshot(`movies-by-cast-grpc-example-result-stream-${i++}`);
+    }
+  });
+  it('should fetch movies by cast as a subscription correctly', async () => {
+    const MoviesByCastSubscription = await readFile(
+      join(__dirname, '../example-queries/MoviesByCast.subscription.graphql'),
+      'utf8',
+    );
+    const result = await mesh.execute(MoviesByCastSubscription, undefined);
+    let i = 0;
+    for await (const item of result as AsyncIterable<any>) {
+      expect(item).toMatchSnapshot(`movies-by-cast-grpc-example-result-subscription-${i++}`);
     }
   });
   afterAll(async () => {

@@ -1,6 +1,6 @@
 // Jest Snapshot v1, https://goo.gl/fbAQLP
 
-exports[`gRPC Example should fetch movies by cast as a stream correctly: movies-by-cast-grpc-example-result-0 1`] = `
+exports[`gRPC Example should fetch movies by cast as a stream correctly: movies-by-cast-grpc-example-result-stream-0 1`] = `
 {
   "data": {
     "exampleSearchMoviesByCast": [],
@@ -9,7 +9,7 @@ exports[`gRPC Example should fetch movies by cast as a stream correctly: movies-
 }
 `;
 
-exports[`gRPC Example should fetch movies by cast as a stream correctly: movies-by-cast-grpc-example-result-1 1`] = `
+exports[`gRPC Example should fetch movies by cast as a stream correctly: movies-by-cast-grpc-example-result-stream-1 1`] = `
 {
   "hasNext": true,
   "incremental": [
@@ -35,7 +35,7 @@ exports[`gRPC Example should fetch movies by cast as a stream correctly: movies-
 }
 `;
 
-exports[`gRPC Example should fetch movies by cast as a stream correctly: movies-by-cast-grpc-example-result-2 1`] = `
+exports[`gRPC Example should fetch movies by cast as a stream correctly: movies-by-cast-grpc-example-result-stream-2 1`] = `
 {
   "hasNext": true,
   "incremental": [
@@ -61,15 +61,50 @@ exports[`gRPC Example should fetch movies by cast as a stream correctly: movies-
 }
 `;
 
-exports[`gRPC Example should fetch movies by cast as a stream correctly: movies-by-cast-grpc-example-result-3 1`] = `
+exports[`gRPC Example should fetch movies by cast as a stream correctly: movies-by-cast-grpc-example-result-stream-3 1`] = `
 {
   "hasNext": false,
 }
 `;
 
+exports[`gRPC Example should fetch movies by cast as a subscription correctly: movies-by-cast-grpc-example-result-subscription-0 1`] = `
+{
+  "data": {
+    "exampleSearchMoviesByCast": {
+      "cast": [
+        "Tom Cruise",
+        "Simon Pegg",
+        "Jeremy Renner",
+      ],
+      "name": "Mission: Impossible Rogue Nation",
+      "rating": 0.9700000286102295,
+      "year": 2015,
+    },
+  },
+}
+`;
+
+exports[`gRPC Example should fetch movies by cast as a subscription correctly: movies-by-cast-grpc-example-result-subscription-1 1`] = `
+{
+  "data": {
+    "exampleSearchMoviesByCast": {
+      "cast": [
+        "Tom Cruise",
+        "Simon Pegg",
+        "Henry Cavill",
+      ],
+      "name": "Mission: Impossible - Fallout",
+      "rating": 0.9300000071525574,
+      "year": 2018,
+    },
+  },
+}
+`;
+
 exports[`gRPC Example should generate correct schema: grpc-schema 1`] = `
 "schema {
   query: Query
+  subscription: Subscription
 }
 
 directive @grpcMethod(rootJsonName: String, objPath: String, methodName: String, responseStream: Boolean) on FIELD_DEFINITION
@@ -166,6 +201,13 @@ enum ConnectivityState {
   SHUTDOWN
 }
 
+type Subscription {
+  "search movies by the name of the cast"
+  exampleSearchMoviesByCast(input: SearchByCastRequest_Input): Movie @grpcMethod(rootJsonName: "Root0", objPath: "Example", methodName: "SearchMoviesByCast", responseStream: true)
+  "search movies by the name of the cast"
+  anotherExampleSearchMoviesByCast(input: SearchByCastRequest_Input): Movie @grpcMethod(rootJsonName: "Root0", objPath: "AnotherExample", methodName: "SearchMoviesByCast", responseStream: true)
+}
+
 scalar ObjMap"
 `;
 

@@ -419,11 +419,22 @@ export default class GrpcHandler implements MeshHandler {
                   objPath,
                   creds,
                 });
-                field.resolve = this.getFieldResolver({
-                  client,
-                  methodName,
-                  isResponseStream: responseStream,
-                });
+                if (rootType.name === 'Subscription') {
+                  field.subscribe = this.getFieldResolver({
+                    client,
+                    methodName,
+                    isResponseStream: responseStream,
+                  });
+                  field.resolve = function identityFn(root) {
+                    return root;
+                  };
+                } else {
+                  field.resolve = this.getFieldResolver({
+                    client,
+                    methodName,
+                    isResponseStream: responseStream,
+                  });
+                }
                 break;
               }
               case 'grpcConnectivityState': {
@@ -598,26 +609,29 @@ export default class GrpcHandler implements MeshHandler {
       for (const methodName in nested.methods) {
         const method = nested.methods[methodName];
         const rootFieldName = [...pathWithName, methodName].join('_');
+        const fieldConfigTypeFactory = () => {
+          const baseResponseTypePath = method.responseType?.split('.');
+          if (baseResponseTypePath) {
+            const responseTypePath = this.walkToFindTypePath(
+              rootJson,
+              pathWithName,
+              baseResponseTypePath,
+            );
+            return getTypeName(this.schemaComposer, responseTypePath, false);
+          }
+          return 'Void';
+        };
         const fieldConfig: ObjectTypeComposerFieldConfigAsObjectDefinition<any, any> = {
           type: () => {
-            const baseResponseTypePath = method.responseType?.split('.');
-            if (baseResponseTypePath) {
-              const responseTypePath = this.walkToFindTypePath(
-                rootJson,
-                pathWithName,
-                baseResponseTypePath,
-              );
-              let typeName = getTypeName(this.schemaComposer, responseTypePath, false);
-              if (method.responseStream) {
-                typeName = `[${typeName}]`;
-              }
-              return typeName;
+            const typeName = fieldConfigTypeFactory();
+            if (method.responseStream) {
+              return `[${typeName}]`;
             }
-            return 'Void';
+            return typeName;
           },
           description: method.comment,
         };
-        fieldConfig.args = {
+        const fieldConfigArgs = {
           input: () => {
             if (method.requestStream) {
               return 'File';
@@ -635,6 +649,7 @@ export default class GrpcHandler implements MeshHandler {
             return undefined;
           },
         };
+        fieldConfig.args = fieldConfigArgs;
         const methodNameLowerCased = methodName.toLowerCase();
         const prefixQueryMethod = this.config.prefixQueryMethod || QUERY_METHOD_PREFIXES;
         const rootTypeComposer = prefixQueryMethod.some(prefix =>
@@ -659,6 +674,26 @@ export default class GrpcHandler implements MeshHandler {
             ],
           },
         });
+        if (method.responseStream) {
+          this.schemaComposer.Subscription.addFields({
+            [rootFieldName]: {
+              args: fieldConfigArgs,
+              description: method.comment,
+              type: fieldConfigTypeFactory,
+              directives: [
+                {
+                  name: 'grpcMethod',
+                  args: {
+                    rootJsonName,
+                    objPath,
+                    methodName,
+                    responseStream: true,
+                  },
+                },
+              ],
+            },
+          });
+        }
       }
       const connectivityStateFieldName = pathWithName.join('_') + '_connectivityState';
       this.schemaComposer.addDirective(grpcConnectivityStateDirective);

@@ -187,6 +187,7 @@ scalar ObjMap"
 exports[`gRPC Handler Interpreting Protos should load the Empty proto 1`] = `
 "schema {
   query: Query
+  subscription: Subscription
 }
 
 directive @enum(value: String) on ENUM_VALUE
@@ -286,6 +287,13 @@ enum ConnectivityState {
   SHUTDOWN
 }
 
+type Subscription {
+  "search movies by the name of the cast"
+  io_xtech_Example_SearchMoviesByCast(input: io_xtech_SearchByCastRequest_Input): io_xtech_Movie @grpcMethod(rootJsonName: "Root0", objPath: "io.xtech.Example", methodName: "SearchMoviesByCast", responseStream: true)
+  "search movies by the name of the cast"
+  io_xtech_AnotherExample_SearchMoviesByCast(input: io_xtech_SearchByCastRequest_Input): io_xtech_Movie @grpcMethod(rootJsonName: "Root0", objPath: "io.xtech.AnotherExample", methodName: "SearchMoviesByCast", responseStream: true)
+}
+
 scalar ObjMap"
 `;
 
@@ -453,6 +461,7 @@ scalar ObjMap"
 exports[`gRPC Handler Interpreting Protos should load the Movie proto 1`] = `
 "schema {
   query: Query
+  subscription: Subscription
 }
 
 directive @enum(value: String) on ENUM_VALUE
@@ -549,6 +558,13 @@ enum ConnectivityState {
   SHUTDOWN
 }
 
+type Subscription {
+  "search movies by the name of the cast"
+  io_xtech_Example_SearchMoviesByCast(input: io_xtech_SearchByCastRequest_Input): io_xtech_Movie @grpcMethod(rootJsonName: "Root0", objPath: "io.xtech.Example", methodName: "SearchMoviesByCast", responseStream: true)
+  "search movies by the name of the cast"
+  io_xtech_AnotherExample_SearchMoviesByCast(input: io_xtech_SearchByCastRequest_Input): io_xtech_Movie @grpcMethod(rootJsonName: "Root0", objPath: "io.xtech.AnotherExample", methodName: "SearchMoviesByCast", responseStream: true)
+}
+
 scalar ObjMap"
 `;
 
@@ -661,6 +677,7 @@ scalar ObjMap"
 exports[`gRPC Handler Interpreting Protos should load the Outside proto 1`] = `
 "schema {
   query: Query
+  subscription: Subscription
 }
 
 directive @grpcMethod(rootJsonName: String, objPath: String, methodName: String, responseStream: Boolean) on FIELD_DEFINITION
@@ -767,6 +784,13 @@ input io_xtech_SearchByCastRequest_Input {
   castName: String
 }
 
+type Subscription {
+  "search movies by the name of the cast"
+  io_xtech_Example_SearchMoviesByCast(input: io_xtech_SearchByCastRequest_Input): io_xtech_Movie @grpcMethod(rootJsonName: "Root0", objPath: "io.xtech.Example", methodName: "SearchMoviesByCast", responseStream: true)
+  "search movies by the name of the cast"
+  io_xtech_AnotherExample_SearchMoviesByCast(input: io_xtech_SearchByCastRequest_Input): io_xtech_Movie @grpcMethod(rootJsonName: "Root0", objPath: "io.xtech.AnotherExample", methodName: "SearchMoviesByCast", responseStream: true)
+}
+
 scalar ObjMap"
 `;
 
@@ -859,6 +883,7 @@ scalar ObjMap"
 exports[`gRPC Handler Interpreting Protos should load the With Underscores proto 1`] = `
 "schema {
   query: Query
+  subscription: Subscription
 }
 
 directive @enum(value: String) on ENUM_VALUE
@@ -955,13 +980,21 @@ enum ConnectivityState {
   SHUTDOWN
 }
 
+type Subscription {
+  "search movies by the name of the cast"
+  io_xtech_Example_SearchMoviesByCast(input: io_xtech_SearchByCastRequest_Input): io_xtech_Movie @grpcMethod(rootJsonName: "Root0", objPath: "io.xtech.Example", methodName: "SearchMoviesByCast", responseStream: true)
+  "search movies by the name of the cast"
+  io_xtech_AnotherExample_SearchMoviesByCast(input: io_xtech_SearchByCastRequest_Input): io_xtech_Movie @grpcMethod(rootJsonName: "Root0", objPath: "io.xtech.AnotherExample", methodName: "SearchMoviesByCast", responseStream: true)
+}
+
 scalar ObjMap"
 `;
 
 exports[`gRPC Handler Load proto with prefixQueryMethod should load the retrieve-movie.proto 1`] = `
 "schema {
   query: Query
   mutation: Mutation
+  subscription: Subscription
 }
 
 directive @enum(value: String) on ENUM_VALUE
@@ -1069,5 +1102,12 @@ input io_xtech_SearchByCastRequest_Input {
   castName: String
 }
 
+type Subscription {
+  "search movies by the name of the cast"
+  io_xtech_Example_SearchMoviesByCast(input: io_xtech_SearchByCastRequest_Input): io_xtech_Movie @grpcMethod(rootJsonName: "Root0", objPath: "io.xtech.Example", methodName: "SearchMoviesByCast", responseStream: true)
+  "search movies by the name of the cast"
+  io_xtech_AnotherExample_SearchMoviesByCast(input: io_xtech_SearchByCastRequest_Input): io_xtech_Movie @grpcMethod(rootJsonName: "Root0", objPath: "io.xtech.AnotherExample", methodName: "SearchMoviesByCast", responseStream: true)
+}
+
 scalar ObjMap"
 `;

@@ -46,6 +46,7 @@
     "@graphql-mesh/string-interpolation": "^0.5.2",
     "@graphql-tools/batch-delegate": "^9.0.0",
     "@graphql-tools/delegate": "^10.0.0",
+    "@graphql-tools/executor": "^1.2.0",
     "@graphql-tools/wrap": "^10.0.0",
     "@whatwg-node/fetch": "^0.9.0",
     "graphql-jit": "0.8.2"

@@ -1,6 +1,5 @@
 import {
   DocumentNode,
-  execute,
   getOperationAST,
   GraphQLObjectType,
   GraphQLSchema,
@@ -35,10 +34,10 @@ import {
   PubSub,
 } from '@graphql-mesh/utils';
 import { CreateProxyingResolverFn, Subschema, SubschemaConfig } from '@graphql-tools/delegate';
+import { normalizedExecutor } from '@graphql-tools/executor';
 import {
   ExecutionResult,
   getRootTypeMap,
-  inspect,
   isAsyncIterable,
   isPromise,
   mapAsyncIterator,
@@ -76,14 +75,6 @@ const memoizedGetEnvelopedFactory = memoize1(function getEnvelopedFactory(
   });
 });
 
-const memoizedGetOperationType = memoize1((document: DocumentNode) => {
-  const operationAST = getOperationAST(document, undefined);
-  if (!operationAST) {
-    throw new Error('Must provide document with a valid operation');
-  }
-  return operationAST.operation;
-});
-
 export function wrapFetchWithPlugins(plugins: MeshPlugin<any>[]): MeshFetch {
   const onFetchHooks: OnFetchHook<any>[] = [];
   for (const plugin of plugins as MeshPlugin<any>[]) {
@@ -92,24 +83,6 @@ export function wrapFetchWithPlugins(plugins: MeshPlugin<any>[]): MeshFetch {
     }
   }
   return function wrappedFetchFn(url, options, context, info) {
-    if (url != null && typeof url !== 'string') {
-      throw new TypeError(`First parameter(url) of 'fetch' must be a string, got ${inspect(url)}`);
-    }
-    if (options != null && typeof options !== 'object') {
-      throw new TypeError(
-        `Second parameter(options) of 'fetch' must be an object, got ${inspect(options)}`,
-      );
-    }
-    if (context != null && typeof context !== 'object') {
-      throw new TypeError(
-        `Third parameter(context) of 'fetch' must be an object, got ${inspect(context)}`,
-      );
-    }
-    if (info != null && typeof info !== 'object') {
-      throw new TypeError(
-        `Fourth parameter(info) of 'fetch' must be an object, got ${inspect(info)}`,
-      );
-    }
     let fetchFn: MeshFetch;
     const doneHooks: OnFetchHookDone[] = [];
     function setFetchFn(newFetchFn: MeshFetch) {
@@ -318,7 +291,7 @@ export async function getMesh(options: GetMeshOptions): Promise<MeshInstance> {
 
   const plugins = [
     useEngine({
-      execute,
+      execute: normalizedExecutor,
       validate,
       parse: parseWithCache,
       specifiedRules,
@@ -377,8 +350,13 @@ export async function getMesh(options: GetMeshOptions): Promise<MeshInstance> {
       operationName?: string,
     ) {
       const document = typeof documentOrSDL === 'string' ? parse(documentOrSDL) : documentOrSDL;
-      const executeFn = memoizedGetOperationType(document) === 'subscription' ? subscribe : execute;
       const contextValue$ = contextFactory(contextValue);
+      const operationAST = getOperationAST(document, operationName);
+      if (!operationAST) {
+        throw new Error(`Cannot execute a request without a valid operation.`);
+      }
+      const isSubscription = operationAST.operation === 'subscription';
+      const executeFn = isSubscription ? subscribe : execute;
       if (isPromise(contextValue$)) {
         return contextValue$.then(contextValue =>
           executeFn({

@@ -55,13 +55,7 @@ const getIntrospectionOperationType = memoize1(function getIntrospectionOperatio
 
 function getExecuteFn(subschema: Subschema) {
   const compiledQueryCache = new WeakMap<DocumentNode, CompiledQuery>();
-  const transformedRequestCache = new WeakMap<
-    DocumentNode,
-    {
-      transformedRequest: ExecutionRequest;
-      transformationContext: Record<string, any>;
-    }
-  >();
+  const transformedDocumentNodeCache = new WeakMap<DocumentNode, DocumentNode>();
   return function subschemaExecute(args: TypedExecutionArgs<any>): any {
     const originalRequest: ExecutionRequest = {
       document: args.document,
@@ -101,7 +95,7 @@ function getExecuteFn(subschema: Subschema) {
     };
     let executor = subschema.executor;
     if (executor == null) {
-      if (isStream) {
+      if (isStream || operationAST.operation === 'subscription') {
         executor = createDefaultExecutor(subschema.schema);
       } else {
         executor = function subschemaExecutor(request: ExecutionRequest): any {
@@ -158,41 +152,41 @@ function getExecuteFn(subschema: Subschema) {
       executor = createBatchingExecutor(executor);
     }
     */
-    let transformedRequestAndContext = transformedRequestCache.get(originalRequest.document);
-    if (!transformedRequestAndContext) {
-      const transformationContext: Record<string, any> = {};
-      const transformedRequest = applyRequestTransforms(
-        originalRequest,
-        delegationContext,
-        transformationContext,
-        subschema.transforms,
-      );
-      transformedRequestAndContext = {
-        transformedRequest,
-        transformationContext,
-      };
-      transformedRequestCache.set(originalRequest.document, transformedRequestAndContext);
+    const transformationContext: Record<string, any> = {};
+    const transformedRequest = applyRequestTransforms(
+      originalRequest,
+      delegationContext,
+      transformationContext,
+      subschema.transforms,
+    );
+    const cachedTransfomedDocumentNode: DocumentNode = transformedDocumentNodeCache.get(
+      originalRequest.document,
+    );
+    if (cachedTransfomedDocumentNode) {
+      transformedRequest.document = cachedTransfomedDocumentNode;
+    } else {
+      transformedDocumentNodeCache.set(originalRequest.document, transformedRequest.document);
     }
     function handleResult(originalResult: MaybeAsyncIterable<ExecutionResult>) {
       if (isAsyncIterable(originalResult)) {
         return mapAsyncIterator(originalResult, singleResult =>
           applyResultTransforms(
             singleResult,
             delegationContext,
-            transformedRequestAndContext.transformationContext,
+            transformationContext,
             subschema.transforms,
           ),
         );
       }
       const transformedResult = applyResultTransforms(
         originalResult,
         delegationContext,
-        transformedRequestAndContext.transformationContext,
+        transformationContext,
         subschema.transforms,
       );
       return transformedResult;
     }
-    const originalResult$ = executor(transformedRequestAndContext.transformedRequest);
+    const originalResult$ = executor(transformedRequest);
     if (isPromise(originalResult$)) {
       return originalResult$.then(handleResult);
     }

@@ -2,9 +2,11 @@ import { dset } from 'dset';
 import {
   GraphQLInterfaceType,
   GraphQLObjectType,
+  GraphQLScalarType,
   GraphQLSchema,
   GraphQLUnionType,
   isObjectType,
+  isSpecifiedScalarType,
 } from 'graphql';
 import { entitiesField, EntityType, serviceField } from '@apollo/subgraph/dist/types.js';
 import { stringInterpolator } from '@graphql-mesh/string-interpolation';
@@ -275,6 +277,28 @@ export default class FederationTransform implements MeshTransform {
           },
         };
       },
+      [MapperKind.SCALAR_TYPE]: type => {
+        if (isSpecifiedScalarType(type)) {
+          return type;
+        }
+        return new GraphQLScalarType({
+          ...type.toConfig(),
+          astNode: type.astNode && {
+            ...type.astNode,
+            directives: type.astNode.directives?.filter(directive =>
+              federationDirectives.includes(directive.name.value),
+            ),
+          },
+          extensions: {
+            ...type.extensions,
+            directives: Object.fromEntries(
+              Object.entries(type.extensions?.directives || {}).filter(([key]) =>
+                federationDirectives.includes(key),
+              ),
+            ),
+          },
+        });
+      },
     });
   }
 }

@@ -3287,7 +3287,7 @@
     tslib "^2.4.0"
     ws "8.14.1"
 
-"@graphql-tools/executor@^1.0.0":
+"@graphql-tools/executor@^1.0.0", "@graphql-tools/executor@^1.2.0":
   version "1.2.0"
   resolved "https://registry.yarnpkg.com/@graphql-tools/executor/-/executor-1.2.0.tgz#6c45f4add765769d9820c4c4405b76957ba39c79"
   integrity sha512-SKlIcMA71Dha5JnEWlw4XxcaJ+YupuXg0QCZgl2TOLFz4SkGCwU/geAsJvUJFwK2RbVLpQv/UMq67lOaBuwDtg==