1.x: reduce stack depth with switchIfEmpty

[akarnokd]

Stack depths may get deep if switchIfEmpty has to switch to the alternative source at the end of an already long chain leading to the onCompleted. Using multiple switchIfEmpty amplifies the impact on the stack depth.

This PR introduces trampolined subscribing to both sources (similar to how concat works), thus given synchronous sources, the stack depth rewinds to the level it was when the switchIfEmpty was subscribed to. In addition, to reduce the depth even further, the operator has been turned into an OnSubscribe-based implementation to avoid the extra depth due to lift.

Related: rxjava StackOverflowError exception in long chain

[codecov]

Codecov Report

Merging #5125 into 1.x will decrease coverage by -0.05%.
The diff coverage is 85.71%.

@@             Coverage Diff              @@
##                1.x    #5125      +/-   ##
============================================
- Coverage     84.38%   84.34%   -0.05%     
- Complexity     2876     2878       +2     
============================================
  Files           290      290              
  Lines         18070    18085      +15     
  Branches       2468     2473       +5     
============================================
+ Hits          15249    15254       +5     
- Misses         1951     1966      +15     
+ Partials        870      865       -5

Impacted Files	Coverage Δ	Complexity Δ
src/main/java/rx/Observable.java	99.45% <100%> (ø)	448 <ø> (ø)	❌
...x/internal/operators/OnSubscribeSwitchIfEmpty.java	90.32% <85%> (ø)	2 <1> (?)
.../rx/internal/operators/OperatorWindowWithTime.java	42.04% <ø> (-6.44%)	3% <ø> (ø)
...n/java/rx/subjects/SubjectSubscriptionManager.java	80.71% <ø> (-1.43%)	15% <ø> (-1%)
...main/java/rx/internal/operators/OperatorMerge.java	86.6% <ø> (+0.23%)	7% <ø> (ø)	❌
...n/java/rx/subscriptions/CompositeSubscription.java	77.92% <ø> (+1.29%)	25% <ø> (+1%)	✅
.../rx/internal/schedulers/CachedThreadScheduler.java	89.32% <ø> (+1.94%)	6% <ø> (ø)	❌
.../java/rx/internal/operators/BackpressureUtils.java	72.72% <ø> (+2.27%)	29% <ø> (+1%)	✅
...rx/internal/operators/OnSubscribeFromIterable.java	93.9% <ø> (+2.43%)	5% <ø> (ø)	❌
...ain/java/rx/internal/schedulers/SchedulerWhen.java	87.83% <ø> (+4.05%)	4% <ø> (ø)	❌
... and 1 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 5468972...f677a8c. Read the comment docs.

[JakeWharton]

Is this variable needed? Seems like source == null would accomplish the same thing.

[akarnokd]

Unfortunately, no. Here is a scenario:

1. Thread 1 subscribes and calls subscribe(source)
2. through subscribeOn, Thread 2 calls onCompleted(), and thus subscribe(null) but finds Thread 1 still inside the loop (wip != 0).
3. Thread 1 detects the completion of Thread 2 and loops back. Since source != null still in this method, the main source is resubscribed again instead of the alternative.

The secondary, persisted indicator works across threads and makes sure no matter where onComplete comes from, the second time we get into that inner part, the alternative source is subscribed next.

[akarnokd]

On a second thought, you are right. If I set source = null after the source.unsafeSubscribe() the next round will properly go into the alternative. Updated the PR.