RxJava 错误处理而不吞没错误

Question

我对 RxJava 有点陌生。如果在不丢失错误的情况下调用 onError()，我将尝试发出另一个项目（我仍然希望在观察者上调用 onError()）。但是当我实现在 docs 中声明的每个错误处理方法时，错误被吞没并且没有被调用。有什么解决办法吗？

编辑：这就是我昨天尝试做的 -

@Override
public Observable<ArrayList<Address>> getAirports() {
    return new Observable<ArrayList<AirportPOJO>>() {
        @Override
        protected void subscribeActual(Observer<? super ArrayList<AirportPOJO>> observer) {
            try {
                // get airports from api list and map it
                ArrayList<AirportPOJO> airportsList = apiDb.getAirportsList(POJOHelper.toPOJO(AppCredentialManager.getCredentials()));
                observer.onNext(airportsList);
            } catch (Exception e) {
                e.printStackTrace();
                observer.onError(handleException(e));
            }
        }
    }.map(AirportsMappers.getAirportsPojoToDomainAirportsMapper()).doOnNext(new Consumer<ArrayList<Address>>() {
        @Override
        public void accept(ArrayList<Address> airportsList) throws Exception {
            // if airports loaded from api - save them to local db
            if (airportsList != null) {
                try {
                    localDb.saveAirportList(AirportsMappers.getAirportsToLocalDbAirportsMapper().apply(airportsList));
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }).onErrorResumeNext(new Function<Throwable, ObservableSource<? extends ArrayList<Address>>>() {
        @Override
        public ObservableSource<? extends ArrayList<Address>> apply(final Throwable throwable) throws Exception {
            // load the local airports - 
            ArrayList<LocalDbAirportEntity> localAirportsEntities = localDb.getAirports();
            // map
            ArrayList<Address> airports = AirportsMappers.getLocalDbAirportsToAirportsMapper().apply(localAirportsEntities);
            // return the concat observable with the error
            return Observable.just(airports).concatWith(Observable.
                    <ArrayList<Address>>error(new Callable<Throwable>() {
                        @Override
                        public Throwable call() throws Exception {
                            return throwable;
                        }
                    }));
        }
    });
}

今天我意识到我可能做错了并尝试过 -

@Override
public Observable<ArrayList<Address>> getAirports() {
    ArrayList<Observable<ArrayList<Address>>> observables = new ArrayList<>();
    observables.add(apiDb.getAirportsList(POJOHelper.toPOJO(AppCredentialManager.getCredentials())).map(AirportsMappers.getAirportsPojoToDomainAirportsMapper()));
    observables.add(localDb.getAirports().map(AirportsMappers.getLocalDbAirportsToAirportsMapper()));
    Observable<ArrayList<Address>> concatenatedObservable = Observable.concatDelayError(observables);
    return concatenatedObservable;
}

但我得到了相同的结果。 onNext() 使用第二个 observable 的数据调用，之后没有调用 onError()。

Answer 1

使用连接原始错误的所需值继续：

source.onErrorResumeNext(error ->
    Observable.just(item).concatWith(Observable.<ItemType>error(error))
);

RxJava 错误处理而不吞没错误

RxJava error handling withought swallowing the error

java

rx-java

reactivex