RxJava Android -在适当的线程上加载缓存显示数据

Question

我正在探索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()永远不会被调用。组合这些可观察对象以实现我所拥有的方案的正确方法是什么?

Answer 1

用onErrorResumeNext(Observable.just(itemsDao.getCachedItems()))代替onErrorReturn()解决