Any/All should not unsubscribe downstream.

[akarnokd]

Should fix issue #1935

When in doubt, see take.

[myinsiders]

Thanks, I'll give this a go and verify it fixes it for me.

Others that look like they may be susceptible:

OperatorAll
DebounceWithSelector
OperatorSkipTimed
OperatorSkipUntil

[akarnokd]

Thanks for the other cases. I'll add the fixes for them in this PR.

[akarnokd]

I find DebounceWithSelector, OperatorSkipTimed and OperatorSkipUntil suspicious, but since these don't just return a single value, I don't know what effect it has on backpressure if I break the chain the way Any or All does.

While we are at it, I think Any and All should request(Long.MAX_VALUE) regardless of the downstream because they spin on a single flag and thus can handle any throughput. The current version request 1-by-1 which adds large overhead.

[benjchristensen]

I think Any and All should request(Long.MAX_VALUE) regardless of the downstream because they spin on a single flag and thus can handle any throughput

Agreed.

since these don't just return a single value, I don't know what effect it has on backpressure if I break the chain the way Any or All does

We would have to manually compose through the backpressure if we break the chain. Why though should we need to decouple on those as they shouldn't need to unsubscribe early? If we are we probably shouldn't be. Having them chained is far more efficient so we don't want to break the chain if we don't need to.

[akarnokd]

We would have to manually compose through the backpressure if we break the chain. Why though should we need to decouple on those as they shouldn't need to unsubscribe early? If we are we probably shouldn't be. Having them chained is far more efficient so we don't want to break the chain if we don't need to.

They unsubscribe themselves in onCompleted() and onError() and may prevent the async use of the preceding onNext(). In other words, if I manually schedule a task for each onNext event and add the cancellation subscription to the downstream subscriber, the onCompleted() in these operators will cancel that task. (ObserveOn was fixed a couple of times to make this not a problem.)

[benjchristensen]

But why do they need to unsubscribe in onComplete and onError? Those are terminal events, we shouldn't need to unsubscribe as that will happen automatically after the terminal event is emitted.

[akarnokd]

Because they need to terminate a companion observable or timer when the main terminates.

[benjchristensen]

I know they need to cleanup, but why does it need to happen eagerly right then? The contract is that after it emits onError or onComplete it will eventually be unsubscribed after all downstream work is done. So why do we need to call unsubscribe eagerly? If I remove those unsubscribe calls no tests break, and unsubscribe will still end up being called.

For all three it seems there is no need to be eager as these do not terminate their origin early. Cleanup of their companion Observable can happen eager if they want (such as SkipUntil) but that is not part of the chain nor does it require unsubscribing the entire operator.

[benjchristensen]

Merging as this PR is fine, the discussion is about other things.