CombineLatestDelayError emits error before all inner observables emit result

[UMFsimke]

Recently I came across following issue while using combineLatestDealyError operator. This code has been tested on Android, Nexus 5x running OS 6.0.1.

Observable<Boolean> errorObservable = Observable.create(subscriber -> {
    subscriber.onError(new NullPointerException());
});

Observable timeoutObservable = Observable.create(subscriber -> {
    subscriber.onNext(true);
    subscriber.onCompleted();
})
.delay(5, TimeUnit.SECONDS, AndroidSchedulers.mainThread())
.doOnNext(result -> Log.d("Combine", "Delay emitted"));

Observable<Object> zippedObservable = Observable.zip(
        Observable.just(true), timeoutObs,
        (flag1, flag2) -> {
            Log.d("Combine", "On Zip emitted");
            return new Object();
        });

return Observable.combineLatestDelayError(
        Arrays.asList(errorObservable, zippedObservable),
        result -> {
            Boolean flag1 = (Boolean) result[0];
            Object object = result[1];
            Log.d("Combine", "Result of errorObservable: " + flag1);
            Log.d("Combine", "Result of zippedObservable: " + object);
            return null;
        }).subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(new Subscriber<Object>() {
            @Override
            public void onCompleted() {
                Log.d("Combine", "OnCompleted");
            }

            @Override
            public void onError(Throwable e) {
                Log.d("Combine", "OnError");
            }

            @Override
            public void onNext(Object o) {
                Log.d("Combine", "OnNext");
            }
        });

This piece of code is trying to simulate callbacks using delays - for example Android Pay availability callback.

Problem here arises once this code is run. The log will be:

OnError
Delay emitted
On Zip emitted

This is quite contradictory to the docs that state that expected behavior should be:

Delay emitted
On Zip emitted
OnError

The expected result does happens if you add subscriber.onNext(true) to the errorObservable before subscriber.onError...

[akarnokd]

Your errorObservable doesn't emit any value and since combineLatestDelayError won't ever be able to combine values, it terminates with the error immediately.

[UMFsimke]

Docs state:

Concatenates the Observable sequence of Observables into a single sequence by subscribing to each inner Observable, one after the other, one at a time and delays any errors till the all inner and the outer Observables terminate.

In diff:
https://github.com/ReactiveX/RxJava/pull/3759/files#diff-c3bbc6e9e497930d46361b0b8140431cR1163

So empahsis is on:

and delays any errors till the ALL inner and the outer Observables terminate.

In this scenario second observable is not finished and error is not delayed.

What if you want to make two API calls no matter of the result of previous and combine their result? I do understand that there is always different type of solution and approach but I just believe that either docs are not correct or this is potential bug since it should behave as stated.

[akarnokd]

The documentation is incorrect as there is no "outer Observable" with this operator. Otherwise, it's an undocumented corner case that needs a revisit. Note however that even if the operator delayed the error to the very end, you wouldn't see any combined values ever (but perhaps the errors of the others).
I'll see what it takes to make the operator wait out all events.

[akarnokd]

This is more complicated to work out the proper rules. For example, what should happen if you combineLatestDelayError(empty(), error(x), func2) or combineLatestDelayError(error(x), empty(), just(1), func3)?

[UMFsimke]

For combineLatestDelayError(empty(), error(x), func2) it makes sense that empty() will terminate and onError will be invoked?

For second, I would say empty(), then just(1) then onError will be invoked?

I'm just following up with an idea, its not that I expect it to work that way.

[akarnokd]

combineLatestDelayError can wait out all terminal events to make sure errors are reported after. The problem is that if you write combineLatest(empty(), just()) and combineLatestDelayError(empty(), just()), they are no longer behaving the same way.

[akarnokd]

/cc @zsxwing

[akarnokd]

Sorry for the delay. Once RC3 is released, I'll look into it and work out the details.

[akarnokd]

I've considered many alternatives and can give you a reasonable workaround: apply cache() on the source Observable which you want to be consumed entirely even if the combineLatest cancels its consumer.

[akarnokd]

I spent a considerable amount of time on this and was unable to come up with a reasonable way for changing combineLatest inside RxJava. All I can offer is I write you a custom operator that awaits all sources to terminate in some fashion, regardless of having been emitted values or not.