Rxdart: Subject won't close if combined observable was piped into it

[westracer]

I'm not sure if it's a bug or some misunderstanding. Here's example code to reproduce the problem:

final test1 = BehaviorSubject<int>();
final test2 = BehaviorSubject<int>();
final piped = BehaviorSubject<int>();

Observable.combineLatest2<int, int, int>(test1, test2, (a, b) => a + b).pipe(piped);

Timer(Duration(milliseconds: 500), () {
  piped.close();
});

Exception is thrown when I try to close the subject:

Unhandled Exception: Bad state: You cannot close the subject while items are being added from addStream

I've seen a solution to call drain() method for stream/subject first and then close the it, but drain doesn't return any value, because drain method only returns a value _after_ you close stream.

In some cases replacing Subject with StreamController to pipe observable in helps. Closing StreamController works just fine and doesn't throw exception, though it did before with Subject.
I use Dart 2.4.0 and Flutter 1.7.8+hotfix.4.

[frankpepermans]

Hi there,

pipe is in fact part of the Dart SDK for Streams,
I don't use it that often, best to read up on it (https://api.flutter.dev/flutter/dart-async/Stream/pipe.html)

As soon as you do pipe, then then closing is delegated to the Stream that is being piped, i.e. if that closes, then the pipe target also closes.

[westracer]

Hello, @frankpepermans

Alright, maybe pipe isn't important in this case. My main problem is I don't understand how can I correctly cancel all subscriptions that were created by "combineLatest" method when I want to dispose my bloc, so "combiner" won't trigger on changes of combined streams any further?

[frankpepermans]

If you don't use pipe, but listen instead, you get a StreamSubscription which can be cancelled, if you use pipe, then just make sure all Streams in combineLatest close at some point, which will then auto-close the pipe target.

[westracer]

And what if I want to have multiple listeners of combined observable? If I use .asBroadcastStream(), it will never cancel subscription and combiner won't dispose.

[frankpepermans]

You will just have to cancel each subscription when doing cleanup, asBroadcastStream simply allows multiple listeners, and does not replay the last value (if using a standard StreamController).

Again, this is not rxDart, but general Dart Streams stuff :)

[westracer]

I'm not sure I totally understand how I should cancel subscription created by asBroadcastStream method. Alright, thank you for your time.

[frankpepermans]

You create a Stream when you do asBroadcastStream, you create a Subscription when you then listen on that Stream.

The Subscriptions should be cancelled at some point, the Stream is different, in your case it suffices to just close the 2 Streams that you combine, if you implement the pipe.