Possible bug in subscription?

[quant-daddy]

Hi @gmac Thanks for creating this repo, it's so helpful!

I was wondering if the way you are creating the remote subscriber here is the right way. Wouldn't you want to create a separate instance of the client for each subscription request? Currently, the same client is shared for every request here. In this case how can we use different connection params for different users?

connectionParams: {
  userId: context?.userId,
},

[gmac]

Hi @skk2142, thanks for the kind words and a very good question. If I understand you correctly, you’re asking if the client on line 8 should be moved down to line 10 so that it falls inside of the instanced subscriber handler? Hmm. Maybe? I’ve actually never implemented subscriptions in a production app. Subscriptions were something of a lost art from the Apollo days that no one in The Guild was directly familiar with when they took over; and the old connection preference was transitioning over to the new graphql-ws protocol. I hewed all the new documentation from trial and error while putting together this demo. Additional docs are available here and here about setting up subscriptions, both of which I know have gotten recent updates from the author of graphql-ws. Tinker a bit and report back on your findings. I don’t think you want lots of client instances, but I don’t know for sure. If you find differently, we can update the example.

[gmac]

Sorry for closing! I fat-fingered the wrong button.

[gmac]

Reading your question again, it looks like you’re doing something very specific with passing userID in the connection. Instancing based on request or pooling clients by userID in a weak map may very well be appropriate for what you’re looking to do. That said, I’m not convinced that the docs or this example are wrong — they are optimized for the simplest use case, and expect developers like yourself to customize to match your own implementation requirements. Thoughts?

[quant-daddy]

So I was thinking along the similar lines. Looking at this line in the source code for creating a client, it seems like a websocket (Connected object in the code) is created for a client when subscribe is called on the client. This websocket is used for subsequent interactions. I create a cilent using the function below.

const clients: {
  [key: string]: { [userId: string]: Client };
} = {};

const getClient = (url: string, userId: string | null) => {
  let client: Client | null = null;
  if (!userId || !clients[url] || !clients[url][userId]) {
    client = createClient({
      url,
      webSocketImpl: ws,
      connectionParams: {
        userId: userId,
      },
      on: {
        closed: () => {
          console.log("closed url ", url);
        },
        connected: () => {
          console.log("connected url ", url);
        },
      },
    });
    // if a client is create for a user, the cache it
    if (userId) {
      clients[url] = {
        ...clients[url],
        [userId]: client,
      };
    }
  }
  // return the cached version for a user
  if (userId) {
    return clients[url][userId];
  }
  // for non users, return the generated client
  return client as Client;
};

Every time a user connect to the gateway (which uses schema stitching on apollo federation schema and a few subscription schemas), it creates a persistent websocket connection between the user browser and the gateway. For every subscription request, the schema stitching creates a websocket connection with the appropriate upstream graphql subscription service. The code above creates a separate socket connection for each user and each upstream service that the user subscribes to. So let's say we have two graphql subscription services. For N users connecting to the gateway, there will be N downstream connections at the gateway and 2N upstream connections from the gateway. I also added the possibility of using the same websocket connection for executing realtime queries and mutations. This is how I create a subscriber for schema stitching (using mostly your code)

// see https://github.com/enisdenjo/graphql-ws#async-iterator
export function makeRemoteSubscriber(url: string) {
  return async ({
    document,
    variables,
    context,
  }: {
    document: string | ASTNode;
    variables?: {};
    context?: ApolloContext;
  }) => {
    logger.info(
      `MakeRemoteSubscriber at url ${url}`,
      context?.userId
    );
    const client = getClient(url, context?.userId || null);
    const pending: unknown[] = [];
    let deferred: null | {
      resolve: (value: any) => void;
      reject: (value: any) => void;
    } = null;
    let error: Error | null = null;
    let done = false;

    const query =
      typeof document === "string" ? document : print(document);
    const operation = getGraphqlOperationInfo(query);
    if (operation.operation !== "subscription") {
      const result = await new Promise((resolve, reject) => {
        let result: any;
        client.subscribe(
          {
            query,
            variables,
          },
          {
            next: (data) => (result = data),
            error: reject,
            complete: () => resolve(result),
          }
        );
      });
      return result;
    }
    const dispose = client.subscribe(
      {
        query,
        variables,
      },
      {
        next: (data) => {
          pending.push(data);
          deferred && deferred.resolve(false);
        },
        error: (err: Error) => {
          error = err;
          deferred && deferred.reject(error);
        },
        complete: () => {
          done = true;
          deferred && deferred.resolve(true);
        },
      }
    );

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

I still am not sure if creating a separate websocket connection for each user is the right thing to do: can we multiplex on the same socket connection? Also, the socket connection to each upstream is closed once the downstream client closes the connection but may be we need to garbage collect the client object too? Would WeakMap help with that? I have never used it
so not sure :) Would love to hear your thoughts. Also would reference this issue in graphql-ws repo.

[quant-daddy]

Here's a discussion on this topic: enisdenjo/graphql-ws#194 (comment)

[gmac]

I'm going to close this one out. I think the takeaway for me in enisdenjo/graphql-ws#194 is that the handbook's example follows the recommended best pattern, though that's not to say that you can't tailor your own implementation to your specific needs. On this point, it seems best that the handbook makes the most conservative suggestion possible. If you'd like to write a section summarizing your findings in the chapter notes, contributions are most welcome.