Приложение падает после восстановления интернета, RxJava и Retrofit

Question

Мне нужно получить данные из Api для каждого элемента списка, а также мне нужно реализовать проверку интернет-соединения.

1. Затем программа запускается, все прекрасно работает.
2. Я отключаю Интернет в прямом эфиреи снова запустить метод (используя swipeRefreshLayout), я получаю InternetErrorToast, отлично!
3. Я восстановил Интернет в режиме реального времени и снова запустить метод (используя swipeRefreshLayout), приложение вылетело с ошибкой.

Теперь у меня есть следующий метод.Заметка!dataManager.getCityConditionsResponse () return Single

@Override
public void updateCitiesList() {
    List<City> citiesList = dataManager.getCitiesFromDb();
    if (!dataManager.isInternetConnected()) {
        view.showCitiesList(citiesList);
        view.showInternetErrorToast();
    } else {
        compositeDisposable.add(Observable.fromIterable(citiesList)
                .doOnNext(city -> dataManager.getCityConditionsResponse(city.getQuery())
                        .subscribe(
                                response -> {
                                    city.setTemp(response.getTemp());
                                    city.setIcon(response.getIcon());
                                },
                                error -> view.showServerErrorToast()))
                .toList()
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(
                        list -> view.showCitiesList(list),
                        error -> view.showServerErrorToast()
                ));
    }
    view.hideRefreshingStatus();
}

На шаге 3 у меня появляется эта ошибка

018-10-12 16:34:54.033 19013-19046/mike.weather E/AndroidRuntime: FATAL EXCEPTION: RxCachedThreadScheduler-1
Process: mike.weather, PID: 19013
io.reactivex.exceptions.CompositeException: 2 exceptions occurred. 
    at io.reactivex.internal.observers.ConsumerSingleObserver.onError(ConsumerSingleObserver.java:49)
    at io.reactivex.internal.operators.observable.ObservableSingleSingle$SingleElementObserver.onError(ObservableSingleSingle.java:93)
    at retrofit2.adapter.rxjava2.BodyObservable$BodyObserver.onError(BodyObservable.java:72)
    at retrofit2.adapter.rxjava2.CallExecuteObservable.subscribeActual(CallExecuteObservable.java:56)
    at io.reactivex.Observable.subscribe(Observable.java:12090)
    at retrofit2.adapter.rxjava2.BodyObservable.subscribeActual(BodyObservable.java:34)
    at io.reactivex.Observable.subscribe(Observable.java:12090)
    at io.reactivex.internal.operators.observable.ObservableSingleSingle.subscribeActual(ObservableSingleSingle.java:35)
    at io.reactivex.Single.subscribe(Single.java:3438)
    at io.reactivex.Single.subscribe(Single.java:3424)
    at mike.weather.ui.main.MainActivityPresenter.lambda$updateCitiesList$3(MainActivityPresenter.java:49)
    at mike.weather.ui.main.-$$Lambda$MainActivityPresenter$UqLPaAef0SB9PT-Rz654tgX3dnA.accept(Unknown Source:4)
    at io.reactivex.internal.operators.observable.ObservableDoOnEach$DoOnEachObserver.onNext(ObservableDoOnEach.java:93)
    at io.reactivex.internal.operators.observable.ObservableDoOnEach$DoOnEachObserver.onNext(ObservableDoOnEach.java:101)
    at io.reactivex.internal.operators.observable.ObservableFromIterable$FromIterableDisposable.run(ObservableFromIterable.java:98)
    at io.reactivex.internal.operators.observable.ObservableFromIterable.subscribeActual(ObservableFromIterable.java:58)
    at io.reactivex.Observable.subscribe(Observable.java:12090)
    at io.reactivex.internal.operators.observable.ObservableDoOnEach.subscribeActual(ObservableDoOnEach.java:42)
    at io.reactivex.Observable.subscribe(Observable.java:12090)
    at io.reactivex.internal.operators.observable.ObservableDoOnEach.subscribeActual(ObservableDoOnEach.java:42)
    at io.reactivex.Observable.subscribe(Observable.java:12090)
    at io.reactivex.internal.operators.observable.ObservableToListSingle.subscribeActual(ObservableToListSingle.java:58)
    at io.reactivex.Single.subscribe(Single.java:3438)
    at io.reactivex.internal.operators.single.SingleDoOnSuccess.subscribeActual(SingleDoOnSuccess.java:35)
    at io.reactivex.Single.subscribe(Single.java:3438)
    at io.reactivex.internal.operators.single.SingleSubscribeOn$SubscribeOnObserver.run(SingleSubscribeOn.java:89)
    at io.reactivex.Scheduler$DisposeTask.run(Scheduler.java:578)
    at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:66)
    at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:57)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:301)
  ........................

Answer 1

Было бы лучше Rx идиоматично, если бы вы так переписали свою наблюдаемую

compositeDisposable.add(Observable.fromArray(citiesList)
            .flatMap(Observable::fromIterable)
            .flatMapSingle(city -> 
                dataManager.getCityConditionsResponse(city.getQuery())
                    .map(response -> { 
                        city.setTemp(response.getTemp());
                        city.setIcon(response.getIcon()); 
                        return city;
                    })
            )
            .toList()
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(
                    list -> view.showCitiesList(list),
                    error -> view.showServerErrorToast()
            ));

После этого я думаю, что ваша проблема исчезнет.