RxJava: prevent an observable from emitting until data from another observable is emitted

Question

The block of code below is designed to be offline-first. If data is emitted by the memory observable the local and remote observable will never fire. If data is not held in memory the local observable will attempt to read data from room database and if all else fails then the remote observable queries an API.

The remote source uses retrofit to send queries and returns a flowable which is then converted into an observable. Before the remote observable fires, however, I have another observable that returns location data needed by the query. In other words, the remote observable is dependent on the location observable. How do I can I use RxJava to prevent the remote observable from being called in the Concat operator until the location data is available?

locationObservable = locationSource.getLocationObservable();
memory = source.getSuggestionsFromMemory();
local = source.getSuggestionsFromDisk();


remote = source.getSuggestionsFromNetwork(parameters)
                    .skipUntil(locationObservable);

locationObservable.subscribe(
                    source -> parameters = ParamManager.queryParameters(
                                    source.getLatitude() + "," + source.getLongitude()),

                    error -> Log.println(Log.ERROR, TAG, error.getMessage()
                    )
            );

Observable.concat(memory,local, remote)
            .firstElement()
            .subscribeOn(Schedulers.io())
            .toObservable()
            .observeOn(AndroidSchedulers.mainThread());

the remote observable:

public Observable<List<Venue>> getSuggestionsFromNetwork(HashMap<String, String> parameters){
    return remoteSource.getData(parameters).doOnNext(
            data -> {
                localSource.cacheDataToDisk(data);
                memorySource.cacheDataInMemory(data);
            });
}

remote source:

Observable<List<Venue>> getData(HashMap<String, String> params){
    return Flowable.zip(loadSearchVenues(params), loadTrendingVenues(params),
            loadRecommendedVenues(params), (search, trending, recommended) -> {

                generalVenues = search.getResponse().getSuggestions();
                trendingVenues = trending.getResponse().getSuggestions();
                recommendedVenues = recommended.getResponse().getSuggestions();

                allVenues.addAll(generalVenues);
                allVenues.addAll(trendingVenues);
                allVenues.addAll(recommendedVenues);

                return allVenues;
            }).toObservable();
}

error:

2019-11-15 09:18:08.703 29428-29491/com.example.suggest E/MemorySource: getData() called
2019-11-15 09:18:08.703 29428-29491/com.example.suggest E/LocalSource: getData() called
2019-11-15 09:18:08.767 29428-29428/com.example.suggest E/MainViewModel: Query map was null (parameter #3)

you asked the question really bad :) try to explain better what you want to achieve — Daniele Segato
If I understand your question correctly you want to fetch the data from memory first and if it's not there from Database and if it fails there also fetch from API — Mightian
so you want to read local data, if cache miss go online, always (aka = you don't care about stale content), correct? — Daniele Segato
yes, that is what the code does above. if there is not any data available the composite observables will call onComplete without emitting data. However, in order for the remote observable to query the API, it needs the devices' longitude and latitude.I was trying to figure out is there a way to prevent the remote observable from being called until location data is available. — onStackOverflowListener

bubbles bubbles · Accepted Answer · 2019-11-16T14:16:44

If you can wait until the next location emission you can do the following :

locationObservable = locationSource.getLocationObservable();
memory = source.getSuggestionsFromMemory();
local = source.getSuggestionsFromDisk();
remote = locationObservable
           .map(source -> ParamManager.queryParameters(source.getLatitude() + "," 
                     + source.getLongitude()))
           .concatMap(params -> source.getSuggestionsFromNetwork(params));

Observable
  .concat(memory,local, remote)
  .firstElement()
...

But if you cannot you have to store the last location in a variable that can be used directly, something like :

remote = Optional.ofNullable(getLastLocation())
           .map(Observable::just)
           .orElse(locationObservable)
           .map(source -> ParamManager.queryParameters(source.getLatitude() + "," 
                     + source.getLongitude()))
           .concatMap(params -> source.getSuggestionsFromNetwork(params));

And elsewhere :

locationObservable.subscribe(location -> setLastLocation(location));

RxJava: prevent an observable from emitting until data from another observable is emitted

1 Answers