Automatic unsubscribing in OperatorMerge

[pyrtsa]

It seems to me Observable.merge(...) doesn't unsubscribe its inputs one by one as they complete, but only when unsubscribed manually or when all of the inputs have completed. It seems counter-intuitive. Is that intentional?

[benjchristensen]

Generally the cleanup unsubscribe does not occur until the final subscribe, not for intermediate operators. There is precedent for eagerly unsubscribing, such as in take so as to not delay unsubscribing until some point in the future, and merge is a gray area for being an intermediate operator as it is the terminal state for the Observables it is merging.

We can look at doing so for merge assuming there is not some unconsidered edge-case. If we do merge we should look at the other combinatorial operators as well including zip and concat.

Curious, what is the use case where the Observable emits onCompleted and at that point is not cleaned up and needs to wait until unsubscribe?

[pyrtsa]

I'm playing with long-running (hours, not weeks) streams… not sure if RxJava is ready for that, though! I have a long-running input stream Observable<Observable<A>> xs of short-running input streams that I eventually want to flatten to Observable<A>. However small their footprint may be, I'm concerned with the idea that when Observable.merge(xs) is unsubscribed, I can see as many calls to unsubscribe of the nested Observable<A>s as there were elements in xs.

Of course I can try to make sure that I free up everything I can in their onCompleted method, but something is worse than nothing.

My understanding is, as soon as an Observable completes, it's of no use and thus should be unsubscribed by its subscriber. I can't see an edge-case where it would be otherwise.

[pyrtsa]

FWIW, zip, concat and mergeMap/flatMap should definitely follow the same guideline: as soon as an input Observable completes its unsubscribe is called. For a toy example: xs.map(f) should have the same effect as xs.flatMap(x => observableWithJust(x)) but with the current behavior, the latter has an accumulating memory footprint whereas the former doesn't.

[benjchristensen]

Good points. If we're retaining subscriptions that's a bad thing. I'll take a look and resolve this.

long-running (hours, not weeks) streams… not sure if RxJava is ready for that

This has not been the area where we (Netflix) have used it in production thus far so it is definitely not battle-tested and I'm not surprised there are bugs. We are starting to and have found bugs and expect to find more as it gets the same level of use and testing for long-running streams.

Thank you for the bug report.

[pyrtsa]

Thanks. I'll try to write a reproducing test when the hurry settles down a bit.

[benjchristensen]

Can you take a look at this: #904

I have only spent a few minutes on this so not 100% certain it is done.

[pyrtsa]

Both the implementation and the tests look alright on a first glance. I'll try it out in a minute.

Thanks a lot for such a quick reaction!

[benjchristensen]

This is released in RC4

[headinthebox]

Nice!