socket.close fired but not 'onClose' callback from useServer

[c4c1us]

Hi,

I was trying to fire the onClose callback from useServer by closing my tab on Chrome. The function is never fired, but socket.on("close",..) is.

Is this a bug or a normal behaviour ?

useServer(
    {
      onConnect: ctx => {

        ctx.extra.socket.on("close", o => {
          console.log(`REALLY CLOSED`);
        });

      return true;
     },
     onClose: () => {
        console.log("CLOSED BY SERVER");
      },
   }

[enisdenjo]

Hmm, the onClose should most definitely be called. I listen for the close the same way you listen for it in the onConnect example:

graphql-ws/src/use/ws.ts

Lines 107 to 118 in c0b171e

	socket.once('close', (code, reason) => {
	if (pongWait) clearTimeout(pongWait);
	if (pingInterval) clearInterval(pingInterval);
	if (!isProd && code === 1002)
	console.warn(
	`WebSocket protocol error occured. It was most likely caused due to an ` +
	`unsupported subprotocol "${socket.protocol}" requested by the client. ` +
	`graphql-ws implements exclusively the "${GRAPHQL_TRANSPORT_WS_PROTOCOL}" subprotocol, ` +
	'please make sure that the client implements it too.',
	);
	closed(code, reason);
	});

and then the closed call does:

graphql-ws/src/server.ts

Lines 731 to 740 in c0b171e

	// wait for close, cleanup and the disconnect callback
	return async (code, reason) => {
	if (connectionInitWait) clearTimeout(connectionInitWait);
	for (const sub of Object.values(ctx.subscriptions)) {
	// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
	await sub?.return!(); // iterator must implement the return method
	}
	if (ctx.acknowledged) await onDisconnect?.(ctx, code, reason);
	await onClose?.(ctx, code, reason);
	};

Could you share a simple repro? Maybe some of the cleanup actions stall.

[c4c1us]

Hum,

I think it's related to my async iterator on my subscription (I will share a repo if this still not help).

This code calls onClose:

useServer(
    {
     ...
      schema,
      roots: {
        subscription: {
          async *allChannelEvent() {
            logger.info("allChannelEvent: ");

            try {
              for (const hi of ["Hi", "Bonjour", "Hola", "Ciao", "Zdravo"]) {
                yield {
                  allChannelEvent: {
                    channelId: "$ID",
                    type: "IS_TYPING",
                    userId: "$ID",
                    __typename: "ChannelEventResult"
                  }
                };
              }
            } catch (e) {
              console.log(`Error allChannelEvent: ${e}`);
            } finally {
              console.log("Finally allChannelEvent");
            }
          }
        }
     ...
  }
)

This code not fire 'onClose'

useServer(
    {
     ...
      schema,
      roots: {
        subscription: {
          async *allChannelEvent() {
            logger.info("allChannelEvent: ");

            try {
              const iterator = pubsub.asyncIterator("$ID");

              let next;

              while ((next = await iterator.next()).done === false) {
                yield {
                  allChannelEvent: next.value
                };
              }
            } catch (e) {
              console.log(`Error allChannelEvent: ${e}`);
            } finally {
              console.log("Finally allChannelEvent");
            }
          }
        }
     ...
  }
)

The pubsub code is similar to this repository.

[n1ru4l]

What version of graphql.js and Node.js are you using?

[enisdenjo]

Can you check if the while loop breaks when returning the iterable (you can just add a log statement below it)? My guess is that it gets stuck.

Also, why use the while loop instead of a for async of loop? Like this:

const iterator = pubsub.asyncIterator('$ID');

for await (const value of iterator) {
  yield {
    allChannelEvent: value,
  };
}

[enisdenjo]

Any news @c4c1us?

[c4c1us]

Hi,

I'm testing your code above.

[c4c1us]

Hi again,

@n1ru4l Node v14.15.1 / GraphQL 14.5.8

@enisdenjo
Still not working with the for loop, kind of stuck like you said. I faked an async iterator with a setTimeoutand a got the same issue, the WebSocket won't close.

  function sleep(ms: number) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }

  async function getAsyncData() {
    await sleep(15000); // simulate database/network delay...
    return [1, 2, 3, 4, 5]; // ...then return some data
  }

  const asyncIterable = {
    async *[Symbol.asyncIterator]() {
      yield* await getAsyncData();
    }
  };

useServer(
    {
     ...
      schema,
      roots: {
        subscription: {
          async *allChannelEvent() {
            logger.info("allChannelEvent: ");

            try {
              for await (const item of asyncIterable) {
                console.log(item)
                yield {
                  ...
                };
              }
            } catch (e) {
              console.log(`Error allChannelEvent: ${e}`);
            } finally {
              console.log("Finally allChannelEvent");
            }
          }
        }
     ...
  }
)

[n1ru4l]

might be related to this graphql/graphql-js#2983 (Note: i dis not verify, just saw the issue and it seems this might be related)

[enisdenjo]

@c4c1us well, the for loop must break after returning the iterator for the teardown to work... As you mentioned in #152 (comment), if the loop breaks, onClose gets called; therefore - this issue is not related to graphql-ws.

[c4c1us]

@n1ru4l I downgrade to 14.10.1, It didn't fix the issue.

@enisdenjo I change my subscription and put this :

...
subscription: {
allChannelEvent: () => ({
            async return() {
              console.log("RETURN");
              await sleep(1000);
            },
            async throw() {
              console.log("THROW");
              await sleep(1000);
            },
            async *[Symbol.asyncIterator]() {
              const iterator = pubsub.asyncIterator(
                "$ID"
              );

              for await (const value of iterator) {
                yield {
                  allChannelEvent: value
                };
              }
            }
          })          
}
...

I see in the closed function that the library call the return function of each subscriptions.

Hmm, the onClose should most definitely be called. I listen for the close the same way you listen for it in the onConnect example:

graphql-ws/src/use/ws.ts

Lines 107 to 118 in c0b171e

	socket.once('close', (code, reason) => {
	if (pongWait) clearTimeout(pongWait);
	if (pingInterval) clearInterval(pingInterval);
	if (!isProd && code === 1002)
	console.warn(
	`WebSocket protocol error occured. It was most likely caused due to an ` +
	`unsupported subprotocol "${socket.protocol}" requested by the client. ` +
	`graphql-ws implements exclusively the "${GRAPHQL_TRANSPORT_WS_PROTOCOL}" subprotocol, ` +
	'please make sure that the client implements it too.',
	);
	closed(code, reason);
	});

and then the closed call does:

graphql-ws/src/server.ts

Lines 731 to 740 in c0b171e

	// wait for close, cleanup and the disconnect callback
	return async (code, reason) => {
	if (connectionInitWait) clearTimeout(connectionInitWait);
	for (const sub of Object.values(ctx.subscriptions)) {
	// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
	await sub?.return!(); // iterator must implement the return method
	}
	if (ctx.acknowledged) await onDisconnect?.(ctx, code, reason);
	await onClose?.(ctx, code, reason);
	};

Could you share a simple repro? Maybe some of the cleanup actions stall.

So in my code, the return function should be called and display RETURN in the console regardless the if the loop breaks, right ?

I tried this too (without the function returning the value):

...
subscription: {
allChannelEvent: {
            async return() {
              console.log("RETURN");
              await sleep(1000);
            },
            async throw() {
              console.log("THROW");
              await sleep(1000);
            },
            async *[Symbol.asyncIterator]() {
              const iterator = pubsub.asyncIterator(
                "$ID"
              );

              for await (const value of iterator) {
                yield {
                  allChannelEvent: value
                };
              }
            }        
}
...

I'm new to iterator & generator but I can't see what I'm doing wrong here unfortunatly.

[enisdenjo]

Just calling return does not break the loop, you must indicate that the loop is done in the next method too. The following example shows how to convert a graphql-ws subscription to an async iterator, this should help you understand how the iterator works:

*Copied from AsyncIterator recipe

import { createClient, SubscribePayload } from 'graphql-ws';

const client = createClient({
  url: 'wss://iterators.ftw/graphql',
});

function subscribe<T>(payload: SubscribePayload): AsyncIterableIterator<T> {
  let deferred: {
    resolve: (done: boolean) => void;
    reject: (err: unknown) => void;
  } | null = null;
  const pending: T[] = [];
  let throwMe: unknown = null,
    done = false;
  const dispose = client.subscribe<T>(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 return() {
      dispose();
      return { done: true, value: undefined };
    },
  };
}

(async () => {
  const subscription = subscribe({
    query: 'subscription { greetings }',
  });
  // subscription.return() to dispose

  for await (const result of subscription) {
    // next = result = { data: { greetings: 5x } }
  }
  // complete
})();

[c4c1us]

I understand that calling the return does not break the loop, but in my example, the return function wasn't called ("RETURN" wasn't display in the console).

[c4c1us]

I gonna search again, thanks for help btw.

[enisdenjo]

I suggest reading on async iterators first. I find this article quite good: https://javascript.info/async-iterators-generators.

In essence the pubsub.asyncIterator('$ID') should already give you a ready to use working async iterator. It might be a problem with graphql-subscriptions too. Am not sure what is happening here; but, it seems like this part simply gets stuck:

graphql-ws/src/server.ts

Line 736 in 77a2981

await sub?.return!(); // iterator must implement the return method

[c4c1us]

I found my mistake !

As you said I should use the async iterator of the pubsub. I previously use Apollo Server to handle my subscriptions. Then I move to your package and I start to iterate my async iterator ever since because It was not working when I returned it.

Why ? Because with Apollo Server you have to "yield" only the value and with graphql-ws { subscriptionName: value }. So tried to correct an issue and I create another one. Anyway, I learn a lot about iterator / generator and graphql-ws.

Thanks for your help and patience ! 🙏
Have a nice day

[enisdenjo]

Glad you got it working! 🎉