我正在探索 RxJava 及其对 Android 的适用性，并且我正在尝试实现一个简单的 load-cache-display 用例，如下面的 ASCII 图表所示：
---------------
--- failure --| Load data |-- success ---
| --------------- |
V V
------------------- -------------
| Get from cache | | Filter |
------------------- -------------
| |
| V
| ---------------- -------------
------------>| Display |<------| Cache |
---------------- -------------
这是我最初想出的代码：
subscription = AndroidObservable.bindFragment(this, restClient.getItems())
.onErrorReturn(new Func1<Throwable, List<Item>>() {
@Override public List<Item> call(Throwable throwable) {
return itemsDao.getCachedItems();
}
})
.flatMap(new Func1<ItemContainer, Observable<Item>>() {
@Override public Observable<Item> call(ItemContainer itemContainer) {
return Observable.from(itemContainer.getItems());
}
})
.filter(new Func1<Item, Boolean>() {
@Override public Boolean call(Item item) {
return item.getName().startsWith("B");
}
})
.toList()
.map(new Func1<List<Item>, List<Item>>() {
@Override public List<Item> call(List<Item> items) {
itemsDao.cacheItems(items);
return items;
}
})
.subscribeOn(Schedulers.from(AsyncTask.THREAD_POOL_EXECUTOR))
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<List<Item>>() {
@Override public void call(List<Item> items) {
displayData(items);
}
});
正如预期的那样，网络和缓存是在后台线程上执行的，而数据的显示则是在 UI 线程上进行的。问题是返回的数据onErrorReturn()经过相同的过滤和缓存循环，这是多余的。但是，如果我将代码更改为：
subscription = AndroidObservable.bindFragment(this, restClient.getItems())
.flatMap(new Func1<ItemContainer, Observable<Item>>() {
@Override public Observable<Item> call(ItemContainer itemContainer) {
return Observable.from(itemContainer.getItems());
}
})
.filter(new Func1<Item, Boolean>() {
@Override public Boolean call(Item item) {
return item.getName().startsWith("B");
}
})
.toList()
.map(new Func1<List<Item>, List<Item>>() {
@Override public List<Item> call(List<Item> items) {
itemsDao.cacheItems(items);
return items;
}
})
.onErrorReturn(new Func1<Throwable, List<Item>>() {
@Override public List<Item> call(Throwable throwable) {
return itemsDao.getCachedItems();
}
})
.subscribeOn(Schedulers.from(AsyncTask.THREAD_POOL_EXECUTOR))
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Action1<List<Item>>() {
@Override public void call(List<Item> items) {
displayData(items);
}
});
displayData()永远不会被调用。组成这些可观察对象以实施我所拥有的方案的正确方法是什么？