java - Android，Java Rx，异步

我有一个显示项目列表的屏幕。


首先，我想显示本地数据库中的元素。
用户查看本地元素后，检查互联网并
更新它们（如果设备具有互联网）。


我有以下代码片段

Observable<Data> fetchFromdisk = ...;
Observable<Data> fetchFromNetwork = ...;

Observable<Data>  fetchData= Observable.concat (fetchFromdisk ,fetchFromNetwork);

然后我正在做以下

mSubscriptions.clear();
Subscription subscription = fetchData
        .subscribeOn(mSchedulerProvider.computation())
        .observeOn(mSchedulerProvider.ui())
        .subscribe(
                new Subscriber<List<Data>>() {
                    @Override
                    public void onCompleted() {
                        view.setLoadingIndicator(false);
                    }

                    @Override
                    public void onError(Throwable e) {
                        processError(e);
                    }

                    @Override
                    public void onNext(List<Data> data) {
                        processData(data);
                    }
                }

        );
mSubscriptions.add(subscription);

但是正在发生的事情是先运行fetchFromdisk，然后执行FetchFromNetwork。发射元素结束后，将开始执行ProcessData。

这与我要实现的目标完全相反。
我想让用户优先查看快速数据，然后再访问互联网。
我究竟做错了什么？

最佳答案

一旦提取了缓存，就可以调用方法.doOnCompleted来执行代码。

mSubscriptions.clear();
Subscription subscription = fetchData
    .subscribeOn(mSchedulerProvider.computation())
    .doOnCompleted( new Action0<List<Notification>>() {
                @Override
                public void call(List<Data> cachedData) {
                      // Get data from server here.

                      // Compare new data against cached data.

                      // Update cache.

                      // Process new data.
                }
            })
    .observeOn(mSchedulerProvider.ui())
    .subscribe(
            new Subscriber<List<Data>>() {
                @Override
                public void onCompleted() {
                    view.setLoadingIndicator(false);
                }

                @Override
                public void onError(Throwable e) {
                    processError(e);
                }

                @Override
                public void onNext(List<Data> data) {
                    processData(data);
                }
            }

    );
    mSubscriptions.add(subscription);

/////////非解决方案提示////////

尝试使用Single而不是Observable，它可以摆脱onCompleted。
https://github.com/lettuce-io/lettuce-core/issues/139