Handling client side unsubscribe

[boivie]

When implementing the server side of GraphQL subscriptions, there is a callback called subscribe where you often delegate to withFilter. This is called when the client subscribes and you provide a filterFn that will be called on every emitted item.

When the clients calls unsubscribe, it will send GQL_STOP and this should delete the subscription. How should the server implementor handle this? There doesn't seem to be a unsubscribe callback. What actually happens when the client sends GQL_STOP is that the filterFn is called with the payload set to undefined (but this doesn't seem to be documented?)

To prevent memory leaks, cleanup is necessary.

[namnm]

Same question. There's also a memory leak report in #97 from almost a month ago, but there's no signal from the team...

[ItsEcholot]

As @namnm mentioned I suspect that exactly this is the issue which causes the memory leak I found. Would be happy if someone from the team could speak his mind on this. @helfer @Urigo

[NeoPhi]

I maybe misunderstanding the issue. The client's unsubscribe call includes the id that was returned by the original subscribe. The server can use this to cleanup any resources associated with the subscription.

[boivie]

@NeoPhi Well, possibly - if there was a callback at the server side which would be called when a client would unsubscribe which would provide the id. Is there one?

[NeoPhi]

Yes, the server implementation defines this hook:
https://github.com/apollographql/graphql-subscriptions/blob/master/src/pubsub-engine.ts#L4

[Torsten85]

I don't get it how to implement this. Is there an example?
@NeoPhi you linked the unsubscribe method of pubsub ... how should someone use this? As I see it this is not a hook.

In my case there are some watchers that need to be setup based on the subscription parameters. After the last client unsubscribes from a subscription I like to deconstruct the watcher to save some server resources.

[Karabur]

The connection status handled in subscription server code, not the grapqhl-subscriptions library.
But, if you use subscriptions-transport-ws to create a subscriptions server, it will call return method on AsyncIterable returned as subscribe field of resolver if connection closed or client unsibscribed. So all you need is to provide AsyncIterable with custom return method, in which you can cleanup your subscription.

grapqh-subscriptions does not provide a way to configure return property, but it is easy to override. For example:

export function withCancel<T>(asyncIterator: AsyncIterator<T | undefined>, onCancel: Function): AsyncIterator<T | undefined> {
  return {
    ...asyncIterator,
    return() {
      onCancel()
      return asyncIterator.return ? asyncIterator.return() : Promise.resolve({ value: undefined, done: true })
    }
  }
}

.... later in your resolvers you can do:

  Subscription: {
    mySubscription: {
      subscribe: () => withCancel(pubsub.asyncIterator(`Topic`), () => {
        console.log(`Subscription closed, do your cleanup`)
      })
    }
  }

[grantwwu]

This seems to be mostly resolved. If you want to learn more about the nitty-gritty of cleanup, #143 is a good place to start.

I'm going to close this issue for now since it's been a while there was last any activity on it.. I'm going to leave conversation open, so feel free to respond if you want to pursue this further.

[ruohki]

the onCancel is called twice is that intended?

[grantwwu]

When/where are you observing that behavior?

[zhaoyao91]

what an important missing feature!

[grantwwu]

PRs to make it easier to override the return method of the AsyncIterator are welcome

[leszekhanusz]

Using Karabur code worked for me using local graphql-subscriptions but started to fail once I started to use graphql-redis-subscriptions:

"error": {
  "message": "Subscription field must return Async Iterable. Received: { pubsub: { triggerTransform: [function], reviver: undefined, redisPublisher: [Redis], redisSubscriber: [Redis], subscriptionMap: [Object], subsRefsMap: {}, currentSubscriptionId: 1 }, options: undefined, pullQueue: [], pushQueue: [], listening: true, eventsArray: [\"device_added\"], allSubscribed: {}, return: [function return] }"
}

To make it work I had to change the withCancel function to this:

function withCancel(asyncIterator, onCancel) {                                                                        
                                                                                                                    
  let saved_return = asyncIterator.return                                                                               
                                                                                                                    
  asyncIterator.return = () => {                                                                                        
    onCancel()                                                                                                          
    return saved_return ? saved_return.call(asyncIterator) : Promise.resolve({ value: undefined, done: true });         
  }                                                                                                                     
                                                                                                                    
  return asyncIterator                                                                                                  
}

[derbenoo]

@leszekhanusz ran into the same issue with graphql-redis-subscriptions. Your solution still throws an error for me, I had to modify it in the following way to get it to work:

export function withCancel<T>(
  asyncIterator: AsyncIterator<T | undefined>,
  onCancel: () => void
): AsyncIterator<T | undefined> {
  if (!asyncIterator.return) {
    asyncIterator.return = () => Promise.resolve({ value: undefined, done: true });
  }

  const savedReturn = asyncIterator.return.bind(asyncIterator);
  asyncIterator.return = () => {
    onCancel();
    return savedReturn();
  };

  return asyncIterator;
}

For some reason the this context of the asyncIterator needs to be preserved so I added:
const savedReturn = asyncIterator.return.bind(asyncIterator);
( The rest of the implementation should be equivalent)

[iambumblehead]

imo this ticket should not be closed,

But, if you use subscriptions-transport-ws to create a subscriptions server, it will call

subscriptions-transport-ws is long un-supported and, regardless, apollo should provide its own solution and not rely on un-documented work-arounds requiring a specific plugin. Further, the proposed solution is complex and fragile looking. A normal un-subscribe interface should be provided, that does not require importing and composing various things. The PRs-are-welcome attitude seems basically inattentive.

[RichardWright]

@iambumblehead Totally agree. Current situation is a total mess