Repository: mercurius-js/mercurius (js)

Issue:
How to allow queries and mutations over a WebSocket connection?
Hello! 

So I use `fastify` and `fastify-gql` for my API gateway. I have some subscription enabled via the option flag `subscription: true`. The react client uses `import { ApolloClient, HttpLink, InMemoryCache, split } from "@apollo/client"` and `import { WebSocketLink } from "@apollo/link-ws"`. And everything works.

A part of my app is a real-time text editor and pushing operations using POST is just not resulting in a great experience. However, using a `WebSocket connection` works great.

On the server, is there a way to allow the same WebSocket connection (from enabling Subscriptions) to be used to receive and perform a `query` or `mutation` without having two different WebSocket servers side by side?

Right now, the client is setup as follows:

```typescript
import { ApolloClient, HttpLink, InMemoryCache, split } from "@apollo/client"
import { getMainDefinition } from "@apollo/client/utilities"
import { WebSocketLink } from "@apollo/link-ws"

const wsLink = new WebSocketLink({
	uri: `ws://localhost:5000/graphql`,
	options: {
		reconnect: true,
	},
})

const httpLink = new HttpLink({
	uri: "http://localhost:5000/graphql",
})

// The split function takes three parameters:
//
// * A function that's called for each operation to execute
// * The Link to use for an operation if the function returns a "truthy" value
// * The Link to use for an operation if the function returns a "falsy" value

const splitLink = split(
	(op: any) => {
		const { query, operationName } = op
		const definition = getMainDefinition(query)
		return (
			(
				// any mutation's name I want to send via WS ends with "OverWS"
				typeof operationName === `string` && operationName.endsWith(`OverWS`)
			) 
				||
			(
				definition.kind === "OperationDefinition" &&
				definition.operation === "subscription"
			)
		)
	},
	wsLink,
	httpLink
)

export const client = new ApolloClient({
	cache: new InMemoryCache(),
	link: splitLink,
})

```

```typescript
// And logs (Chrome Dev)

/* SENT */ {"type":"connection_init","payload":{}}
/* RECV */ {"type":"connection_ack"}
/* SENT */ {"id":"1","type":"start","payload":{"variables":{"message":"TEST!"},"extensions":{},"operationName":"InsertNotificationOverWS","query":"mutation InsertNotificationOverWS($message: String!) {\n  insertNotification(message: $message)\n}\n"}}
/* RECV */ {"type":"error","id":"1","payload":"Subscription field must return Async Iterable. Received: undefined"}
/* SENT */ {"id":"1","type":"stop"}
```

The server is as in the "Simple Subscription example":

```typescript
import fastify from "fastify"
import cors from "fastify-cors"
import GQL from "fastify-gql"
import { buildSchema } from "graphql"
import { PubSub } from 'graphql-subscriptions'

export const pubsub = new PubSub();

const schemaAsString = `
	type Notification {
		id: ID!
		message: String
	}
   
	type Query {
		notifications: [Notification]
	}
   
	type Mutation {
		insertNotification(message: String): Notification
		addNotification(message: String): Notification
	}
   
	type Subscription {
		notificationAdded: Notification
	}
	 `
	 
const schema = buildSchema(schemaAsString)

let idCount = 1
const notifications = [
	{
		id: idCount,
		message: "Notification message",
	},
]

export const createServer = async (port: number = 5000) => {

	const app = fastify()

	app.register(cors, { origin: true, })

	app.register(GQL, {
		graphiql: true,
		schema,
		resolvers: {
			Query: {
				notifications: () => notifications,
			},
			Mutation: {
				insertNotification: async (_, { message }) => {
					const id = idCount++
					const notification = {
						id,
						message,
					}

					notifications.push(notification)

					pubsub.publish(`NOTIFICATION_ADDED`, { 
						notificationAdded: notification 
					});

					return notification
				}
			},
			Subscription: {
				notificationAdded: {
					subscribe: () => pubsub.asyncIterator(`NOTIFICATION_ADDED`),
				},
			},
		},
		subscription: true,
	})

	app.listen(port)

	console.log(`Listening on port ${port}!`)

	return app
}
```

Thank you! And very appreciative of any pointer!

Pull request:
Queries and Mutations via WebSocket
This proposal adds a new flag, `subscription.fullWsTransport`: when set to `true` we also allow `Query` and `Mutation` root operations via the subscription channel (WebSocket).

There're also some minor changes (docs, etc). _See commits_.

Resolves #195 (and more related I guess, e.g. #532).

Patch:
diff --git a/README.md b/README.md
index a0ec45b..6e46407 100644
--- a/README.md
+++ b/README.md
@@ -34,11 +34,11 @@ Features:
 - [Batched Queries](docs/batched-queries.md)
 - [Persisted Queries](docs/persisted-queries.md)
 - [TypeScript Usage](/docs/typescript.md)
+- [HTTP](/docs/http.md)
+- [GraphQL over WebSocket](/docs/graphql-over-websocket.md)
 - [Integrations](docs/integrations/)
 - [Related Plugins](docs/plugins.md)
-- [Protocol Extensions](/docs/protocol-extension.md)
 - [Faq](/docs/faq.md)
-- [HTTP](/docs/http.md)
 - [Acknowledgements](#acknowledgements)
 - [License](#license)
 
diff --git a/docs/api/options.md b/docs/api/options.md
index b230229..b2160cb 100644
--- a/docs/api/options.md
+++ b/docs/api/options.md
@@ -60,6 +60,7 @@
   - `subscription.onConnect`: `Function` A function which can be used to validate the `connection_init` payload. If defined it should return a truthy value to authorize the connection. If it returns an object the subscription context will be extended with the returned object.
   - `subscription.onDisconnect`: `Function` A function which is called with the subscription context of the connection after the connection gets disconnected.
   - `subscription.keepAlive`: `Integer` Optional interval in ms to send the `GQL_CONNECTION_KEEP_ALIVE` message.
+  - `subscription.fullWsTransport`: `Boolean` Enable sending every operation via WS.
 - `federationMetadata`: Boolean. Enable federation metadata support so the service can be deployed behind an Apollo Gateway
 - `gateway`: Object. Run the GraphQL server in gateway mode.
 
@@ -85,6 +86,8 @@
       - `wsConnectionParams.failedConnectionCallback`: `Function` A function called after a `connection_error` message is received, the first argument contains the message payload.
       - `wsConnectionParams.failedReconnectCallback`: `Function` A function called if reconnect is enabled and maxReconnectAttempts is reached.
       - `wsConnectionParams.rewriteConnectionInitPayload`: `Function` A function that gets the original `connection_init` payload along with the context as a parameter and returns an object that replaces the original `connection_init` payload before forwarding it to the federated service
+  - `gateway.retryServicesCount`: `Number` Specifies the maximum number of retries when a service fails to start on gateway initialization. (Default: 10)
+  - `gateway.retryServicesInterval`: `Number` The amount of time(in milliseconds) between service retry attempts in case a service fails to start on gateway initialization. (Default: 3000)
 
 - `persistedQueries`: A hash/query map to resolve the full query text using it's unique hash. Overrides `persistedQueryProvider`.
 - `onlyPersisted`: Boolean. Flag to control whether to allow graphql queries other than persisted. When `true`, it'll make the server reject any queries that are not present in the `persistedQueries` option above. It will also disable any ide available (graphiql). Requires `persistedQueries` to be set, and overrides `persistedQueryProvider`.
@@ -545,7 +548,7 @@ throw new mercurius.ErrorWithProps('Invalid User ID', {moreErrorInfo}, 200)
 // using the defaultErrorFormatter, the response statusCode will be 200 when there is a single error
 
 const error = new mercurius.ErrorWithProps('Invalid User ID', {moreErrorInfo}, 500)
-error.data = {foo: 'bar'} 
+error.data = {foo: 'bar'}
 throw error
 // using the defaultErrorFormatter, the response status code will be always 200 because error.data is defined
 ```
