CODEWITHSUNDEEP
rx-javarx-android
onStackOverflowListener
onStackOverflowListener

Reputation: 186

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

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)

Upvotes: 1

Views: 777

Answers (1)

bubbles
bubbles

Reputation: 2717

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

Upvotes: 1

Related Questions