onSubscriptionData not triggering when extending useSubscription to handle multiple subscription (with different remote urls)

[otsab19]

Issue Description

React hook implementation: https://github.com/apollographql/apollo-client/blob/main/src/react/hooks/useSubscription.ts from official doc.

I tried using the same implementation but with multiple subscription for multiple remote urls at once.

Subscription handler

const setSubscription = () => {
    return clients.map((clientId) => {
        return client.subscribe({
            query: query,
            variables: {
                variables
                clientId: clientId, // extra variable is required, otherwise apollo considers all subscription as same, to make it unique (Possible fix  [#10215 ](https://github.com/apollographql/apollo-client/issues/10202). in future version)
            },
            fetchPolicy: options?.fetchPolicy || 'no-cache',
            context: {
                ...context,
                subscribeId: clientId,
            },
        });
    });
};

Setting the subscription

    if (options?.skip || !systemIds?.length || loading) {
      return null;
    }
    return setSubscription();
  });

Subscribing logic

 useEffect(() => {
    if (!observable) {
      return;
    }

    const subscriptions = observable.filter(Boolean).map((obs, idx) =>
      obs.subscribe({
        next(fetchResult) {
          const res = {
            loading: false,
            // TODO: fetchResult.data can be null but SubscriptionResult.data
            // expects TData | undefined only
            data: fetchResult.data!,
            error: void 0,
            variables: options?.variables,
          };
          setResult(res);

          if (ref.current.options?.onSubscriptionData) {
            ref.current.options.onSubscriptionData({
              client,
              subscriptionData: res,
            });
          }
        },
        error(error) {
          setResult({
            loading: false,
            data: void 0,
            error,
            variables: options?.variables,
          });
          ref.current.options?.onError?.(error);
        },
        complete() {
          if (ref.current.options?.onSubscriptionComplete) {
            ref.current.options.onSubscriptionComplete();
          }
        },
      })
    );

    return () => {
      subscriptions.map((obs) => obs.unsubscribe());
    };
  }, [observable]); 

The remote urls are set using the Apollo Link (distinguished as per the subscribeId sent in context) This case lets say I have 3 clients as:

• wss:client1.sub
• wss:client2.sub
• wss:client3.sub

Now if I subscribe to the same subscription using the multiple subscription hook implemented above:

subscription test{
    test{
      id
     name
    }
  }

All of the 3 clients are subscribing to the subscription query as expected. But the problem is it cannot trigger the onSubscriptionData method when data is being sent from server through websocket.
This might be related to #10693 .

Another issue I encountered here is we need to pass extra variables to be able to initiate same subscription endpoint to multiple remote wss urls
This is related to this: #10215 . Seems like that fix was reverted.

Link to Reproduction

n/a

Reproduction Steps

• Add multiple wss links
• Subscribe to same query over all the remote urls

No response

[phryneas]

Phew. You would have to change a lot more than these parts of useSubscription - and I'm not sure if that's a really good idea, since you'll always be behind on bugfixes we make to the hook.

Is there an option to call useSubscription three times instead?

[otsab19]

@phryneas The problem is the wss clients are dynamic. Those URL endpoints are fetched beforehand and the Apollo client is created. So I need to loop through useSubscription hook which react does not recommend. I also tried that because the length is fixed so the hook rules will not be broken. That results to the same problem as mentioned above.

[phryneas]

And your usecase is essentially that you have a number of wss clients and you want to merge their results together?

In that case, I would suggest writing a custom Link for that instead 🤔

[otsab19]

I created custom link to know which request goes through which wss url. Is that what you meant? I don't need to merge data as of now, just need to get the data from those different remote wss endpoints.

This is the Link:

let allLinks;
  for (let client of clients) {
    allLinks = split(
      ({ getContext }) => {
        return getContext().subscribeId === client.id;
      },
      getLink(client.id),
      allLinks
    );
  }

const getLink = (url) => {
  return split(
    ({ query }) => {
      const { kind, operation } = getMainDefinition(query) as OperationDefinitionNode;
      return kind === 'OperationDefinition' && operation === 'subscription';
    },
    new WebSocketLink({
      uri: url,
      options: {
        timeout: 5000,
        reconnect: true,
      },
    }),
    new HttpLink({
      uri: httpUri,
      credentials: 'include',
      headers: {
      },
    })
  );
};

[phryneas]

I was thinking something along the lines of (untested)

const linksBySubscribeId: { [id: string]: ApolloLink } = {};
const httpLink = new HttpLink({
  uri: httpUri,
  credentials: "include",
  headers: {},
});
export class SelectingLink extends ApolloLink {
  public request(
    operation: Operation,
    forward: NextLink
  ): Observable<FetchResult> {
    const subscribeId = operation.getContext().subscribeId;
    let link = !subscribeId ? httpLink : linksBySubscribeId[subscribeId];
    if (!link) {
      link = linksBySubscribeId[subscribeId] = new WebSocketLink({
        uri: `ws://localhost:4000/subscriptions/${subscribeId}`,
        options: {
          timeout: 5000,
          reconnect: true,
        },
      });
    }
    return link.request(operation, forward);
  }
}

Of course the lazy link instantiation is optional ;)

[otsab19]

@phryneas Thanks for the response.
But still isn't extending the Apollolink (like you showed above): SelectingLink and using split : split approach the same? I did try your approach, but still it does not solve the issue I mentioned in this issue post.

[otsab19]

@lovasoa answer in this issue #10693 worked for me. I am not sure whether it is the ideal approach.

[phryneas]

Ah! My assumption after reading your initial code was that there was a problem where probably only onSubscriptionData for one of the subscriptions would trigger (and present itself to you as "just not triggering".
But if it's a "unsubscribe & restart" thing, then yes, that solution probably works.

[otsab19]

I guess I got you confused a little 😀
In the above implementation for multiple subscriptions hook, only one of the subscriptions from the remote wss triggers the onSubscriptionData. I can see the connection of WebSocket and data incoming from each of the remote websocket in the network tab just fine.
Updating with unsubscribe & restart worked fine.

export function subscribe_forever<T>(create_observable: () => Observable<T>): Observable<T> {
	let observable: Observable<T> = create_observable();
	return new Observable(function resubscribe(subscriber) {
		const initial_observable = observable;
		const subscription = observable.subscribe(
			(value) => subscriber.next(value),
			(error) => subscriber.error(error),
			function onComplete() {
				if (observable === initial_observable) observable = create_observable();
				const new_sub = resubscribe(subscriber);
				subscription.unsubscribe = new_sub.unsubscribe.bind(new_sub);
			}
		);
		return subscription;
	});
}

As a wrapper while setting client.subscribe