RxJava error handling withought swallowing the error

Question

I'm a little bit new to RxJava. I am trying to emit another item if onError() get called without losing the error(I still want onError() to be called on the observer). but when I'm implementing each of the error handling methods declared in the docs the error being swallowed and on error isn't being called. any solutions?

edit: that's what I've tried to do yesterday -

@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;
                        }
                    }));
        }
    });
}

today I tought I might doing it wrong and tried -

@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;
}

but I've got the same result. the onNext() called with the data of the second observable and the onError() not being called afterwards.

Why are you doing new Observable? Also you implemented it incorreclty. Why not use Observable.fromCallable there? Do you have observeOn anywhere else? Use this overload to prevent an error cutting ahead. Why do you use the error(Callable) overload instead of the direct overload I suggested? — akarnokd
couldn't found the overload you suggested. I have observeOn() in my tranformer so I tried downstream.observeOn(scheduler,true) but it's still swallowing the error. — Ofek Regev
Then it's time to add doOnNext and doOnError operators all around to see where events disappear. — akarnokd
Solved by finding the overload you suggested initially and add dowstream.observeOn(scheduler,true). Thanks for the information. — Ofek Regev

akarnokd akarnokd · Accepted Answer · 2018-10-21T13:55:03

Resume with the desired value concatenated with the original error:

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

RxJava error handling withought swallowing the error

1 Answers