Combination of several methods in RXJava2

Question

The fact is that I need to simultaneously pull in data from the local database, from the server, while checking the connection to the Internet.

Without checking the internet is easy. But when I turn off mobile data, crashes.

I do not understand how to combine and decided to do this:

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

As you can see, connect the local database separately, simultaneously check the Internet and upload data from the server.

But it seems to me not quite right. Moreover, the subscriber is duplicated and so on.

I saw a lot of tutorials, where the combination of the local database with the API is described, but I did not see it at the same time processing the connection error with the Internet.

I think many people faced such a problem and how did you solve it?

Answer 1

Im' not android developer, but in my mind methods return types should be something like this:

//just for demonstration
static boolean isOnline = false;

static class NoInternet extends RuntimeException {
}

private static Completable ensureOnline() {
    if (isOnline)
        return Completable.complete();
    else
        return Completable.error(new NoInternet());

}

private static Single<String> getDataFromServer() {
    return Single.just("From server");
}

private static Maybe<String> getDataFromLocal() {
    return Maybe.just("From local");//or Maybe.never()
}

We can run all in parallel with Observable.merge. But what if error NoIternet happens? Merged observable will fail. We can use materialisation - transform all emission and errors to onNext value.

private static void loadData() {

    Observable<Notification<String>> fromServer = ensureOnline().andThen(getDataFromServer()).toObservable().materialize();

    Observable<Notification<String>> fromLocaldb = getDataFromLocal().toObservable().materialize();

    Observable.merge(fromLocaldb, fromServer)
            .subscribe(notification -> {
                if (notification.isOnNext()) {
                    //calls one or two times(db+server || db || server)
                    //show data in ui
                } else if (notification.isOnError()) {
                    if (notification.getError() instanceof NoInternet) {
                        //show no internet
                    } else {
                        //show another error
                    }
                } else if (notification.isOnComplete()){
                    //hide progress bar
                }



            });
}

Answer 2

Suppose You have two Obsevable: one from server and another from database

You can merge them into one stream like below:

  public Observable<Joke> getAllJokes() {

    Observable<Joke> remote = mRepository.getAllJokes()
            .subscribeOn(Schedulers.io());


    Observable<Joke> local = mRepository.getAllJokes().subscribeOn(Schedulers.io());

      return Observable.mergeDelayError(local, remote).filter(joke -> joke != null);
}