fix/feat(serve): WebSockets for GW/Client comm

.changeset/@graphql-mesh_serve-cli-7766-dependencies.md

@@ -0,0 +1,6 @@
+---
+"@graphql-mesh/serve-cli": patch
+---
+dependencies updates:
+  - Added dependency [`graphql-ws@^5.16.0` ↗︎](https://www.npmjs.com/package/graphql-ws/v/5.16.0) (to `dependencies`)
+  - Added dependency [`ws@^8.18.0` ↗︎](https://www.npmjs.com/package/ws/v/8.18.0) (to `dependencies`)

.changeset/lazy-spies-watch.md

@@ -0,0 +1,7 @@
+---
+'@graphql-mesh/transport-http-callback': minor
+'@graphql-mesh/serve-runtime': minor
+'@graphql-mesh/serve-cli': minor
+---
+
+Fix/Implement WebSockets support for the Gateway and Client communication

e2e/federation-subscriptions-passthrough/federation-subscriptions-passthrough.test.ts

@@ -1,55 +1,82 @@
 import { setTimeout } from 'timers/promises';
-import { createClient, type Client } from 'graphql-sse';
+import {
+  createClient as createSSEClient,
+  type Client as SSEClient,
+  type ClientOptions as SSEClientOptions,
+} from 'graphql-sse';
+import {
+  createClient as createWSClient,
+  type Client as WSClient,
+  type ClientOptions as WSClientOptions,
+} from 'graphql-ws';
+import webSocketImpl from 'ws';
+import { getLocalHostName } from '@e2e/opts';
 import { createTenv, getAvailablePort } from '@e2e/tenv';
 import { fetch } from '@whatwg-node/fetch';
 import { TOKEN } from './services/products/server';
 
 const { composeWithApollo, service, serve } = createTenv(__dirname);
 
-let client: Client | null = null;
+let client: WSClient | SSEClient | null = null;
 
 afterEach(() => client?.dispose());
 
-it('should subscribe and resolve via websockets', async () => {
-  const supergraphFile = await composeWithApollo([
-    await service('products'),
-    await service('reviews'),
-  ]);
-  const { port } = await serve({ supergraph: supergraphFile });
-
-  client = createClient({
-    url: `http://localhost:${port}/graphql`,
-    retryAttempts: 0,
-    headers: {
-      Authorization: TOKEN,
-    },
-    fetchFn: fetch,
-  });
-  const sub = client.iterate({
-    query: /* GraphQL */ `
-      subscription OnProductPriceChanged {
-        productPriceChanged {
-          # Defined in Products subgraph
-          name
-          price
-          reviews {
-            # Defined in Reviews subgraph
-            score
+const subscriptionsClientFactories = [
+  ['SSE', createSSEClient],
+  ['WS', createWSClient],
+] as [
+  string,
+  (opts: Partial<SSEClientOptions> & Partial<WSClientOptions>) => SSEClient | WSClient,
+][];
+
+subscriptionsClientFactories.forEach(([protocol, createClient]) => {
+  if (protocol === 'WS' && process.version.startsWith('v18')) {
+    return;
+  }
+  describe(`with ${protocol}`, () => {
+    const headers = {
+      authorization: TOKEN,
+    };
+    it('should subscribe and resolve via websockets', async () => {
+      const supergraphFile = await composeWithApollo([
+        await service('products'),
+        await service('reviews'),
+      ]);
+      const { port } = await serve({ supergraph: supergraphFile });
+
+      client = createClient({
+        url: `http://localhost:${port}/graphql`,
+        retryAttempts: 0,
+        headers,
+        connectionParams: headers,
+        fetchFn: fetch,
+        webSocketImpl,
+      });
+      const sub = client.iterate({
+        query: /* GraphQL */ `
+          subscription OnProductPriceChanged {
+            productPriceChanged {
+              # Defined in Products subgraph
+              name
+              price
+              reviews {
+                # Defined in Reviews subgraph
+                score
+              }
+            }
           }
+        `,
+      });
+
+      const msgs = [];
+      for await (const msg of sub) {
+        msgs.push(msg);
+        if (msgs.length >= 3) {
+          break;
         }
       }
-    `,
-  });
-
-  const msgs = [];
-  for await (const msg of sub) {
-    msgs.push(msg);
-    if (msgs.length >= 3) {
-      break;
-    }
-  }
 
-  expect(msgs).toMatchInlineSnapshot(`
+      expect(msgs).toMatchInlineSnapshot(`
 [
   {
     "data": {
@@ -95,111 +122,111 @@ it('should subscribe and resolve via websockets', async () => {
   },
 ]
 `);
-});
+    });
 
