1.x: fix timeout (timed, selector) unsubscribe bug

[akarnokd]

This PR fixes an unsubscribe bug somewhere in the timed timeout operator reported in #5657 by implementing it in a algorithmically fresh manner.

[codecov]

Codecov Report

Merging #5660 into 1.x will decrease coverage by 0.14%.
The diff coverage is 88.7%.

@@             Coverage Diff             @@
##                1.x   #5660      +/-   ##
===========================================
- Coverage     84.34%   84.2%   -0.15%     
- Complexity     2888    2889       +1     
===========================================
  Files           291     290       -1     
  Lines         18199   18256      +57     
  Branches       2480    2495      +15     
===========================================
+ Hits          15350   15372      +22     
- Misses         1983    2001      +18     
- Partials        866     883      +17

Impacted Files	Coverage Δ	Complexity Δ
src/main/java/rx/Observable.java	99.64% <100%> (ø)	449 <1> (+1)	⬆️
...rators/OnSubscribeTimeoutSelectorWithFallback.java	86.17% <86.17%> (ø)	2 <2> (?)
...operators/OnSubscribeTimeoutTimedWithFallback.java	91.25% <91.25%> (ø)	2 <2> (?)
...rx/internal/util/atomic/MpscLinkedAtomicQueue.java	74.19% <0%> (-12.91%)	7% <0%> (-1%)
.../java/rx/internal/util/unsafe/MpscLinkedQueue.java	75.75% <0%> (-12.13%)	9% <0%> (-1%)
...java/rx/internal/schedulers/ExecutorScheduler.java	77.46% <0%> (-7.05%)	2% <0%> (ø)
...ternal/operators/OperatorOnBackpressureLatest.java	78.84% <0%> (-3.85%)	3% <0%> (ø)
src/main/java/rx/observers/SerializedObserver.java	97.82% <0%> (-2.18%)	19% <0%> (ø)
...x/internal/operators/DeferredScalarSubscriber.java	98.27% <0%> (-1.73%)	24% <0%> (-1%)
...in/java/rx/internal/operators/OperatorPublish.java	77.91% <0%> (-0.84%)	8% <0%> (ø)
... and 12 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 81542cd...ab16779. Read the comment docs.

[vanniktech]

Is OperatorTimeoutBase still used somewhere?

[akarnokd]

In OperatorTimeoutWithSelector.

[sfitts]

I probably should have mentioned in my original bug, but this variant of timeout (which uses OperatorTimeoutWithSelector) also exhibits the same problem:

public final <U, V> Observable<T> timeout(Func0<? extends Observable<U>> firstTimeoutSelector, Func1<? super T, ? extends Observable<V>> timeoutSelector, Observable<? extends T> other)

Can this fix be adapted to cover that variant as well (which we also use)?

[akarnokd]

Yes, they were built on the same abstraction. It wasn't part of the initial report and I try to minimize my interaction with 1.x.

[sfitts]

Understood. As I mention above, the new operator actually allows us to solve both of our use cases, so I'm fine with leaving the other operator alone.

[sfitts]

Nice -- did not know about TestScheduler, very handy.

[sfitts]

If I'm reading this correctly, I think there may be a slight behavior difference between this implementation and the existing one. The existing implementation will timeout (and thus switch) if there are 0 calls to onNext before the timeout. Here I think there needs to be at least 1 call before the timer is set. I'll confirm this and send a test case when I get a chance.

Turns out we actually want both semantics. We're using the other timeout variant mentioned below to get the "have to have at least one message" semantic by having firstTimeoutSelector return Observable.never(). If there were a way to get both behaviors with just this operator, that would great.

[sfitts]

Never mind -- I see the startTimeout call in call above.

[sfitts]

With this operator I can override call to make that optional and get the 2 semantics I'm interested in, so this will be perfect. Thanks.

[sfitts]

Except that it's final, doh!. I'm assuming that's a standard policy for operators. In that case would it be possible to add a boolean parameter to the constructor where true means start the timer on subscribe and false means don't? Understood if that seems too specialized, but it would save us having to use the more complex operator.

[akarnokd]

Sorry, no more new features for 1.x, only bugfixes.

[sfitts]

Understood.

[akarnokd]

Posted fix for the selector version as well.

[sfitts]

Thanks -- we're using the new operators in our code base and they are working as expected.

[akarnokd]

Great!