Any example of graphql-ws gateway?

[andreisoktoev]

[enisdenjo]

I recommend using graphql-mesh for constructing GraphQL gateways, it has built in support for graphql-ws as the subscription transport.

However, here's a simple proxy implementation that forwards all requests to an upstream graphql-ws server (untested):

import { ExecutionArgs, ExecutionResult } from 'graphql';
import { WebSocket, WebSocketServer, CloseEvent } from 'ws';
import { Client, createClient, SubscribePayload } from 'graphql-ws';
import { useServer } from 'graphql-ws/lib/use/ws';

const server = new WebSocketServer({
  port: 4000,
  path: '/graphql',
});

type GatewayExecutionArgs = ExecutionArgs & {
  contextValue: {
    client: Client;
    payload: SubscribePayload;
  };
};

useServer<{ token: string }, { client: Client; clientSocket: WebSocket }>(
  {
    onConnect: (ctx) =>
      new Promise<void>((resolve, reject) => {
        ctx.extra.client = createClient({
          url: 'ws://upstream.server:4000/graphql',
          webSocketImpl: WebSocket,
          connectionParams: {
            // pass along the authentication token from client
            token: ctx.connectionParams?.token,
          },
          lazy: false, // connect immediatelly
          retryAttempts: 0, // upstream client shouldnt retry
          on: {
            connected: (socket) => {
              ctx.extra.clientSocket = socket as WebSocket;
              resolve();
            },
            error: (err) => {
              console.error('Error while connecting to upstream', err);
              reject(new Error('Upstream connection error'));
            },
            closed: (_event) => {
              const event = _event as CloseEvent;
              console.error('Close connection to upstream', event);
              reject(new Error('Upstream connection error'));

              // rid of client when upstream disconnects
              ctx.extra.clientSocket?.close(event.code, event.reason);
              ctx.extra.client?.dispose();
              ctx.extra.client = undefined;
              ctx.extra.clientSocket = undefined;
            },
          },
        });
      }),
    onClose: (ctx, code, reason) => {
      // rid of upstream connection when client disconnects
      ctx.extra.clientSocket?.close(code, reason);
      ctx.extra.client?.dispose();
      ctx.extra.client = undefined;
      ctx.extra.clientSocket = undefined;
    },
    onSubscribe: (ctx, msg) => {
      const args: GatewayExecutionArgs = {
        schema: null as any, // irrelevant, operation be executed upstream
        document: null as any, // irrelevant, operation be executed upstream
        contextValue: {
          client: ctx.extra.client!, // must exist at this point
          payload: msg.payload,
        },
      };
      return args;
    },
    // perform operations on upstream
    execute: (_args) =>
      // see "Client usage with Promise" recipe
      new Promise((resolve, reject) => {
        const args = _args as GatewayExecutionArgs;
        const client = args.contextValue.client;
        let result: ExecutionResult;
        client.subscribe<any, any>(args.contextValue.payload, {
          next: (data) => (result = data),
          error: reject,
          complete: () => resolve(result),
        });
      }),
    subscribe: (_args): AsyncGenerator<ExecutionResult> => {
      // see "Client usage with AsyncIterator" recipe
      const args = _args as GatewayExecutionArgs;

      let deferred: {
        resolve: (done: boolean) => void;
        reject: (err: unknown) => void;
      } | null = null;
      const pending: ExecutionResult[] = [];
      let throwMe: unknown = null,
        done = false;
      const dispose = args.contextValue.client.subscribe<any, any>(
        args.contextValue.payload,
        {
          next: (data) => {
            pending.push(data);
            deferred?.resolve(false);
          },
          error: (err) => {
            throwMe = err;
            deferred?.reject(throwMe);
          },
          complete: () => {
            done = true;
            deferred?.resolve(true);
          },
        },
      );

      return {
        [Symbol.asyncIterator]() {
          return this;
        },
        async next() {
          if (done) return { done: true, value: undefined };
          if (throwMe) throw throwMe;
          if (pending.length) return { value: pending.shift()! };
          return (await new Promise<boolean>(
            (resolve, reject) => (deferred = { resolve, reject }),
          ))
            ? { done: true, value: undefined }
            : { value: pending.shift()! };
        },
        async throw(err) {
          throw err;
        },
        async return() {
          dispose();
          return { done: true, value: undefined };
        },
      };
    },
  },
  server,
);

console.log('Listening to port 4000');

[andreisoktoev]

it has built in support for graphql-ws as the subscription transport

And what about queries and mutations over WebSocket?

[enisdenjo]

It is possible, just connect to the Mesh server using graphql-ws on the client and all operations will work.

However, using WebSockets only for an upstream server is currently not possible. See ardatan/graphql-mesh#4269.

[andreisoktoev]

I've tried to implement fullWsTransport in Mercurius Gateway. Maybe you can take a look at my example and get some ideas?
mercurius-js/mercurius#834