River_pod: Possible memory leak on StreamProvider.autoDispose

[fpabl0]

Describe the bug
I am trying to replace StreamBuilders in favor of StreamProvider.autoDispose, however they don't behave equally (don't know if I am misunderstanding StreamProvider, if so, then sorry). Using watch(streamProvider) does not unsubscribe from the stream when the ui is destroyed as a consequence, every time I returned to that screen a new subscription is done with disposing the previous one. That's not the case with the StreamBuilder.

To Reproduce

I have this method in a class that return a stream:

final waypointsRepoProvider = Provider<WaypointsRepository>(
  (ref) => WaypointsFirestoreRepository(),
);

class WaypointsFirestoreRepository implements WaypointsRepository {
  @override
  Stream<List<WaypointModel>> getWaypointsStream() {
    final route = FirebaseFirestore.instance
            .collection("drivers")
            .doc("J5AIHrG0CH3kOpJUpwCq")
            .collection("route");
    return route.snapshots().map<List<WaypointModel>>((snapshot) {
      print("here"); // this message is the key to see if the stream has still listeners when the ui is destroyed
      return snapshot.docs.map((el) {
        return WaypointModel(
          position: el.get("position"),
          address: el.get("address"),
          completed: el.get("completed"),
        );
      }).toList();
    });
  }
}

With StreamBuilder (when user goes to other screen, if data changes "here" is not displayed, because the stream subscription has been closed by the streambuilder):

  Widget build(BuildContext context, ScopedReader watch) {
    final repo = watch(waypointsRepoProvider);
    return StreamBuilder(
      stream: repo.getWaypointsStream(),
      builder: (context, snapshot) => Container(
        child: snapshot.data != null
            ? Text("${snapshot.data[0]?.address}")
            : Text("no data"),
      ),
    );
 }

With StreamProvider (when user goes to other screen, "here" message still triggers when data changes, and you can get a lot of them at once, if you returned to the ui that use the stream many times):

final waypointsStreamProvider = StreamProvider.autoDispose<List<WaypointModel>>((ref) {
  final repo = ref.watch(waypointsRepoProvider);
  return repo.getWaypointsStream();
});

  Widget build(BuildContext context, ScopedReader watch) {
    final waypoints = watch(waypointsStreamProvider);
    return waypoints.maybeWhen(
      data: (w) => Container(child: Text("${w[0].address}")),
      orElse: () => Container(child: Text("no data")),
    );
 }

Expected behavior
I expect that the subscriptions are totally disposed when the ui that use the streamProvider is destroyed.

[rrousselGit]

Could you share fully executable examples with step by steps way to reproduce the problem?

[fpabl0]

Yes, of course, I have created a github repository with an example with the problem:
https://github.com/fpabl0/riverpod_streamprovider_test

[rrousselGit]

Thanks!

I'm able to reproduce the issue. I will investigate

[rrousselGit]

I think that's a bug related to broadcast streams:

void main() {
  final stream = Stream.periodic(Duration(seconds: 5), (n) {
    print('$n');
    return n;
  });

  final sub = stream.asBroadcastStream().listen((value) {
    print('value $value');
  });

  print('cancel');
  sub.cancel();
}

This will print cancel then keep printing an incrementing number (but the "value" print does not appear)

[rrousselGit]

Interesting case. thanks for the report! Learned something new today

TL;DR, the bug was, Riverpod does:

StreamSubscription sub;
Stream publicStream;

initState() {
  publicStream = stream.asBroadcastStream();
  sub = publicStream.listen(...);
}

dispose() {
  sub.cancel();
}

But with this, while the subscriptions are closed, the broadcast stream never closes

The fix appears to be:

StreamSubscription sub;
Stream publicStream;
StreamController controller;

initState() {
  controller = StreamController.broadcast();
  publicStream = controller.stream;
  sub = stream.listen(controller.add, onError: controller.addError, onDone: controller.close);
}

