Wsrtjtyk

Currently I have a method in a repository class which fetches data from both a local cache and a remote API.

public Observable<List<Items>> getItemsForUser(String userId {

    return Observable.concatArrayEager(

            getUserItemsLocal(userId), // returns Observable<List<Items>>

            getUserItemsRemote(userId) // returns Observable<List<Items>>

    );

}

Currently, the method fetches the local data first (which may be outdated) and returns it, then updates it with the fresh data from the remote API.

I want to change the implementation to use Observable.merge so that if the remote API request completes first, that data gets shown first. However, if I just use Observable.merge I'm concerned that the local database request may return stale data, which will then overwrite the fresh data from the remote.

Basically, I want something like:

public Observable<List<ShoutContent>> getItemsForUser(String userId, ErrorCallback errorCallback) {

    return Observable.merge(

            getUserItemsRemote(userId),

            getUserItemsLocal(userId)

                .useOnlyIfFirstResponse()

}

So if the remote API request completes first, then that response is the only one that gets returned. But if the local request completes first, I want to return that, and then return the remote request once it is completed. Does RxJava have anything like this built in?

Edit: I would like to add that getUserItemsRemote does update the local database when the Observable emits, but I don't think that I can ensure that the database will be updated before the local request completes, which leaves the possibility that the local request will respond with stale data.

You can make use of the takeUntil operator.

takeUntil returns an Observable that emits the items emitted by the source Observable until a second ObservableSource emits an item.

In your case, you need to stop observing the local observable, once the remote Observable is emitted. The code is demonstrated below.

public Observable<String> getUserItemsLocal() {

    return Observable.just("Local db response")

            .delay(5, TimeUnit.SECONDS);  // assume local db takes 5 seconds to emit

}





public Observable<String> getUserItemsRemote() {

    return Observable.just("Remote Data")

            .delay(1, TimeUnit.SECONDS); // remote data comes quicker, in 1 second

}

Your repository code goes like

Observable<String> remoteResponse = getUserItemsRemote();

getUserItemsLocal().takeUntil(remoteResponse)

            .mergeWith(remoteResponse)

            .subscribe(new Consumer<String>() {

                @Override

                public void accept(String s) throws Exception {

                    Log.d(TAG, "result: " + s);

                }

            });