Комбинация нескольких методов в RXJava2

01 мая 2021, 18:30

Дело в том, что мне необходимо одновременно подтягивать данные из локальной базы, с сервера, при этом проверять подключение к интернету.

Без проверки интернета получается легко. Но при отключении мобильных данных, крашится.

Не пойму как скомбинировать и решил поступить таким образом:

private void getCategories() {
    composite.add(getDataFromLocal(context)
            .observeOn(AndroidSchedulers.mainThread()).flatMap(new Function<PromoFilterResponse, ObservableSource<List<FilterCategory>>>() {
                @Override
                public ObservableSource<List<FilterCategory>> apply(PromoFilterResponse promoFilterResponse) throws Exception {
                    if (promoFilterResponse != null) {
                        PreferencesHelper.putObject(context, PreferencesKey.FILTER_CATEGORIES_KEY, promoFilterResponse);
                        return combineDuplicatedCategories(promoFilterResponse);
                    } else {
                        return Observable.empty();
                    }
                }
            })
            .subscribe(new Consumer<List<FilterCategory>>() {
                @Override
                public void accept(List<FilterCategory> categories) throws Exception {
                    if (mView != null) {
                        mView.hideConnectingProgress();
                        if (categories != null && categories.size() > 0) {
                            mView.onCategoriesReceived(categories);
                        }
                    }
                }
            }));
    composite.add(InternetUtil.isConnectionAvailable().subscribe(isOnline -> {
        if (isOnline) {
            composite.add(
                    getDataFromServer(context)
                            .flatMap(new Function<PromoFilterResponse, ObservableSource<List<FilterCategory>>>() {
                                @Override
                                public ObservableSource<List<FilterCategory>> apply(PromoFilterResponse promoFilterResponse) throws Exception {
                                    if (promoFilterResponse != null) {
                                        PreferencesHelper.putObject(context, PreferencesKey.FILTER_CATEGORIES_KEY, promoFilterResponse);
                                        return combineDuplicatedCategories(promoFilterResponse);
                                    } else {
                                        return Observable.empty();
                                    }
                                }
                            })
                            .observeOn(AndroidSchedulers.mainThread())
                            .subscribe(categories -> {
                                if (mView != null) {
                                    mView.hideConnectingProgress();
                                    if (categories != null && categories.size() > 0) {
                                        mView.onCategoriesReceived(categories);
                                    } else {
                                        mView.onCategoriesReceivingFailure(errorMessage[0]);
                                    }
                                }
                            }, throwable -> {
                                if (mView != null) {
                                    if (throwable instanceof HttpException) {
                                        ResponseBody body = ((HttpException) throwable).response().errorBody();
                                        if (body != null) {
                                            errorMessage[0] = body.string();
                                        }
                                    }
                                    mView.hideConnectingProgress();
                                    mView.onCategoriesReceivingFailure(errorMessage[0]);
                                }
                            }));
        } else {
            mView.hideConnectingProgress();
            mView.showOfflineMessage();
        }
    }));
} 

private Single<Boolean> checkNetwork(Context context) {
    return InternetUtil.isConnectionAvailable()
            .subscribeOn(Schedulers.io())
            .doOnSuccess(new Consumer<Boolean>() {
                @Override
                public void accept(Boolean aBoolean) throws Exception {
                    getDataFromServer(context);
                }
            });
}
private Observable<PromoFilterResponse> getDataFromServer(Context context) {
    return RetrofitHelper.getApiService()
            .getFilterCategories(Constants.PROMO_FILTER_CATEGORIES_URL)
            .subscribeOn(Schedulers.io())
            .retryWhen(BaseDataManager.isAuthException())
            .publish(networkResponse ->  Observable.merge(networkResponse,  getDataFromLocal(context).takeUntil(networkResponse)))
            .doOnNext(new Consumer<PromoFilterResponse>() {
                @Override
                public void accept(PromoFilterResponse promoFilterResponse) throws Exception {
                    PreferencesHelper.putObject(context, PreferencesKey.FILTER_CATEGORIES_KEY, promoFilterResponse);
                }
            })
            .doOnError(new Consumer<Throwable>() {
                @Override
                public void accept(Throwable throwable) throws Exception {
                    LogUtil.e("ERROR", throwable.getMessage());
                }
            });
}
private Observable<PromoFilterResponse> getDataFromLocal(Context context) {
    PromoFilterResponse response = PreferencesHelper.getObject(context, PreferencesKey.FILTER_CATEGORIES_KEY, PromoFilterResponse.class);
    if (response != null) {
        return Observable.just(response)
                .subscribeOn(Schedulers.io());
    } else {
        return Observable.empty();
    }
}

Как видите, отдельно подгружаю локальную базу, параллельно проверяю интернет и подгружаю данные с сервера.

Но мне кажется это не совсем правильно. Тем более подписчик дублируется и тд.

Видел много туториалов, где описывается комбинация локальной базы с АПИ, но не видел чтоб одновременно обрабатывали ошибку соединения с интернетом.

Думаю многие сталкивались с такой задачей и как вы это решали?

Answer 1

Просто оборачиваем результат в свою обертку с необходимыми флагами или данными. Думаю на этом примере понятно как решить проблему

static class Resource<T> {
    @Nullable final T mValue;
    @Nullable final Throwable mThrowable;
    static <T> Resource<T> success(@Nullable final T value) {
        return new Resource<>(value, null);
    }
    static <T> Resource<T> error(@Nullable final Throwable error) {
        return new Resource<>(null, error);
    }
    Resource(@Nullable final T value, @Nullable final Throwable throwable) {
        mValue = value;
        mThrowable = throwable;
    }
    boolean isSuccess() {
        return mThrowable == null;
    }
}
public final Flowable getCategories() {
    final Flowable networkFlowable = Flowable.just(Resource.success(Arrays.asList(1, 2, 3)))
            .onErrorReturn(throwable -> Resource.error(throwable))
            .map(Optional::of)
            .startWith(Optional.empty());// Запрос в сеть
    final Flowable localFlowable = Flowable.just(Resource.success(Arrays.asList(1, 2, 3)))
            .map(Optional::of)
            .startWith(Optional.empty()); // Запрос локальных данных
    return Flowable.combineLatest(localFlowable, networkFlowable, (Optional<Resource<List<Integer>>> localData, Optional<Resource<List<Integer>>> networkData) -> {
        if (networkData.isPresent()) {
            final Resource<List<Integer>> networkDataResource = networkData.get();
            if (networkDataResource.isSuccess()) {
                //Данные успешно загружены
                return ViewModel(networkDataResource.mValue);
            } else if (localData.isPresent() {
                return ViewModel(localData.get().mValue, networkDataResource.mThrowable); //Ошибка при загрузке оставляем старые данные и добавляем данные об ошибке
            }
        } else if (localData.isPresent()) {   //Если данных с сети нет то показываем локальные
            final Resource<List<Integer>> localDataResource = localData.get();
            if (localDataResource.isSuccess()) {
                return ViewModel(localDataResource.mValue);
            } else {
                return ViewModel(localData.get().mValue, localDataResource.mThrowable); //Ошибка при загрузке добавляем данные об ошибке
            }
        } else {
            return EmptyViewModel();
        }
    });
}