2.x: Add Flowable.switchMapCompletable{DelayError} operator

[akarnokd]

This PR adds the Flowable.switchMapCompletable and Flowable.switchMapCompletableDelayError operators as requested by #4853.

The associated new marbles are:

[codecov]

Codecov Report

Merging #5870 into 2.x will increase coverage by 0.07%.
The diff coverage is 98.98%.

@@             Coverage Diff              @@
##                2.x    #5870      +/-   ##
============================================
+ Coverage     96.54%   96.61%   +0.07%     
- Complexity     5862     5868       +6     
============================================
  Files           647      648       +1     
  Lines         42766    42865      +99     
  Branches       5933     5952      +19     
============================================
+ Hits          41289    41416     +127     
+ Misses          568      552      -16     
+ Partials        909      897      -12

Impacted Files	Coverage Δ	Complexity Δ
src/main/java/io/reactivex/Flowable.java	100% <100%> (ø)	541 <2> (+2)	⬆️
.../operators/mixed/FlowableSwitchMapCompletable.java	98.94% <98.94%> (ø)	2 <2> (?)
.../io/reactivex/internal/functions/ObjectHelper.java	94.73% <0%> (-5.27%)	20% <0%> (-1%)
...tivex/internal/schedulers/TrampolineScheduler.java	92.3% <0%> (-2.57%)	6% <0%> (ø)
...perators/single/SingleFlatMapIterableFlowable.java	95.83% <0%> (-2.5%)	2% <0%> (ø)
...va/io/reactivex/internal/queue/SpscArrayQueue.java	97.61% <0%> (-2.39%)	22% <0%> (-1%)
...nternal/operators/parallel/ParallelReduceFull.java	91.17% <0%> (-1.97%)	2% <0%> (ø)
...internal/operators/flowable/FlowableSwitchMap.java	95.28% <0%> (-1.42%)	3% <0%> (ø)
...ava/io/reactivex/processors/BehaviorProcessor.java	95.06% <0%> (-0.9%)	60% <0%> (ø)
...nternal/operators/observable/ObservableCreate.java	97.43% <0%> (-0.86%)	2% <0%> (ø)
... and 25 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 84004a6...75b7558. Read the comment docs.

[artem-zinnatullin]

Looks good, but comment about mapper

[artem-zinnatullin]

[all of the]m

[artem-zinnatullin]

hm, I'm looking at Flowable.switchMap and Observable.switchMap and they dispose current first:

RxJava/src/main/java/io/reactivex/internal/operators/flowable/FlowableSwitchMap.java

Line 109 in 8068404

inner.cancel();

RxJava/src/main/java/io/reactivex/internal/operators/observable/ObservableSwitchMap.java

Line 105 in 8068404

inner.cancel();

While this operator maps first and then disposes current which give current more time to work which makes it more delayErrors friendly I guess, however I still have a question about delayErrors below

[artem-zinnatullin]

I've tried to adapt Flowable.switchMap behavior to respect delayErrors if mapper throws but failed so far, this is the test I wrote, maybe it'll help you:

@Test
public void delayErrorsErrorInMapper() {
    final PublishProcessor<Object> trigger = PublishProcessor.create();

    TestSubscriber<Integer> ts = Flowable
            .just(1, 2)
            .switchMapDelayError(new Function<Integer, Publisher<Integer>>() {
                @Override
                public Publisher<Integer> apply(Integer integer) throws Exception {
                    if (integer.equals(1)) {
                        return Flowable
                                .combineLatest(Flowable.just(1), trigger, new BiFunction<Integer, Object, Integer>() {
                                    @Override
                                    public Integer apply(Integer integer, Object o) {
                                        return integer;
                                    }
                                })
                                .doOnCancel(new Action() {
                                    @Override
                                    public void run() throws Exception {
                                        trigger.toString();
                                    }
                                });
                    } else {
                        // Fail for second item.
                        throw new TestException("Item " + integer + " mapping failure.");
                    }
                }
            })
            .test();

    ts.assertNoValues();
    ts.assertNotTerminated();

    trigger.onNext(new Object());

    ts.assertValue(1);
    ts.assertError(new Predicate<Throwable>() {
        @Override
        public boolean test(Throwable throwable) {
            return throwable instanceof TestException && throwable.getMessage().equals("Item 2 mapping failure.");
        }
    });
}

[akarnokd]

Yes, essentially it gives a larger window for errors but otherwise has no effect because there are no items to forward and a late completion has no effect anyway.

Mapping errors are terminal errors, as if the upstream ended with a failure. The problem is that the operator can't decide if it is okay to continue. You can always do try-catch and return a Completable.error.

[artem-zinnatullin]

Why do we have to do it in loop? Do we expect concurrent onNext from upstream? Or I'm missing something…

[akarnokd]

current could be null or the previous inner observer. When that old inner observer is completing, it CASes itself to null, which fails this CAS and forces a retry. In addition, a dispose can also change the inner to the INNER_DISPOSED concurrently at which point it shouldn't be overridden and the loop should simply end.

[akarnokd]

Rebased and refactored due to concatMapCompletable introducing the internal.mixed package.