Subscription.messages leaks messages

[are]

Description

Subscription.messages stream is not closed on Subscription.unsubscribe and leads to leaking messages.

Steps to reproduce:

1. Subscribe to some channel a.
2. Add a listener to subscription.messages stream.
3. Unsubscribe using subscription.unsubscribe().
4. Create new subscription to the same channel a.

[client-engineering-bot]

@are this issue is addressed in v1.0.5

[aadil058]

@are Seems like issue is not yet fixed It is still leaking messages. Push notifications are fixed though in latest release.

[aadil058]

I have properly unsubscribed/cancelled subscriptions in dispose() method but I can still see duplicate subscriptions when I return back to "chat screen".

I confirmed the same using below code in initState() method -

    chatService.pubnub.keysets.forEach((key, value) {
      value.settings.forEach((k, v) {
        if(k == '#subscriptions') {
          print(v.runtimeType);
          Set<Subscription> set = v as Set;
          set.forEach((Subscription subscription) {
            print(subscription.channels);
          });
        }
      });
    });

[are]

Could you post the code where you subscribe? I cannot replicate this issue anymore.

[aadil058]

Code inside initState() =>

      chatSubscription = chatService.subscribeToChannel('patientDoctorId');
      chatSubscription.messages.listen((message) async {
        if(mounted) {
          logger.logger.d('MESSAGE ARRIVED ${message.payload}');
          ChatMessage msg = (message.payload.runtimeType == String) ? ChatMessage.fromJson(jsonDecode(message.payload)) : ChatMessage.fromJson(message.payload);
            Map<String, dynamic> entry = { 
              'channel': 'patientDoctorId', 
              'message': msg.message, 
              'publisher': msg.publisher, 
              'timestamp': msg.timestamp, 
            };
            await chatService.insertMessageEntry(entry);
            setState(() => messages.insert(0, msg));
        }
      }

code inside dispose() =>

chatSubscription?.unsubscribe();

chatService's subscribeToChannel() method =>

  Subscription subscribeToChannel(String channelName) {
    return pubnub.channel(channelName, using: 'wellowise').subscribe();
  }

Pubnub initialization =>

  Future<void> init(String unique) async {
    pubnub = PubNub(); 
    pubnub.keysets.add(Keyset(subscribeKey: Config.PUBNUB_SUBSCRIBE_KEY, publishKey: Config.PUBNUB_PUBLISH_KEY, uuid: UUID(unique)), name: 'wellowise');
    await initDatabase();
  }

[are]

So this is actually not the same issue. The code is behaving as intended: every time you call the pubnub.channel(...).subscribe() method, a new subscription is created. Those old subscriptions are unused tho, so they won't receive new messages. I should probably make the keyset.settings['#subscriptions'] an Expando so they can be garbage collected. I will take care of this in the next release.

[aadil058]

Those old subscriptions are unused tho, so they won't receive new messages.

Actaully they are receiving new messages. If I revisit "chat screen" 5 times then I can see 5 subscriptions and hence I am receiving new messages 5 times. First 4 callbacks were calling setState() on the state object which had been disposed, that's why I am checking if state is mounted or not first.

If I called unsubscribe on subscription when I pressed back button then It should not receive messages when I come back to the screen.

[are]

I am pushing a fix that may fix this, but in the meantime - can you make sure that the dispose (and therefore unsubscribe) is actually called? Flutter doesn't always dispose of Widgets when you use Navigator.push.

Also, I don't recommend creating a subscription every time you leave a view - it would be a better idea to have one created permamently and then calling unsubscribe and subscribe on this one subscription.

[aadil058]

I am sure dispose is called because the route is being popped from the navigator stack and I also confirmed it with a print statement.

I think I can use global listener but for now if(mounted) {} workaround is working good.

[are]

I've released v1.1.0 - but you have to keep few things in mind - pubnub.subscribe now returns Future<Subscription>. But you should be able to create a subscription as a static variable using pubnub.subscription, and then in initState you will be able to call subscription.subscribe() and in dispose subscription.unsubscribe(). Let me know if there are still issues with it.

[aadil058]

Sure thing. I will let you know if it worked out for me when I try it.