2.x: improve Flowable.timeout()

[akarnokd]

This PR improves the internal overhead of the Flowable.timeout operator (its 2 timed and 2 selector-based versions) to use the adapted indexed atomic state transition approach.

In addition, there was a race condition and potential event loss in TestScheduler:

• peek() could return null if the task was removed from the queue (via disposing it) between the isEmpty() check and peek() itself.
• In addition, if the task was disposed between the peek and remove, the remove could remove the new head of the queue which is no longer what peek saw and results in tasks being dropped.

[codecov]

Codecov Report

Merging #5661 into 2.x will decrease coverage by 0.03%.
The diff coverage is 97.18%.

@@             Coverage Diff              @@
##                2.x    #5661      +/-   ##
============================================
- Coverage     96.24%   96.21%   -0.04%     
+ Complexity     5853     5818      -35     
============================================
  Files           635      633       -2     
  Lines         41651    41553      -98     
  Branches       5768     5752      -16     
============================================
- Hits          40089    39982     -107     
- Misses          616      621       +5     
- Partials        946      950       +4

Impacted Files	Coverage Δ	Complexity Δ
...in/java/io/reactivex/schedulers/TestScheduler.java	100% <100%> (ø)	11 <0> (ø)	⬇️
...x/internal/operators/flowable/FlowableTimeout.java	97.54% <96.89%> (+4.3%)	3 <0> (ø)	⬇️
...ernal/operators/flowable/FlowableTimeoutTimed.java	98.37% <97.56%> (+5.04%)	3 <0> (-1)	⬇️
...ava/io/reactivex/processors/BehaviorProcessor.java	88.49% <0%> (-4.87%)	62% <0%> (ø)
...erators/completable/CompletableConcatIterable.java	95.91% <0%> (-4.09%)	2% <0%> (ø)
...l/operators/observable/ObservableFlatMapMaybe.java	84.96% <0%> (-3.93%)	2% <0%> (ø)
...ternal/operators/observable/ObservablePublish.java	92.98% <0%> (-3.51%)	11% <0%> (ø)
.../io/reactivex/internal/schedulers/IoScheduler.java	89.24% <0%> (-3.23%)	9% <0%> (ø)
...a/io/reactivex/internal/util/QueueDrainHelper.java	60.28% <0%> (-2.84%)	32% <0%> (-2%)
...rnal/operators/flowable/FlowableFlatMapSingle.java	92.93% <0%> (-2.72%)	2% <0%> (ø)
... and 34 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 e02cb58...46c39f2. Read the comment docs.

[artem-zinnatullin]

nit: maybe leave a link to your blog post in comment? So readers could understand the pattern used here

[akarnokd]

Not really keen on this. The PR will reference the blog.

[artem-zinnatullin]

Right, my point is not to promote your blog haha

Just want to leave some bread crumbs in the code so one could understand that MAX_VALUE is used as a notion of terminal state here, one might confuse it with something related to unbounded request mode

I'd just put

@see <a href="https://akarnokd.blogspot.hu/2017/09/java-9-flow-api-timing-out-events.html">Operator Design Principles</a>

To the class's javadoc.

It's obviously a nit and up2u :)

[akarnokd]

Thanks for the suggestion, but I won't put it there.

[artem-zinnatullin]

Extract get() result to local var here?

[akarnokd]

Sure.