Fixed SubscribeUntil not unsubscribing from its sources properly.

[akarnokd]

The operator had several bugs:

• If either the main source or the other source completed, there was no proper unsubscription happening to either of them*.
• When subscribing to the other source, it could overwrite the producer of the main source.
• One should not unsubscribe the downstream sequence, but there were tests anticipating this so I left it so with a warning; maybe there is a legitimate case for this in Android I don't know about.

* probably

[dlew]

This was a long time coming. The operator was based on an old version of takeUntil() and I'd seen that it had been changed since then.

I'm busy with the IO conference this week but I'll have time to review in-depth soon.

[dlew]

I finally got a chance to look over it - looks great to me. 👍

One should not unsubscribe the downstream sequence

Regarding that - the issue that commonly comes up is thus:

1. An Activity starts up (not sure how familiar you are with Android, but this is basically a new screen).
2. The Activity's code sets up a few subscriptions that should last until the Activity is dismissed.
3. The Activity runs as normal.
4. The Activity is dismissed (say, by the user pressing the back button).
5. Unsubscribe from subscriptions to avoid memory leaks.

The entire point of this exercise is to make step 5 easier. You could, of course, manually handle all your Subscriptions, but that has proven to be a big pain, especially because (for maximum correctness) there are different points in the Activity lifecycle where you might want to unsubscribe various sequences.

We could've just used takeUntil but we didn't want onComplete called in the case that the Activity is just going away - we wanted it to act in the same way as if you had manually unsubscribed.

If you have a better suggestion for handling this problem, I'm all ears.

[JakeWharton]

@dlew you want it merged as is or do you want continue the discussion around downstream subscription?

[dlew]

@JakeWharton This is a definitive improvement on what we currently have, but let's continue the discussion for now - we can always merge & iterate later if need be.

[akarnokd]

There is an option you are not going to like and breaks the established protocol: on termination, emit an ActivityDestroyedException. It will clean up the chain and won't trigger any normal completion paths, but clients have to swallow/ignore this specific exception to avoid log cluttering or error dialogs from then on.

Same thing happens in InputStream.read() when an async InputStream.close() is issued.

[dlew]

You're right, I'm not a fan of that option. The goal is to reduce repetitive work; having to handle this special onError everywhere (especially when 99.99% of the time you just want to ignore it) runs counter to that goal.

What is the danger in unsubscribing the downstream subscriber if it's a known side effect of this custom operator?

[akarnokd]

Many RxJava operators stop the propagation of upstream unsubscription which may leave other operators hanging further down the stream. So unless this subscribeUntil is the very last in the chain, I'd expect leaks.

However, asynchronously unsubscribing the terminal Subscriber is allowed and cleans up properly, but then you need to hijack every subscribe() call in some way.

[dlew]

It seems like the options are:

1. Use takeUntil instead (thus having to deal with onComplete arriving due to Activity destruction)
2. Contractually obligate people to use this method last in their chain
3. Emit an Exception on completion, client must handle
4. Hijack all subscribe() calls so we can guaranteedly handle the terminal Subscriber

I'm beginning to think that the first option makes the most sense, though it would change the way clients interact with sequences, since they would never know why the sequence completed.

[akarnokd]

I'd give the clients the choice through an enum parameter: COMPLETE, THROW, UNSUBSCRIBE and educate them about these options. Option 4 might not work all the time because there is only a single RxJavaObservableExecutionHook attachable at a given time.

[dlew]

I tend to shy away from options because a good default is often easier. A default would be alright in this case since (if you need a different behavior) you could always handle the Subscriber manually.

@JakeWharton What do you think about the enum?

[JakeWharton]

I'm usually not a fan of options because from what we've seen users rarely take the time to understand the implications of the choice and choose improperly. If we have no other choice then that's fine, but ideally we could just have a default–even if it means that it comes with caveats.

[dlew]

Alright, my vote is to two things:

• Merge this PR for now (since it maintains current behavior, even if it's potentially buggy).
• Create a discussion issue about the best default to pick (or if we should even pick one).

[JakeWharton]

I charge you with the holy quest of filing the issue for further discussion.