feat(url-loader): add support for legacy subscription client (#2894)

* feat(url-loader): add support for legacy subscription client

* fix(url-loader): pass headers to subscription client

* feat(url-loader): pass connectionParams into ws client creation instead of headers only

* chore: cleanup

* chore(url-loader): added unit test for legacy ws subscriber

* chore: added changeset

Author: ntziolis

.changeset/lemon-days-drop.md

@@ -0,0 +1,6 @@
+---
+'@graphql-tools/url-loader': minor
+---
+
+- Added support for legacy ws protocol
+- Ensured that headers are passed into ws connection params

packages/loaders/url/package.json

@@ -33,22 +33,23 @@
   },
   "dependencies": {
     "@graphql-tools/delegate": "^7.0.1",
-    "@graphql-tools/wrap": "^7.0.4",
     "@graphql-tools/utils": "^7.1.5",
+    "@graphql-tools/wrap": "^7.0.4",
     "@types/websocket": "1.0.2",
     "cross-fetch": "3.1.4",
-    "extract-files": "9.0.0",
     "eventsource": "1.1.0",
+    "extract-files": "9.0.0",
+    "form-data": "4.0.0",
     "graphql-upload": "^11.0.0",
+    "graphql-ws": "^4.4.1",
     "is-promise": "4.0.0",
     "isomorphic-ws": "4.0.1",
-    "form-data": "4.0.0",
+    "sse-z": "0.3.0",
+    "subscriptions-transport-ws": "^0.9.18",
+    "sync-fetch": "0.3.0",
     "tslib": "~2.2.0",
     "valid-url": "1.0.9",
-    "graphql-ws": "^4.4.1",
-    "ws": "7.4.5",
-    "sync-fetch": "0.3.0",
-    "sse-z": "0.3.0"
+    "ws": "7.4.5"
   },
   "publishConfig": {
     "access": "public",

packages/loaders/url/src/index.ts

@@ -1,6 +1,15 @@
 /* eslint-disable no-case-declarations */
 /// <reference lib="dom" />
-import { print, IntrospectionOptions, DocumentNode, GraphQLResolveInfo, Kind, parse, buildASTSchema } from 'graphql';
+import {
+  print,
+  IntrospectionOptions,
+  DocumentNode,
+  GraphQLResolveInfo,
+  Kind,
+  parse,
+  buildASTSchema,
+  ExecutionResult,
+} from 'graphql';
 import {
   AsyncExecutor,
   Executor,
@@ -17,7 +26,7 @@ import { isWebUri } from 'valid-url';
 import { fetch as crossFetch } from 'cross-fetch';
 import { SubschemaConfig } from '@graphql-tools/delegate';
 import { introspectSchema, wrapSchema } from '@graphql-tools/wrap';
-import { createClient } from 'graphql-ws';
+import { ClientOptions, createClient } from 'graphql-ws';
 import WebSocket from 'isomorphic-ws';
 import syncFetch from 'sync-fetch';
 import isPromise from 'is-promise';
@@ -26,6 +35,7 @@ import FormData from 'form-data';
 import 'eventsource/lib/eventsource-polyfill';
 import { Subscription, SubscriptionOptions } from 'sse-z';
 import { URL } from 'url';
+import { ConnectionParamsOptions, SubscriptionClient as LegacySubscriptionClient } from 'subscriptions-transport-ws';
 
 export type AsyncFetchFn = typeof import('cross-fetch').fetch;
 export type SyncFetchFn = (input: RequestInfo, init?: RequestInit) => SyncResponse;
@@ -86,6 +96,10 @@ export interface LoadFromUrlOptions extends SingleFileOptions, Partial<Introspec
    * Use SSE for subscription instead of WebSocket
    */
   useSSEForSubscription?: boolean;
+  /**
+   * Use legacy web socket protocol `graphql-ws` instead of the more current standard `graphql-transport-ws`
+   */
+  useWebSocketLegacyProtocol?: boolean;
   /**
    * Additional options to pass to the constructor of the underlying EventSource instance.
    */
@@ -254,14 +268,19 @@ export class UrlLoader implements DocumentLoader<LoadFromUrlOptions> {
     return executor;
   }
 
-  buildWSSubscriber(pointer: string, webSocketImpl: typeof WebSocket): Subscriber {
+  buildWSSubscriber(
+    pointer: string,
+    webSocketImpl: typeof WebSocket,
+    connectionParams: ClientOptions['connectionParams']
+  ): Subscriber {
     const WS_URL = switchProtocols(pointer, {
       https: 'wss',
       http: 'ws',
     });
     const subscriptionClient = createClient({
       url: WS_URL,
       webSocketImpl,
+      connectionParams,
     });
     return async ({ document, variables }: { document: DocumentNode; variables: any }) => {
       const query = print(document);
@@ -282,6 +301,33 @@ export class UrlLoader implements DocumentLoader<LoadFromUrlOptions> {
     };
   }
 
+  buildWSLegacySubscriber(
+    pointer: string,
+    webSocketImpl: typeof WebSocket,
+    connectionParams: ConnectionParamsOptions
+  ): Subscriber {
+    const WS_URL = switchProtocols(pointer, {
+      https: 'wss',
+      http: 'ws',
+    });
+    const subscriptionClient = new LegacySubscriptionClient(
+      WS_URL,
+      {
+        connectionParams,
+      },
+      webSocketImpl
+    );
+
+    return async <TReturn, TArgs>({ document, variables }: { document: DocumentNode; variables: TArgs }) => {
+      return observableToAsyncIterable(
+        subscriptionClient.request({
+          query: document,
+          variables,
+        })
+      ) as AsyncIterator<ExecutionResult<TReturn>>;
+    };
+  }
+
   buildSSESubscriber(pointer: string, eventSourceOptions?: SubscriptionOptions['eventSourceOptions']): Subscriber {
     return async ({ document, variables }: { document: DocumentNode; variables: any }) => {
       const query = print(document);
@@ -413,7 +459,12 @@ export class UrlLoader implements DocumentLoader<LoadFromUrlOptions> {
       subscriber = this.buildSSESubscriber(pointer, options.eventSourceOptions);
     } else {
       const webSocketImpl = await this.getWebSocketImpl(options, asyncImport);
-      subscriber = this.buildWSSubscriber(pointer, webSocketImpl);
+
+      if (options.useWebSocketLegacyProtocol) {
+        subscriber = this.buildWSLegacySubscriber(pointer, webSocketImpl, { headers });
+      } else {
+        subscriber = this.buildWSSubscriber(pointer, webSocketImpl, { headers });
+      }
     }
 
     return {
@@ -448,7 +499,11 @@ export class UrlLoader implements DocumentLoader<LoadFromUrlOptions> {
       subscriber = this.buildSSESubscriber(pointer, options.eventSourceOptions);
     } else {
       const webSocketImpl = this.getWebSocketImpl(options, syncImport);
-      subscriber = this.buildWSSubscriber(pointer, webSocketImpl);
+      if (options.useWebSocketLegacyProtocol) {
+        subscriber = this.buildWSLegacySubscriber(pointer, webSocketImpl, { headers });
+      } else {
+        subscriber = this.buildWSSubscriber(pointer, webSocketImpl, { headers });
+      }
     }
 
     return {

packages/loaders/url/tests/url-loader.spec.ts

@@ -12,6 +12,7 @@ import { join } from 'path';
 import { useServer } from 'graphql-ws/lib/use/ws';
 import { Server as WSServer } from 'ws';
 import http from 'http';
+import { SubscriptionServer } from 'subscriptions-transport-ws';
 
 const SHOULD_NOT_GET_HERE_ERROR = 'SHOULD_NOT_GET_HERE';
 
@@ -333,7 +334,7 @@ input TestInput {
       expect(result.document).toBeDefined();
       expect(print(result.document)).toBeSimilarGqlDoc(testTypeDefs);
     })
-    it('should handle subscriptions', async (done) => {
+    it('should handle subscriptions - new protocol', async (done) => {
       const testUrl = 'http://localhost:8081/graphql';
       const { schema } = await loader.load(testUrl, {
         customFetch: async () => ({
@@ -398,6 +399,68 @@ input TestInput {
       });
 
     });
+    it('should handle subscriptions - legacy protocol', async (done) => {
+      const testUrl = 'http://localhost:8081/graphql';
+      const { schema } = await loader.load(testUrl, {
+        customFetch: async () => ({
+          json: async () => ({
+            data: introspectionFromSchema(testSchema),
+          })
+        }) as any,
+        useWebSocketLegacyProtocol: true,
+      });
+
+      const httpServer = http.createServer(function weServeSocketsOnly(_, res) {
+        res.writeHead(404);
+        res.end();
+      });
+
+
+      httpServer.listen(8081);
+
+      const subscriptionServer = SubscriptionServer.create(
+        {
+          schema: testSchema,
+          execute,
+          subscribe,
+        },
+        {
+          server: httpServer,
+          path: '/graphql',
+        },
+      );
+
+      const asyncIterator = await subscribe({
+        schema,
+        document: parse(/* GraphQL */`
+          subscription TestMessage {
+            testMessage {
+              number
+            }
+          }
+        `),
+        contextValue: {},
+      }) as AsyncIterableIterator<ExecutionResult>;
+
+      expect(asyncIterator['errors']).toBeFalsy();
+      expect(asyncIterator['errors']?.length).toBeFalsy();
+
+
+      // eslint-disable-next-line no-inner-declarations
+      async function getNextResult() {
+        const result = await asyncIterator.next();
+        expect(result?.done).toBeFalsy();
+        return result?.value?.data?.testMessage?.number;
+      }
+
+      expect(await getNextResult()).toBe(0);
+      expect(await getNextResult()).toBe(1);
+      expect(await getNextResult()).toBe(2);
+
+      await asyncIterator.return();
+      subscriptionServer.close();
+      httpServer.close(done);
+    });
     it('should handle multipart requests', async () => {
       let server = mockGraphQLServer({ schema: testSchema, host: testHost, path: testPathChecker, method: 'POST' });