dispose() {
  sub.cancel();
  controller.close();
}

[fpabl0]

void main() {
  final stream = Stream.periodic(Duration(seconds: 5), (n) {
    print('$n');
    return n;
  });

  final sub = stream.asBroadcastStream().listen((value) {
    print('value $value');
  });

  print('cancel');
  sub.cancel();
}

Hmm but in that case the value printed is because of the stream generator, the case in the example is this:

void main() async {
  final stream = Stream.periodic(Duration(seconds: 1), (n) {
    print('$n');
    return n;
  });

  final sub = stream.asBroadcastStream().map((event) {
    print("here");
    return event;
  }).listen((value) {
    print('value $value');
  });

  print('cancel');
  sub.cancel();

  await Future.delayed(Duration(seconds: 5));
}

You will see the numbers printed, but not "here" message, because the subscription was cancelled.

[fpabl0]

Interesting case. thanks for the report! Learned something new today

TL;DR, the bug was, Riverpod does:

StreamSubscription sub;
Stream publicStream;

initState() {
  publicStream = stream.asBroadcastStream();
  sub = publicStream.listen(...);
}

dispose() {
  sub.cancel();
}

But with this, while the subscriptions are closed, the broadcast stream never closes

The fix appears to be:

StreamSubscription sub;
Stream publicStream;
StreamController controller;

initState() {
  controller = StreamController.broadcast();
  publicStream = controller.stream;
  sub = publicStream.listen(controller.add, onError: controller.addError, onDone: controller.close);
}

dispose() {
  sub.cancel();
  controller.close();
}

Thanks to you for this amazing library.
Hmm yes, I didn't know StreamProvider creates internally a broadcast stream, but that should be the solution, although I think the sub.cancel is not needed because controller.close() will dispose all the attached subscriptions?

[rrousselGit]

Yes but that is not what happen here.

The code is not stream.asBroadcast().map() but stream.map().asBroadcast

[fpabl0]

Yes but that is not what happen here.

The code is not stream.asBroadcast().map() but stream.map().asBroadcast

Yep, I see, I didn't realize StreamProvider creates internally a broadcast stream.

[rrousselGit]

Yeah

The broadcast stream is needed for watch(streamProvider.stream).listen(...)

[fpabl0]

Yeah

The broadcast stream is needed for watch(streamProvider.stream).listen(...)

Oh, I see, so that should be the solution, closing the StreamController in the dispose method.

[fpabl0]

Hi @rrousselGit , sorry for comment on this closed issue, I just want to share that I have just realized that the broadcast stream can be closed using asBroadcastStream method too:

void main() async {
  final stream = Stream.periodic(Duration(seconds: 1), (n) {
    print('$n');
    return n;
  });

  final broadcastStream = stream.asBroadcastStream(
    onCancel: (sub) {
      print("don't have listeners now, so close the underlying subscription");
      sub.cancel();
    },
  );

  final sub1 = broadcastStream.listen((value) {
    print('value sub1 -> $value');
  });

  final sub2 = broadcastStream.listen((value) {
    print('value sub2 -> $value');
  });

  print('cancelling sub1');
  sub1.cancel();
  print('cancelling sub2');
  sub2.cancel();

  await Future.delayed(Duration(seconds: 5));
}

[rrousselGit]

Indeed

But in this case it makes more sense to have a StreamController

[fpabl0]

@rrousselGit version 0.12.0 solves this issue?

[rrousselGit]

Yes it does

[fpabl0]

Yes it does

Hmm I have updated the example to that version, however I am still getting the same problem. Am I missing anything?

[rrousselGit]

Ah my bad, I fixed it for StreamProvider but not autoDispose (which had the issue with context.refresh/ref.watch too)

[rrousselGit]

I'll deploy another fix later today, it's a small change

[fpabl0]

I'll deploy another fix later today, it's a small change

Perfect, thank you so much 😄