@@ -558,7 +561,7 @@ By default uses the `defaultErrorFormatter`, but it can be overridden in the [me
 
 **Important**: *using the default formatter, when the error has a data property the response status code will be always 200*
 
-While using custom error formatter, you can access the status code provided by the ErrorWithProps object via 
+While using custom error formatter, you can access the status code provided by the ErrorWithProps object via
 `originalError` property. Please keep in mind though, that `originalError` is a non-enumerable property, meaning it won't
 get serialized and/or logged as a whole.
 
diff --git a/docs/graphql-over-websocket.md b/docs/graphql-over-websocket.md
new file mode 100644
index 0000000..ce966a4
--- /dev/null
+++ b/docs/graphql-over-websocket.md
@@ -0,0 +1,62 @@
+# GraphQL over WebSocket
+
+The GraphQL specification doesn't dictates which transport must be used in order to execute operations. In fact, GraphQL is _transport agnostic_, so implementors can choose which protocol makes more sense for each use case.
+
+Generally, `query` and `mutation` are carried via HTTP, while `subscription` via WebSocket: this is the default behavior in `mercurius` and many other server implementations. However, `query` and `mutation` can also be sent through WebSocket.
+
+## WebSocket subprotocol
+
+As WebSocket is a generic and bidirectional way to send messages, **we need to agree about what each message _means_**: this is defined by the _subprotocol_.
+
+### Supported subprotocols
+
+The GraphQL over WebSocket Protocol (i.e. the WebSocket sub-protocol) used by default is called `graphql-transport-ws` and it's defined here:
+
+- [`graphql-transport-ws` Protocol SPEC](https://github.com/enisdenjo/graphql-ws/blob/master/PROTOCOL.md)
+
+> ⚠️ The subprotocol originally defined by Apollo's `subscriptions-transport-ws` library is also supported. However, that library is **UNMAINTAINED** so it's not recommended to be used: basically **deprecated**. More info [here](https://github.com/apollographql/subscriptions-transport-ws/).
+
+### Supported clients
+
+You should be able to use any major GraphQL client library in order to send operations via WebSocket (e.g. `graphql-ws`, `graphql-hooks`, `apollo`, `urql`…). Depending on which client you use, you have built in support or you may need to use some plugins or middleware.
+
+## Extensions
+
+The `extensions` field is reserved for things that implementors want to add on top of the spec.
+
+### Message structure
+
+This is the structure allowed on each WS message:
+
+```ts
+export interface OperationMessage {
+  payload?: any;
+  id?: string;
+  type: string;
+
+  extensions?: Array<OperationExtension>;
+}
+
+export interface OperationExtension {
+  type: string;
+  payload?: any;
+}
+```
+
+### Server -> Server
+
+In order to achieve _gateway-to-service_ communication and handle `connection_init` per client, an extension is used on the protocol for _server-to-server_ communication. See https://github.com/mercurius-js/mercurius/issues/268 for more details about the original issue and the actual implemented solution.
+
+#### `connectionInit` extension
+
+Gateway uses this extension to share the `connection_init` payload with a service when the connection is already established between gateway and services.
+
+```ts
+export interface ConnectionInitExtension extends OperationExtension {
+  type: string;
+  payload?: Object;
+}
+```
+
+- `type: String` : 'connectionInit'
+- `payload: Object` : optional parameters that the client specifies in connectionParams
diff --git a/docs/protocol-extension.md b/docs/protocol-extension.md
deleted file mode 100644
index 78a25b2..0000000
--- a/docs/protocol-extension.md
+++ /dev/null
@@ -1,38 +0,0 @@
-# GraphQL over WebSocket Protocol extension
-
-The GraphQL over WebSocket Protocol used by `mercurius` follows apollo's protocol defined [here](https://github.com/apollographql/subscriptions-transport-ws/blob/master/PROTOCOL.md). In order to achieve gateway-to-service communication and handle `connection_init` per client, an extension is used on the protocol for server-to-server communication.
-
-## Message structure extension
-
-This is the extended structure of each message:
-
-```ts
-export interface OperationMessage {
-  payload?: any;
-  id?: string;
-  type: string;
-
-  extensions?: Array<OperationExtension>;
-}
-
-export interface OperationExtension {
-  type: string;
-  payload?: any;
-}
-```
-
-## Server -> Server
-
-### `connectionInit` extension
-
-Gateway uses this extension to share the `connection_init` payload with a service when the connection is already established between gateway and services.
-
-```ts
-export interface ConnectionInitExtension extends OperationExtension {
-  type: string;
-  payload?: Object;
-}
-```
-
-- `type: String` : 'connectionInit'
-- `payload: Object` : optional parameters that the client specifies in connectionParams
diff --git a/docsify/sidebar.md b/docsify/sidebar.md
index 1877c10..d8dd761 100644
--- a/docsify/sidebar.md
+++ b/docsify/sidebar.md
@@ -9,7 +9,7 @@
   * [Plugin Options](/docs/api/options#plugin-options)
   * [HTTP Endpoints](/docs/api/options#http-endpoints)
   * [Decorators](/docs/api/options#decorators)
-  * [Error Extensions](/docs/api/options#use-errors-extension-to-provide-additional-information-to-query-errors)
+  * [Error Extensions](/docs/api/options#errors)
 * [Context](/docs/context)
 * [Loaders](/docs/loaders)
 * [Hooks](/docs/hooks)
@@ -20,6 +20,7 @@
 * [Persisted Queries](/docs/persisted-queries)
 * [TypeScript Usage](/docs/typescript)
 * [HTTP](/docs/http)
+* [GraphQL over WebSocket](/docs/graphql-over-websocket.md)
 * [Integrations](/docs/integrations/)
   * [nexus](/docs/integrations/nexus)
   * [TypeGraphQL](/docs/integrations/type-graphql)
@@ -34,5 +35,4 @@
   * [altair-fastify-plugin](/docs/plugins#altair-fastify-plugin)
   * [mercurius-apollo-registry](/docs/plugins#mercurius-apollo-registry)
   * [mercurius-apollo-tracing](/docs/plugins#mercurius-apollo-tracing)
-* [Protocol Extensions](/docs/protocol-extension)
 * [Faq](/docs/faq)
diff --git a/examples/full-ws-transport.js b/examples/full-ws-transport.js
new file mode 100644
index 0000000..a8cfd68
--- /dev/null
+++ b/examples/full-ws-transport.js
@@ -0,0 +1,90 @@
+import Fastify from 'fastify'
+import mercurius from 'mercurius'
+
+const app = Fastify({ logger: true })
+
+let x = 0
+
+const schema = `
+  type ResultChange {
+    operation: String
+    prev: Int
+    current: Int
+  }
+  type Query {
+    result: Int
+  }
+  type Mutation {
+    add(num: Int): Int
+    subtract(num: Int): Int
+  }
+  type Subscription {
+    onResultChange: ResultChange
+  }
+`
+
+const resolvers = {
+  Query: {
+    result: async () => {
+      return x
+    }
+  },
+  Mutation: {
+    add: async (_, args, { pubsub }) => {
+      const prev = x
+      const { num } = args
+
+      x = prev + num
+
+      pubsub.publish({
+        topic: 'RESULT_TOPIC',
+        payload: {
+          onResultChange: {
+            operation: 'add',
+            prev,
+            current: x
+          }
+        }
+      })
+
+      return x
+    },
+    subtract: async (_, args, { pubsub }) => {
+      const prev = x
+      const { num } = args
+
+      x = prev - num
+
+      pubsub.publish({
+        topic: 'RESULT_TOPIC',
+        payload: {
+          onResultChange: {
+            operation: 'subtract',
+            prev,
+            current: x
+          }
+        }
+      })
+
+      return x
+    }
+  },
+  Subscription: {
+    onResultChange: {
+      subscribe: async (_, __, { pubsub }) => {
+        return await pubsub.subscribe('RESULT_TOPIC')
+      }
+    }
+  }
+}
+
+app.register(mercurius, {
+  schema,
+  resolvers,
+  graphiql: true,
+  subscription: {
+    fullWsTransport: true
+  }
+})
+
+app.listen(4000)
diff --git a/examples/gateway-subscription.js b/examples/gateway-subscription.js
index 715fc5c..50d9562 100644
--- a/examples/gateway-subscription.js
+++ b/examples/gateway-subscription.js
@@ -269,7 +269,8 @@ async function start () {
         url: 'http://localhost:4003/graphql',
         wsUrl: 'ws://localhost:4003/graphql',
         wsConnectionParams: {
-          protocols: ['graphql-transport-ws'] // optional, if not set, will use the default protocol (graphql-ws)
+          // OPTIONAL: uncomment this line if you are using the `subscriptions-transport-ws` library
+          // protocols: ['graphql-ws']
         },
         keepAlive: 3000
       }]
diff --git a/index.d.ts b/index.d.ts
index dabe1a1..a52b38f 100644
--- a/index.d.ts
+++ b/index.d.ts
@@ -227,7 +227,7 @@ interface ServiceConfig {
 }
 
 interface Gateway {
-  refresh: () => Promise<GraphQLSchema | null>;
+  refresh: (isRetry?: boolean) => Promise<GraphQLSchema | null>;
   serviceMap: Record<string, ServiceConfig>;
 }
 
@@ -518,6 +518,7 @@ export interface MercuriusCommonOptions {
         }) => Record<string, any> | Promise<Record<string, any>>;
         onDisconnect?: (context: MercuriusContext) => void | Promise<void>;
         keepAlive?: number,
+        fullWsTransport?: boolean,
       };
   /**
    * Enable federation metadata support so the service can be deployed behind an Apollo Gateway
diff --git a/index.js b/index.js
index 0e4d11f..083d6fd 100644
--- a/index.js
+++ b/index.js
@@ -123,6 +123,7 @@ const plugin = fp(async function (app, opts) {
   let onConnect
   let onDisconnect
   let keepAlive
+  let fullWsTransport
 
   if (typeof subscriptionOpts === 'object') {
     if (subscriptionOpts.pubsub) {
@@ -136,6 +137,7 @@ const plugin = fp(async function (app, opts) {
     onConnect = subscriptionOpts.onConnect
     onDisconnect = subscriptionOpts.onDisconnect
     keepAlive = subscriptionOpts.keepAlive
+    fullWsTransport = subscriptionOpts.fullWsTransport
   } else if (subscriptionOpts === true) {
     emitter = mq()
     subscriber = new PubSub(emitter)
@@ -175,9 +177,49 @@ const plugin = fp(async function (app, opts) {
   }
 
   let entityResolversFactory
+  let gatewayRetryIntervalTimer = null
+  const retryServicesCount = gateway && gateway.retryServicesCount ? gateway.retryServicesCount : 10
+
+  const retryServices = (interval) => {
+    let retryCount = 0
+    let isRetry = true
+
+    return setInterval(async () => {
+      try {
+        if (retryCount === retryServicesCount) {
+          clearInterval(gatewayRetryIntervalTimer)
+          isRetry = false
+        }
+        retryCount++
+
+        const context = assignApplicationLifecycleHooksToContext({}, fastifyGraphQl[kHooks])
+        const schema = await gateway.refresh(isRetry)
+        if (schema !== null) {
+          clearInterval(gatewayRetryIntervalTimer)
+          // Trigger onGatewayReplaceSchema hook
+          if (context.onGatewayReplaceSchema !== null) {
+            await onGatewayReplaceSchemaHandler(context, { instance: app, schema })
+          }
+          fastifyGraphQl.replaceSchema(schema)
+        }
+      } catch (error) {
+        app.log.error(error)
+      }
+    }, interval)
+  }
+
   if (gateway) {
+    const retryInterval = gateway.retryServicesInterval || 3000
     gateway = await buildGateway(gateway, app)
 
+    const serviceMap = Object.values(gateway.serviceMap)
+    const failedMandatoryServices = serviceMap.filter(service => !!service.error && service.mandatory)
+
+    if (failedMandatoryServices.length) {
+      gatewayRetryIntervalTimer = retryServices(retryInterval)
+      gatewayRetryIntervalTimer.unref()
+    }
+
     schema = gateway.schema
     entityResolversFactory = gateway.entityResolversFactory
 
@@ -210,6 +252,9 @@ const plugin = fp(async function (app, opts) {
       if (gatewayInterval !== null) {
         clearInterval(gatewayInterval)
       }
+      if (gatewayRetryIntervalTimer !== null) {
+        clearInterval(gatewayRetryIntervalTimer)
+      }
       setImmediate(next)
     })
 
@@ -249,7 +294,8 @@ const plugin = fp(async function (app, opts) {
       lruGatewayResolvers,
       entityResolversFactory,
       subscriptionContextFn,
-      keepAlive
+      keepAlive,
+      fullWsTransport
     })
   }
 
@@ -262,7 +308,9 @@ const plugin = fp(async function (app, opts) {
 
     context = Object.assign(context, { reply: this, app })
     if (app[kFactory]) {
-      this[kLoaders] = app[kFactory].create(context)
+      if (!opts.allowBatchedQueries || !this[kLoaders]) {
+        this[kLoaders] = app[kFactory].create(context)
+      }
     }
 
     return app.graphql(source, context, variables, operationName)
diff --git a/lib/errors.js b/lib/errors.js
index b7888a0..0ca7138 100644
--- a/lib/errors.js
+++ b/lib/errors.js
@@ -205,6 +205,10 @@ const errors = {
     'MER_ERR_GQL_SUBSCRIPTION_MESSAGE_INVALID',
     'Invalid message received: %s'
   ),
+  MER_ERR_GQL_SUBSCRIPTION_INVALID_OPERATION: createError(
+    'MER_ERR_GQL_SUBSCRIPTION_INVALID_OPERATION',
+    'Invalid operation: %s'
+  ),
   /**
    * Hooks errors
    */
@@ -224,6 +228,11 @@ const errors = {
     'MER_ERR_HOOK_UNSUPPORTED_HOOK',
     '%s hook not supported!',
     500
+  ),
+  MER_ERR_SERVICE_RETRY_FAILED: createError(
+    'MER_ERR_SERVICE_RETRY_FAILED',
+    'Mandatory services retry failed - [%s]',
+    500
   )
 }
 
diff --git a/lib/gateway.js b/lib/gateway.js
index 68469d0..e6d4fbc 100644
--- a/lib/gateway.js
+++ b/lib/gateway.js
@@ -16,7 +16,7 @@ const {
   createEntityReferenceResolverOperation,
   kEntityResolvers
 } = require('./gateway/make-resolver')
-const { MER_ERR_GQL_GATEWAY_REFRESH, MER_ERR_GQL_GATEWAY_INIT } = require('./errors')
+const { MER_ERR_GQL_GATEWAY_REFRESH, MER_ERR_GQL_GATEWAY_INIT, MER_ERR_SERVICE_RETRY_FAILED } = require('./errors')
 const findValueTypes = require('./gateway/find-value-types')
 const getQueryResult = require('./gateway/get-query-result')
 const allSettled = require('promise.allsettled')
@@ -257,7 +257,7 @@ function defaultErrorHandler (error, service) {
 async function buildGateway (gatewayOpts, app) {
   const { services, errorHandler = defaultErrorHandler } = gatewayOpts
 
-  const serviceMap = await buildServiceMap(services, errorHandler)
+  const serviceMap = await buildServiceMap(services, errorHandler, app.log)
 
   const serviceSDLs = Object.entries(serviceMap).reduce((acc, [name, value]) => {
     const { schemaDefinition, error } = value
@@ -379,7 +379,8 @@ async function buildGateway (gatewayOpts, app) {
     serviceMap,
     entityResolversFactory: factory,
     pollingInterval: gatewayOpts.pollingInterval,
-    async refresh () {
+    async refresh (isRetry) {
+      const failedMandatoryServices = []
       if (this._serviceSDLs === undefined) {
         this._serviceSDLs = serviceSDLs.join('')
       }
@@ -387,11 +388,27 @@ async function buildGateway (gatewayOpts, app) {
       const $refreshResult = await allSettled(
         Object.values(serviceMap).map((service) =>
           service.refresh().catch((err) => {
-            errorHandler(err, service)
+            // If non-mandatory service or if retry count has exceeded for mandatory service then throw
+            if (!service.mandatory || !isRetry) {
+              errorHandler(err, service)
+            }
+
+            // If service is mandatory and retry count has not exceeded then add to service to
+            // failedMandatoryServices so it can be returned for retrying
+            if (service.mandatory) {
+              failedMandatoryServices.push(service)
+            }
           })
         )
       )
 
+      if (failedMandatoryServices.length > 0) {
+        const serviceNames = failedMandatoryServices.map(service => service.name)
+        const err = new MER_ERR_SERVICE_RETRY_FAILED(serviceNames.join(', '))
+        err.failedServices = serviceNames
+        throw err
+      }
+
       const rejectedResults = $refreshResult
         .filter(({ status }) => status === 'rejected')
         .map(({ reason }) => reason)
diff --git a/lib/gateway/service-map.js b/lib/gateway/service-map.js
index 8200045..7a2ee18 100644
--- a/lib/gateway/service-map.js
+++ b/lib/gateway/service-map.js
@@ -6,7 +6,8 @@ const {
   isTypeDefinitionNode,
   parse,
   buildSchema,
-  GraphQLSchema
+  GraphQLSchema,
+  GraphQLError
 } = require('graphql')
 
 const { buildRequest, sendRequest } = require('./request')
@@ -109,7 +110,7 @@ function safeBuildSchema (schemaDefinition) {
   }
 }
 
-async function buildServiceMap (services, errorHandler) {
+async function buildServiceMap (services, errorHandler, log) {
   const serviceMap = {}
 
   for (const service of services) {
@@ -201,7 +202,10 @@ async function buildServiceMap (services, errorHandler) {
       serviceConfig = await serviceMap[service.name].init()
     } catch (err) {
       serviceConfigErr = err
-      errorHandler(err, service)
+      if (!service.mandatory || err instanceof GraphQLError) {
+        log.warn(`Initializing service "${service.name}" failed with message: "${err.message}"`)
+        errorHandler(err, service)
+      }
     }
 
     if (serviceConfig) {
diff --git a/lib/routes.js b/lib/routes.js
index 429d077..efce708 100644
--- a/lib/routes.js
+++ b/lib/routes.js
@@ -183,7 +183,8 @@ module.exports = async function (app, opts) {
     entityResolversFactory,
     persistedQueryProvider,
     allowBatchedQueries,
-    keepAlive
+    keepAlive,
+    fullWsTransport
   } = opts
 
   // Load the persisted query settings
@@ -308,7 +309,8 @@ module.exports = async function (app, opts) {
       lruGatewayResolvers,
       entityResolversFactory,
       subscriptionContextFn,
-      keepAlive
+      keepAlive,
+      fullWsTransport
     })
   } else {
     app.route(getOptions)
diff --git a/lib/subscription-connection.js b/lib/subscription-connection.js
index 662e39e..d088e66 100644
--- a/lib/subscription-connection.js
+++ b/lib/subscription-connection.js
@@ -1,11 +1,11 @@
 'use strict'
 
 const on = require('events.on')
-const { subscribe, parse, print } = require('graphql')
+const { subscribe, parse, print, getOperationAST } = require('graphql')
 const { SubscriptionContext } = require('./subscriber')
 const { kEntityResolvers } = require('./gateway/make-resolver')
 const sJSON = require('secure-json-parse')
-const { MER_ERR_GQL_SUBSCRIPTION_FORBIDDEN, MER_ERR_GQL_SUBSCRIPTION_UNKNOWN_EXTENSION } = require('./errors')
+const { MER_ERR_GQL_SUBSCRIPTION_FORBIDDEN, MER_ERR_GQL_SUBSCRIPTION_UNKNOWN_EXTENSION, MER_ERR_GQL_SUBSCRIPTION_INVALID_OPERATION } = require('./errors')
 const { preSubscriptionParsingHandler, onSubscriptionResolutionHandler, preSubscriptionExecutionHandler, onSubscriptionEndHandler } = require('./handlers')
 const { kSubscriptionFactory, kLoaders } = require('./symbols')
 const { getProtocolByName } = require('./subscription-protocol')
@@ -20,7 +20,8 @@ module.exports = class SubscriptionConnection {
     onConnect,
     onDisconnect,
     resolveContext,
-    keepAlive
+    keepAlive,
+    fullWsTransport
   }) {
     this.fastify = fastify
     this.socket = socket
@@ -35,6 +36,7 @@ module.exports = class SubscriptionConnection {
     this.isReady = false
     this.resolveContext = resolveContext
     this.keepAlive = keepAlive
+    this.fullWsTransport = fullWsTransport
     this.headers = {}
 
     this.protocolMessageTypes = getProtocolByName(socket.protocol)
@@ -146,10 +148,8 @@ module.exports = class SubscriptionConnection {
   }
 
   async handleGQLStart (data) {
-    // Starting a GraphQL subscription
     const { id, payload, extensions } = data
     const { query, variables, operationName } = payload
-
     let context = { ...this.context }
 
     if (extensions && extensions instanceof Array) {
@@ -171,18 +171,50 @@ module.exports = class SubscriptionConnection {
       }
     }
 
+    const document = typeof query !== 'string' ? query : parse(query)
+
+    if (!document) {
+      throw new Error('Must provide document.')
+    }
+
+    const operationAST = getOperationAST(document, operationName)
+
+    if (operationAST.operation === 'subscription') {
+      await this._executeSubscription({
+        document,
+        query,
+        context,
+        variables,
+        operationName,
+        id
+      })
+    } else if (
+      this.fullWsTransport === true && (operationAST.operation === 'query' || operationAST.operation === 'mutation')) {
+      await this._executeQueryOrMutation({
+        query,
+        context,
+        variables,
+        operationName,
+        id
+      })
+    } else {
+      throw new MER_ERR_GQL_SUBSCRIPTION_INVALID_OPERATION(operationAST.operation)
+    }
+  }
+
+  async _executeSubscription ({ document, query, context, variables, operationName, id }) {
+    const schema = this.fastify.graphql.schema
+
     const sc = new SubscriptionContext({
       fastify: this.fastify,
       pubsub: this.subscriber
     })
 
-    const schema = this.fastify ? this.fastify.graphql.schema : undefined
-
     // Trigger preSubscriptionParsing hook
     if (this.context.preSubscriptionParsing !== null && typeof schema !== 'undefined' && typeof query === 'string') {
       await preSubscriptionParsingHandler({ schema, source: query, context })
     }
-    const document = typeof query !== 'string' ? query : parse(query)
+
     this.subscriptionContexts.set(id, sc)
 
     // Trigger preSubscriptionExecution hook
@@ -191,7 +223,7 @@ module.exports = class SubscriptionConnection {
     }
 
     let subscriptionLoaders
-    if (this.fastify && this.fastify[kSubscriptionFactory]) {
+    if (this.fastify[kSubscriptionFactory]) {
       subscriptionLoaders = this.fastify[kSubscriptionFactory]
     }
 
@@ -209,7 +241,8 @@ module.exports = class SubscriptionConnection {
         reply: {
           entityResolversFactory: this.entityResolversFactory,
           request: { headers: this.headers },
-          [kLoaders]: subscriptionLoaders && subscriptionLoaders.create(context),
+          [kLoaders]:
+            subscriptionLoaders && subscriptionLoaders.create(context),
           [kEntityResolvers]: this.entityResolvers
         }
       },
@@ -234,10 +267,21 @@ module.exports = class SubscriptionConnection {
           return this.handleConnectionClose()
         }
       }
-      this.sendMessage(this.protocolMessageTypes.GQL_DATA, data.id, value)
+      this.sendMessage(this.protocolMessageTypes.GQL_DATA, id, value)
     }
 
-    this.sendMessage(this.protocolMessageTypes.GQL_COMPLETE, data.id, null)
+    this.sendMessage(this.protocolMessageTypes.GQL_COMPLETE, id, null)
+  }
+
+  async _executeQueryOrMutation ({ query, context, variables, operationName, id }) {
+    const response = await this.fastify.graphql(
+      query,
+      context,
+      variables,
+      operationName
+    )
+    this.sendMessage(this.protocolMessageTypes.GQL_DATA, id, response)
+    this.sendMessage(this.protocolMessageTypes.GQL_COMPLETE, id, null)
   }
 
   async handleGQLStop (data) {
@@ -296,7 +340,7 @@ module.exports = class SubscriptionConnection {
       let authorize = false
       try {
         authorize = await this.onConnect({
-          type: 'connection_init',
+          type: this.protocolMessageTypes.GQL_CONNECTION_INIT,
           payload: extension.payload
         })
       } catch (e) {
diff --git a/lib/subscription.js b/lib/subscription.js
index 246470a..7274243 100644
--- a/lib/subscription.js
+++ b/lib/subscription.js
@@ -6,7 +6,7 @@ const { kHooks } = require('./symbols')
 const SubscriptionConnection = require('./subscription-connection')
 const { getProtocolByName } = require('./subscription-protocol')
 
-function createConnectionHandler ({ subscriber, fastify, onConnect, onDisconnect, lruGatewayResolvers, entityResolversFactory, subscriptionContextFn, keepAlive }) {
+function createConnectionHandler ({ subscriber, fastify, onConnect, onDisconnect, lruGatewayResolvers, entityResolversFactory, subscriptionContextFn, keepAlive, fullWsTransport }) {
   return async (connection, request) => {
     const { socket } = connection
 
@@ -45,7 +45,8 @@ function createConnectionHandler ({ subscriber, fastify, onConnect, onDisconnect
       entityResolversFactory,
       context,
       resolveContext,
-      keepAlive
+      keepAlive,
+      fullWsTransport
     })
 
     /* istanbul ignore next */
@@ -59,7 +60,7 @@ function createConnectionHandler ({ subscriber, fastify, onConnect, onDisconnect
 }
 
 module.exports = function (fastify, opts, next) {
-  const { getOptions, subscriber, verifyClient, onConnect, onDisconnect, lruGatewayResolvers, entityResolversFactory, subscriptionContextFn, keepAlive } = opts
+  const { getOptions, subscriber, verifyClient, onConnect, onDisconnect, lruGatewayResolvers, entityResolversFactory, subscriptionContextFn, keepAlive, fullWsTransport } = opts
 
   // If `fastify.websocketServer` exists, it means `fastify-websocket` already registered.
   // Without this check, fastify-websocket will be registered multiple times and raises FST_ERR_DEC_ALREADY_PRESENT.
@@ -82,7 +83,8 @@ module.exports = function (fastify, opts, next) {
       lruGatewayResolvers,
       entityResolversFactory,
       subscriptionContextFn,
-      keepAlive
+      keepAlive,
+      fullWsTransport
     })
   })
 
diff --git a/package.json b/package.json
index bd4b914..23ad423 100644
--- a/package.json
+++ b/package.json
@@ -32,7 +32,7 @@
     "@graphql-tools/merge": "^8.0.0",
     "@graphql-tools/schema": "^8.0.0",
     "@graphql-tools/utils": "^8.0.0",
-    "@sinonjs/fake-timers": "^8.0.1",
+    "@sinonjs/fake-timers": "^9.0.0",
     "@types/node": "^17.0.0",
     "@types/ws": "^8.2.0",
     "@typescript-eslint/eslint-plugin": "^4.28.2",
@@ -41,10 +41,9 @@
     "concurrently": "^7.0.0",
     "docsify-cli": "^4.4.3",
     "fastify": "^3.18.1",
-    "graphql-tools": "^8.0.0",
     "pre-commit": "^1.2.2",
     "proxyquire": "^2.1.3",
-    "sinon": "^12.0.1",
+    "sinon": "^13.0.0",
     "snazzy": "^9.0.0",
     "split2": "^4.0.0",
     "standard": "^16.0.3",


Test patch:
diff --git a/tap-snapshots/test/gateway/remote-services.js.test.cjs b/tap-snapshots/test/gateway/remote-services.js.test.cjs
index f5365fa..4237fc1 100644
--- a/tap-snapshots/test/gateway/remote-services.js.test.cjs
+++ b/tap-snapshots/test/gateway/remote-services.js.test.cjs
@@ -8,3 +8,7 @@
 exports['test/gateway/remote-services.js TAP Does not error if at least one service schema is valid > must match snapshot 1'] = `
 Initializing service "not-working" failed with message: "Unknown type "World"."
 `
+
+exports['test/gateway/remote-services.js TAP Does not error if at least one service schema is valid > must match snapshot 2'] = `
+Initializing service "not-working" failed with message: "Unknown type "World"."
+`
diff --git a/test/gateway/remote-services.js b/test/gateway/remote-services.js
index 6491e07..969475c 100644
--- a/test/gateway/remote-services.js
+++ b/test/gateway/remote-services.js
@@ -120,5 +120,5 @@ test('Does not error if at least one service schema is valid', async (t) => {
   } catch (err) {
     t.error(err)
   }
-  t.equal(warnCalled, 1, 'Warning is called')
+  t.equal(warnCalled, 2, 'Warning is called')
 })
diff --git a/test/gateway/retry-failed-services.js b/test/gateway/retry-failed-services.js
new file mode 100644
index 0000000..74f1cce
--- /dev/null
+++ b/test/gateway/retry-failed-services.js
@@ -0,0 +1,514 @@
+'use strict'
+
+const { test } = require('tap')
+const Fastify = require('fastify')
+const { GraphQLSchema } = require('graphql')
+const GQL = require('../..')
+const FakeTimers = require('@sinonjs/fake-timers')
+
+async function createTestService (port, schema, resolvers = {}) {
+  const service = Fastify()
+  service.register(GQL, {
+    schema,
+    resolvers,
+    federationMetadata: true
+  })
+  await service.listen(port)
+  return service
+}
+
+const users = {
+  u1: {
+    id: 'u1',
+    name: 'John'
+  },
+  u2: {
+    id: 'u2',
+    name: 'Jane'
+  }
+}
+
+const posts = {
+  p1: {
+    pid: 'p1',
+    title: 'Post 1',
+    content: 'Content 1',
+    authorId: 'u1'
+  },
+  p2: {
+    pid: 'p2',
+    title: 'Post 2',
+    content: 'Content 2',
+    authorId: 'u2'
+  },
+  p3: {
+    pid: 'p3',
+    title: 'Post 3',
+    content: 'Content 3',
+    authorId: 'u1'
+  },
+  p4: {
+    pid: 'p4',
+    title: 'Post 4',
+    content: 'Content 4',
+    authorId: 'u1'
+  }
+}
+
+const userService = {
+  schema: `
+  extend type Query {
+    me: User
+  }
+  
+  type User @key(fields: "id") {
+    id: ID!
+    name: String!
+  }
+  `,
+  resolvers: {
+    Query: {
+      me: () => {
+        return users.u1
+      }
+    },
+    User: {
+      __resolveReference: user => {
+        return users[user.id]
+      }
+    }
+  }
+}
+
+const postService = {
+  schema: `
+  type Post @key(fields: "pid") {
+    pid: ID!
+    title: String
+    content: String
+    author: User @requires(fields: "pid title")
+  }
+
+  type User @key(fields: "id") @extends {
+    id: ID! @external
+    name: String @external
+    posts(count: Int): [Post]
+  }
+`,
+  resolvers: {
+    Post: {
+      author: post => {
+        return {
+          __typename: 'User',
+          id: post.authorId
+        }
+      }
+    },
+    User: {
+      posts: (user, { count }) => {
+        return Object.values(posts).filter(p => p.authorId === user.id).slice(0, count)
+      }
+    }
+  }
+}
+
+test('gateway - retry mandatory failed services on startup', async (t) => {
+  t.plan(5)
+  const clock = FakeTimers.install({
+    shouldAdvanceTime: true,
+    advanceTimeDelta: 100
+  })
+
+  const service1 = await createTestService(5001, userService.schema, userService.resolvers)
+
+  let service2 = null
+  setTimeout(async () => {
+    service2 = await createTestService(5002, postService.schema, postService.resolvers)
+  }, 5000)
+
+  const app = Fastify()
+  t.teardown(async () => {
+    await app.close()
+    await service1.close()
+    await service2.close()
+    clock.uninstall()
+  })
+
+  await app.register(GQL, {
+    jit: 1,
+    gateway: {
+      services: [
+        {
+          name: 'user',
+          url: 'http://localhost:5001/graphql',
+          mandatory: false
+        },
+        {
+          name: 'post',
+          url: 'http://localhost:5002/graphql',
+          mandatory: true
+        }
+      ]
+    }
+  })
+
+  app.graphql.addHook('onGatewayReplaceSchema', async (instance, schema) => {
+    t.type(instance, 'object')
+    t.type(schema, GraphQLSchema)
+    t.ok('should be called')
+  })
+
+  await app.ready()
+
+  const query = `
+    query {
+      user: me {
+        id
+        name
+        posts(count: 1) {
+          pid
+        }
+      }
+    }`
+
+  const res = await app.inject({
+    method: 'POST',
+    headers: { 'content-type': 'application/json' },
+    url: '/graphql',
+    body: JSON.stringify({ query })
+  })
+
+  t.same(JSON.parse(res.body), {
+    errors: [
+      {
+        message: 'Cannot query field "posts" on type "User".',
+        locations: [{ line: 6, column: 9 }]
+      }
+    ],
+    data: null
+  })
+
+  for (let i = 0; i < 10; i++) {
+    await clock.tickAsync(2000)
+  }
+
+  const res1 = await app.inject({
+    method: 'POST',
+    headers: { 'content-type': 'application/json' },
+    url: '/graphql',
+    body: JSON.stringify({ query })
+  })
+
+  t.same(JSON.parse(res1.body), {
+    data: {
+      user: {
+        id: 'u1',
+        name: 'John',
+        posts: [
+          {
+            pid: 'p1'
+          }
+        ]
+      }
+    }
+  })
+})
+
+test('gateway - should not call onGatewayReplaceSchemaHandler if the hook is not specified', async (t) => {
+  t.plan(2)
+  const clock = FakeTimers.install({
+    shouldAdvanceTime: true,
+    advanceTimeDelta: 100
+  })
+
+  const service1 = await createTestService(5001, userService.schema, userService.resolvers)
+
+  let service2 = null
+  setTimeout(async () => {
+    service2 = await createTestService(5002, postService.schema, postService.resolvers)
+  }, 5000)
+
+  const app = Fastify()
+  t.teardown(async () => {
+    await app.close()
+    await service1.close()
+    await service2.close()
+    clock.uninstall()
+  })
+
+  await app.register(GQL, {
+    jit: 1,
+    gateway: {
+      services: [
+        {
+          name: 'user',
+          url: 'http://localhost:5001/graphql',
+          mandatory: false
+        },
+        {
+          name: 'post',
+          url: 'http://localhost:5002/graphql',
+          mandatory: true
+        }
+      ],
+      retryServicesCount: 10,
+      retryServicesInterval: 2000
+    }
+  })
+
+  await app.ready()
+
+  const query = `
+    query {
+      user: me {
+        id
+        name
+        posts(count: 1) {
+          pid
+        }
+      }
+    }`
+
+  const res = await app.inject({
+    method: 'POST',
+    headers: { 'content-type': 'application/json' },
+    url: '/graphql',
+    body: JSON.stringify({ query })
+  })
+
+  t.same(JSON.parse(res.body), {
+    errors: [
+      {
+        message: 'Cannot query field "posts" on type "User".',
+        locations: [{ line: 6, column: 9 }]
+      }
+    ],
+    data: null
+  })
+
+  for (let i = 0; i < 10; i++) {
+    await clock.tickAsync(1000)
+  }
+
+  const res1 = await app.inject({
+    method: 'POST',
+    headers: { 'content-type': 'application/json' },
+    url: '/graphql',
+    body: JSON.stringify({ query })
+  })
+
+  t.same(JSON.parse(res1.body), {
+    data: {
+      user: {
+        id: 'u1',
+        name: 'John',
+        posts: [
+          {
+            pid: 'p1'
+          }
+        ]
+      }
+    }
+  })
+})
+
+test('gateway - dont retry non-mandatory failed services on startup', async (t) => {
+  t.plan(2)
+  const clock = FakeTimers.install({
+    shouldAdvanceTime: true,
+    advanceTimeDelta: 100
+  })
+
+  const service1 = await createTestService(5001, userService.schema, userService.resolvers)
+
+  const app = Fastify()
+  t.teardown(async () => {
+    await app.close()
+    await service1.close()
+    clock.uninstall()
+  })
+
+  app.register(GQL, {
+    jit: 1,
+    gateway: {
+      services: [
+        {
+          name: 'user',
+          url: 'http://localhost:5001/graphql',
+          mandatory: false
+        },
+        {
+          name: 'post',
+          url: 'http://localhost:5002/graphql',
+          mandatory: false
+        }
+      ],
+      pollingInterval: 2000
+    }
+  })
+
+  await app.ready()
+
+  const query = `
+    query {
+      user: me {
+        id
+        name
+        posts(count: 1) {
+          pid
+        }
+      }
+    }`
+
+  const res = await app.inject({
+    method: 'POST',
+    headers: { 'content-type': 'application/json' },
+    url: '/graphql',
+    body: JSON.stringify({ query })
+  })
+
+  t.same(JSON.parse(res.body), {
+    errors: [
+      {
+        message: 'Cannot query field "posts" on type "User".',
+        locations: [{ line: 6, column: 9 }]
+      }
+    ],
+    data: null
+  })
+
+  for (let i = 0; i < 10; i++) {
+    await clock.tickAsync(1500)
+  }
+
+  const res1 = await app.inject({
+    method: 'POST',
+    headers: { 'content-type': 'application/json' },
+    url: '/graphql',
+    body: JSON.stringify({ query })
+  })
+
+  t.same(JSON.parse(res1.body), {
+    errors: [
+      {
+        message: 'Cannot query field "posts" on type "User".',
+        locations: [{ line: 6, column: 9 }]
+      }
+    ],
+    data: null
+  })
+})
+
+test('gateway - should log error if retry throws', async (t) => {
+  t.plan(1)
+  const clock = FakeTimers.install({
+    shouldAdvanceTime: true,
+    advanceTimeDelta: 100
+  })
+
+  const service1 = await createTestService(5001, userService.schema, userService.resolvers)
+
+  let service2 = null
+  setTimeout(async () => {
+    service2 = await createTestService(5002, postService.schema, postService.resolvers)
+  }, 2000)
+
+  const app = Fastify()
+
+  let errCount = 0
+  app.log.error = (error) => {
+    if (error.message.includes('kaboom') && errCount === 0) {
+      errCount++
+      t.pass()
+    }
+  }
+
+  t.teardown(async () => {
+    await app.close()
+    await service1.close()
+    await service2.close()
+    clock.uninstall()
+  })
+
+  await app.register(GQL, {
+    jit: 1,
+    gateway: {
+      services: [
+        {
+          name: 'user',
+          url: 'http://localhost:5001/graphql',
+          mandatory: false
+        },
+        {
+          name: 'post',
+          url: 'http://localhost:5002/graphql',
+          mandatory: true
+        }
+      ],
+      retryServicesCount: 1,
+      retryServicesInterval: 2000
+    }
+  })
+
+  app.graphql.addHook('onGatewayReplaceSchema', async () => {
+    throw new Error('kaboom')
+  })
+
+  await app.ready()
+
+  for (let i = 0; i < 10; i++) {
+    await clock.tickAsync(1000)
+  }
+})
+
+test('gateway - stop retrying after no. of retries exceeded', async (t) => {
+  t.plan(1)
+  const clock = FakeTimers.install({
+    shouldAdvanceTime: true,
+    advanceTimeDelta: 100
+  })
+
+  const service1 = await createTestService(5001, userService.schema, userService.resolvers)
+
+  const app = Fastify()
+
+  let errCount = 0
+  app.log.error = (error) => {
+    if (error.code === 'MER_ERR_GQL_GATEWAY_REFRESH' && errCount === 0) {
+      errCount++
+      t.pass()
+    }
+  }
+
+  t.teardown(async () => {
+    await app.close()
+    await service1.close()
+    clock.uninstall()
+  })
+
+  await app.register(GQL, {
+    jit: 1,
+    gateway: {
+      services: [
+        {
+          name: 'user',
+          url: 'http://localhost:5001/graphql',
+          mandatory: false
+        },
+        {
+          name: 'post',
+          url: 'http://localhost:5002/graphql',
+          mandatory: true
+        }
+      ],
+      retryServicesCount: 1,
+      retryServicesInterval: 2000
+    }
+  })
+
+  await app.ready()
+
+  for (let i = 0; i < 10; i++) {
+    await clock.tickAsync(1500)
+  }
+})
diff --git a/test/loaders.js b/test/loaders.js
index 530ac11..6cac9cf 100644
--- a/test/loaders.js
+++ b/test/loaders.js
@@ -40,6 +40,8 @@ const schema = `
 
   type Query {
     dogs: [Dog]
+    puppies: [Dog]
+    elders: [Dog]
   }
 
   type Subscription {
@@ -51,6 +53,12 @@ const resolvers = {
   Query: {
     dogs (_, params, { reply }) {
       return dogs
+    },
+    puppies (_, params, { reply }) {
+      return [dogs[0]]
+    },
+    elders (_, params, { reply }) {
+      return dogs.slice(2, dogs.length)
     }
   }
 }
@@ -64,6 +72,24 @@ const query = `{
   }
 }`
 
+const puppiesQuery = `{
+  puppies {
+    name,
+    owner {
+      name
+    }
+  }
+}`
+
+const eldersQuery = `{
+  elders {
+    name,
+    owner {
+      name
+    }
+  }
+}`
+
 test('loaders create batching resolvers', async (t) => {
   const app = Fastify()
 
@@ -136,6 +162,71 @@ test('loaders create batching resolvers', async (t) => {
   })
 })
 
+test('loaders create batching resolvers with batchedQueries', async (t) => {
+  const app = Fastify()
+
+  const loaders = {
+    Dog: {
+      async owner (queries, { reply }) {
+        // note that the second entry for max is cached
+        t.same(queries, [{
+          obj: {
+            name: 'Max'
+          },
+          params: {}
+        }, {
+          obj: {
+            name: 'Buddy'
+          },
+          params: {}
+        }])
+        return queries.map(({ obj }) => {
+          return { ...owners[obj.name] }
+        })
+      }
+    }
+  }
+
+  app.register(GQL, {
+    schema,
+    resolvers,
+    loaders,
+    allowBatchedQueries: true
+  })
+
+  const res = await app.inject({
+    method: 'POST',
+    url: '/graphql',
+    body: [{ query: puppiesQuery }, { query: eldersQuery }]
+  })
+
+  t.equal(res.statusCode, 200)
+  t.same(JSON.parse(res.body), [{
+    data: {
+      puppies: [{
+        name: 'Max',
+        owner: {
+          name: 'Jennifer'
+        }
+      }]
+    }
+  }, {
+    data: {
+      elders: [{
+        name: 'Buddy',
+        owner: {
+          name: 'Tracy'
+        }
+      }, {
+        name: 'Max',
+        owner: {
+          name: 'Jennifer'
+        }
+      }]
+    }
+  }])
+})
+
 test('disable cache for each loader', async (t) => {
   const app = Fastify()
 
diff --git a/test/subscription-connection.js b/test/subscription-connection.js
index 579cb45..dd60a5e 100644
--- a/test/subscription-connection.js
+++ b/test/subscription-connection.js
@@ -1,6 +1,7 @@
 'use strict'
 const { test } = require('tap')
 const proxyquire = require('proxyquire')
+const sinon = require('sinon')
 const WebSocket = require('ws')
 const fastify = require('fastify')
 const mq = require('mqemitter')
@@ -390,6 +391,9 @@ test('subscription connection send GQL_ERROR message if connectionInit extension
     }
   })
 
+  // TODO FIXME? This case doesn't seem to be allowed by the protocol: https://github.com/enisdenjo/graphql-ws/blob/master/PROTOCOL.md#connectioninit
+  // When onConnect returns a "falsy" value then it CANNOT be "ready", it should just deny the connection.
+  // However, this is an "extension"; if it's valid, perhaps should be better documented?
   sc.isReady = true
   await sc.handleMessage(JSON.stringify({
     id: 1,
@@ -448,6 +452,9 @@ test('subscription connection send GQL_ERROR on unknown extension', async (t) =>
   }, { })
 
   sc.isReady = true
+  // TODO FIXME? This case doesn't seem to be allowed by the protocol: https://github.com/enisdenjo/graphql-ws/blob/master/PROTOCOL.md#subscribe
+  // When type is "Subscribe", the `payload` cannot be empty (the `query` prop is required).
+  // However, this is an "extension"; if it's valid, perhaps should be better documented?
   await sc.handleMessage(JSON.stringify({
     id: 1,
     type: 'subscribe',
@@ -501,3 +508,138 @@ test('subscription connection extends the context with the connection_init paylo
 
   t.same(sc.context._connectionInit, connectionInitPayload)
 })
+
+test('subscription connection sends error when trying to execute invalid operations via WS', async (t) => {
+  t.plan(1)
+
+  const sc = new SubscriptionConnection(
+    {
+      on () {},
+      send (message) {
+        t.equal(
+          JSON.stringify({
+            type: 'error',
+            id: 1,
+            payload: 'Invalid operation: query'
+          }),
+          message
+        )
+      },
+      protocol: GRAPHQL_TRANSPORT_WS
+    },
+    {}
+  )
+
+  sc.isReady = true
+  await sc.handleMessage(
+    JSON.stringify({
+      id: 1,
+      type: 'subscribe',
+      payload: {
+        query: 'query { __typename }'
+      }
+    })
+  )
+})
+
+test('subscription connection handles query when fullWsTransport: true', async (t) => {
+  const send = sinon.stub()
+
+  const sc = new SubscriptionConnection(
+    {
+      on () {},
+      send,
+      protocol: GRAPHQL_TRANSPORT_WS
+    },
+    {
+      fastify: {
+        graphql: () => {
+          return {}
+        }
+      },
+      fullWsTransport: true
+    }
+  )
+
+  sc.isReady = true
+
+  await sc.handleMessage(
+    JSON.stringify({
+      id: 1,
+      type: 'subscribe',
+      payload: {
+        query: 'query { __typename }'
+      }
+    })
+  )
+
+  t.ok(
+    send.withArgs(
+      JSON.stringify({
+        type: 'next',
+        id: 1,
+        payload: {}
+      })
+    ).calledOnce
+  )
+  t.ok(
+    send.withArgs(
+      JSON.stringify({
+        type: 'complete',
+        id: 1,
+        payload: null
+      })
+    ).calledOnce
+  )
+})
+
+test('subscription connection handles mutation when fullWsTransport: true', async (t) => {
+  const send = sinon.stub()
+
+  const sc = new SubscriptionConnection(
+    {
+      on () {},
+      send,
+      protocol: GRAPHQL_TRANSPORT_WS
+    },
+    {
+      fastify: {
+        graphql: () => {
+          return {}
+        }
+      },
+      fullWsTransport: true
+    }
+  )
+
+  sc.isReady = true
+
+  await sc.handleMessage(
+    JSON.stringify({
+      id: 1,
+      type: 'subscribe',
+      payload: {
+        query: 'mutation { __typename }'
+      }
+    })
+  )
+
+  t.ok(
+    send.withArgs(
+      JSON.stringify({
+        type: 'next',
+        id: 1,
+        payload: {}
+      })
+    ).calledOnce
+  )
+  t.ok(
+    send.withArgs(
+      JSON.stringify({
+        type: 'complete',
+        id: 1,
+        payload: null
+      })
+    ).calledOnce
+  )
+})
diff --git a/test/types/index.ts b/test/types/index.ts
index 6ffa431..2adbc42 100644
--- a/test/types/index.ts
+++ b/test/types/index.ts
@@ -466,6 +466,14 @@ app.register(mercurius, {
   }
 })
 
+app.register(mercurius, {
+  schema: schema,
+  resolvers,
+  subscription: {
+    fullWsTransport: true
+  }
+})
+
 app.register(mercurius, {
   schema,
   resolvers,