下面的代码块即将下线。如果数据由内存可观察对象发出，则本地和远程可观察对象将永远不会触发。如果数据未保存在内存中，则本地可观察对象将尝试从房间数据库读取数据，如果所有其他操作都失败，则远程可观察对象将查询API。
远程源使用改造来发送查询并返回一个可流动的，然后将其转换为可观察的。然而，在远程可观察对象触发之前，我有另一个可观察对象返回查询所需的位置数据。换句话说，远程可观察对象依赖于位置可观察对象。如何使用RxJava来防止在Concat运算符中调用远程可观察对象，直到位置数据可用？
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());
远程可观察：
public Observable<List<Venue>> getSuggestionsFromNetwork(HashMap<String, String> parameters){
return remoteSource.getData(parameters).doOnNext(
data -> {
localSource.cacheDataToDisk(data);
memorySource.cacheDataInMemory(data);
});
}
远程源：
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();
}
错误：
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)
如果您可以等到下一个位置发射，您可以执行以下操作：
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()
...
但是如果不能，则必须将最后一个位置存储在可以直接使用的变量中，例如：
remote = Optional.ofNullable(getLastLocation())
.map(Observable::just)
.orElse(locationObservable)
.map(source -> ParamManager.queryParameters(source.getLatitude() + ","
+ source.getLongitude()))
.concatMap(params -> source.getSuggestionsFromNetwork(params));
和其他地方：
locationObservable.subscribe(location -> setLastLocation(location));
