RxJava Tidbits #4: Combining multiple data sources

It’s been more than a year since the last update… so here is Tidbit #4:

This snippet is especially useful in the world of Android where you are usually dealing with two (or more) data sources. For example, a local database containing cached data for offline access (SQLite) and a remote JSON API (RESTful web service) providing the latest, fresh data.

Let’s assume both sources provide their data via repositories using a common interface:

interface Repository {

    Observable<Data> getData();
}

One could assume that the local repository is normally much faster to emit its data than the remote repository. However, that is not guaranteed.

We want to start fetching the data from both repositories simultaneously and emit the local data first, followed by the remote data unless the remote repository emits earlier in which case it should supersede and stop the local repository from emitting its data. Simply said: The remote repository takes priority.

This is how you do it:

static class CompositeRepository implements Repository {

    private final Repository localRepository;
    private final Repository remoteRepository;

    CompositeRepository(Repository localRepository,
                        Repository remoteRepository) {
        this.localRepository = localRepository;
        this.remoteRepository = remoteRepository;
    }

    @Override
    public Observable<Source> getData() {
        return remoteRepository.getData()
                .publish(new Func1<Observable<Source>, Observable<Source>>() {
                    @Override
                    public Observable<Source> call(Observable<Source> remoteObservable) {
                        return remoteObservable.mergeWith(
                                localRepository.getData().takeUntil(remoteObservable)
                        );
                    }
                });
    }
}

RxJava’s publish operator has an overloaded version which takes a selector function as a parameter that can use the multicasted source Observable as many times as needed, without causing multiple subscriptions to it. That means we can publish our remote Observable and merge it with our local Observable which only emits data until the remote Observable emits its data (in which case the local Observable would simply complete).

You can have a look at the following Gist which proves the concept using simple unit tests: https://gist.github.com/jenzz/ae8fe8643e47f431f91712499ef8c4f9