-it('should recycle websocket connections', async () => {
-  const supergraphFile = await composeWithApollo([
-    await service('products'),
-    await service('reviews'),
-  ]);
-  const { port } = await serve({ supergraph: supergraphFile });
-
-  client = createClient({
-    url: `http://localhost:${port}/graphql`,
-    retryAttempts: 0,
-    headers: {
-      Authorization: TOKEN,
-    },
-    fetchFn: fetch,
-  });
+    it('should recycle websocket connections', async () => {
+      const supergraphFile = await composeWithApollo([
+        await service('products'),
+        await service('reviews'),
+      ]);
+      const { port } = await serve({ supergraph: supergraphFile });
+
+      client = createClient({
+        url: `http://localhost:${port}/graphql`,
+        retryAttempts: 0,
+        headers,
+        connectionParams: headers,
+        fetchFn: fetch,
+        webSocketImpl,
+      });
+
+      const query = /* GraphQL */ `
+        subscription OnProductPriceChanged {
+          productPriceChanged {
+            price
+          }
+        }
+      `;
+      for (let i = 0; i < 5; i++) {
+        // connect
+        for await (const msg of client.iterate({ query })) {
+          expect(msg).toMatchObject({
+            data: expect.any(Object),
+          });
+          break; // complete subscription on first received message
+        }
+        // disconnect
 
-  const query = /* GraphQL */ `
-    subscription OnProductPriceChanged {
-      productPriceChanged {
-        price
+        await setTimeout(300); // wait a bit and subscribe again (lazyCloseTimeout is 3 seconds)
       }
-    }
-  `;
-  for (let i = 0; i < 5; i++) {
-    // connect
-    for await (const msg of client.iterate({ query })) {
-      expect(msg).toMatchObject({
-        data: expect.any(Object),
-      });
-      break; // complete subscription on first received message
-    }
-    // disconnect
 
-    await setTimeout(300); // wait a bit and subscribe again (lazyCloseTimeout is 3 seconds)
-  }
+      // the "products" service will crash if multiple websockets were connected breaking the loop above with an error
+    });
 
-  // the "products" service will crash if multiple websockets were connected breaking the loop above with an error
-});
+    it('should subscribe and resolve via http callbacks', async () => {
+      const supergraphFile = await composeWithApollo([
+        await service('products'),
+        await service('reviews'),
+      ]);
+
+      // Get a random available port
+      const availablePort = await getAvailablePort();
+
+      const publicUrl = `http://${getLocalHostName()}:${availablePort}`;
+      await serve({
+        supergraph: supergraphFile,
+        port: availablePort,
+        env: {
+          PUBLIC_URL: publicUrl,
+        },
+      });
+      client = createClient({
+        url: `${publicUrl}/graphql`,
+        retryAttempts: 0,
+        fetchFn: fetch,
+        webSocketImpl,
+      });
+      const sub = client.iterate({
+        query: /* GraphQL */ `
+          subscription CountDown {
+            countdown(from: 4)
+          }
+        `,
+      });
 
-it('should subscribe and resolve via http callbacks', async () => {
-  if (process.version.startsWith('v18')) {
-    return;
-  }
-  const supergraphFile = await composeWithApollo([
-    await service('products'),
-    await service('reviews'),
-  ]);
-
-  // Get a random available port
-  const availablePort = await getAvailablePort();
-
-  const publicUrl = `http://localhost:${availablePort}`;
-  await serve({
-    supergraph: supergraphFile,
-    port: availablePort,
-    env: {
-      PUBLIC_URL: publicUrl,
-    },
-  });
-  client = createClient({
-    url: `${publicUrl}/graphql`,
-    retryAttempts: 0,
-    fetchFn: fetch,
-  });
-  const sub = client.iterate({
-    query: /* GraphQL */ `
-      subscription CountDown {
-        countdown(from: 4)
+      const msgs = [];
+      for await (const msg of sub) {
+        expect(msg).toMatchObject({
+          data: expect.any(Object),
+        });
+        msgs.push(msg);
+        if (msgs.length >= 4) {
+          break;
+        }
       }
-    `,
-  });
 
-  const msgs = [];
-  for await (const msg of sub) {
-    expect(msg).toMatchObject({
-      data: expect.any(Object),
+      expect(msgs).toMatchObject([
+        {
+          data: {
+            countdown: 4,
+          },
+        },
+        {
+          data: {
+            countdown: 3,
+          },
+        },
+        {
+          data: {
+            countdown: 2,
+          },
+        },
+        {
+          data: {
+            countdown: 1,
+          },
+        },
+      ]);
     });
-    msgs.push(msg);
-    if (msgs.length >= 4) {
-      break;
-    }
-  }
-
-  expect(msgs).toMatchObject([
-    {
-      data: {
-        countdown: 4,
-      },
-    },
-    {
-      data: {
-        countdown: 3,
-      },
-    },
-    {
-      data: {
-        countdown: 2,
-      },
-    },
-    {
-      data: {
-        countdown: 1,
-      },
-    },
-  ]);
+  });
 });

e2e/federation-subscriptions-passthrough/gateway.config.ts

@@ -6,7 +6,6 @@ import {
 
 export const gatewayConfig = defineConfig({
   webhooks: true,
-  maskedErrors: false,
   transportEntries: {
     products: {
       options: {

e2e/utils/tenv.ts

@@ -833,14 +833,15 @@ export function getAvailablePort(): Promise<number> {
   const server = createServer();
   return new Promise((resolve, reject) => {
     try {
+      server.once('error', reject);
       server.listen(0, () => {
-        try {
-          const addressInfo = server.address() as AddressInfo;
+        const addressInfo = server.address() as AddressInfo;
+        server.close(err => {
+          if (err) {
+            reject(err);
+          }
           resolve(addressInfo.port);
-          server.close();
-        } catch (err) {
-          reject(err);
-        }
+        });
       });
     } catch (err) {
       reject(err);

packages/serve-cli/package.json

@@ -74,7 +74,9 @@
     "@graphql-tools/utils": "^10.5.5",
     "commander": "^12.0.0",
     "dotenv": "^16.3.1",
-    "parse-duration": "^1.1.0"
+    "graphql-ws": "^5.16.0",
+    "parse-duration": "^1.1.0",
+    "ws": "^8.18.0"
   },
   "devDependencies": {
     "@parcel/watcher": "^2.3.0",

packages/serve-cli/src/cli.ts

@@ -286,7 +286,8 @@
       'Apollo API key to use to authenticate with the managed federation up link',
     ).env('APOLLO_KEY'),
   )
+  .option('--disable-websockets', 'Disable WebSockets support');
   .addOption(
     new Option('--jit', 'Enable Just-In-Time compilation of GraphQL documents')
       .env('JIT')
       .argParser(value => {