Ожидание завершения двух наблюдаемых, прежде чем продолжить?

Question

У меня есть страница загрузки, где я хочу выполнить два сетевых запроса (retrofit2-rxjava) для несвязанной информации. Я не хочу переходить на следующую страницу, пока эти два запроса не будут завершены, даже если один или оба не удаются.

1. Связал запросы вместе, используя zip. Есть ли способ не использовать BiFunction и не возвращать ноль?

2. Запросы A и B имеют .doOnNext и .doOnError. если один из них возвращает ошибку, продолжается ли наблюдаемая молния? и zip-подписчик также возвращает ошибку?

3. Это лучший способ сделать это?

private Disposable retrieveBothThings() {
return Observable.zip(getThingA(), getThingB(),
                    (A, B) -> {
                        onAllCallsComplete();
                        return null;
                    })
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(o -> {}, Logger::e);
}



private Observable<...> getThingA() {
            return SessionManager.getInstance().getApi()
                    .getA()
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .doOnNext(this::onACompleted)
                    .doOnError(this::onAFailed);
}

private Observable<...> getThingB() {
        return SessionManager.getInstance().getApi()
                .getB()
                .subscribeOn(Schedulers.io())
              .observeOn(AndroidSchedulers.mainThread()).toObservable()
                .doOnNext(this::onBSuccess)
                .doOnError(this::onBFailure);
}

private void onBSuccess(...) {
    ...        
}

private void onBFailure(final Throwable throwable) {
    Logger.e(throwable);
}

private void onACompleted(...) {
    ...        
}

private void onAFailed(final Throwable throwable) {
    Logger.e(throwable);
}

Answer 1

Вы можете объединить две наблюдаемые, используя оператор merge(). Вы можете преобразовать ошибку в onComplete(), используя onErrorResumeNext().

Completable.merge(
  observable1
    .doOnNext(this::onACompleted)
    .doOnError(this::onAFailed)
    .onErrorResumeNext( Completable.complete() )
    .toCompletable(),
  observable2
    .doOnNext(this::onBCompleted)
    .doOnError(this::onBFailed)
    .onErrorResumeNext( Completable.complete() ),
    .toCompletable() )
.subscribe( ignore -> {}, 
            error -> {},
            () -> { processCompletion(); } );