java - 如何以 RxJava2 的方式组织异步调用链

Question

我将下一个代码封装在 AsyncTask 类中

        AsyncTask<Void, Void, Void> task = new AsyncTask<Void, Void, Void>() {
        @Override
        protected Void doInBackground(Void... params) {
            try {
                if (NetworkUtils.isNetworkConnected(mContext))
                    mIndicatorTable.sync().blockingGet();
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            } finally {
                try {
                    final MobileServiceList<IndicatorModel> results = mIndicatorTable.table().read(null).blockingGet();
                    if (isViewAttached()) {
                        getMvpView().refreshIndicatorList(results);
                        getMvpView().hideLoading();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            return null;
        };

mIndicatorTable.sync()和 mIndicatorTable.table().read(null)均返回 Single<Boolean> . mIndicatorTable.sync() - 将本地存储与远程存储同步，如果网络不可用，那么我们只需使用 mIndicatorTable.table().read(null) 从本地存储读取当网络准备就绪时 - 我们执行同步，然后从本地存储读取(我们不关心同步是否取消或中断)。毕竟我们应该调用View来刷新RecycleView。如何用 RxJava2 实现？

#UPDATE

工作版本

getDataManager().syncAll()
     .onErrorResumeNext(Single.just(false))
     .flatMap(list -> toSingle(mIndicatorTable.getTable().read(null)))
     .subscribeOn(Schedulers.newThread())
     .observeOn(AndroidSchedulers.mainThread())
     .subscribe(results -> {
        if (isViewAttached()) {
            getMvpView().refreshIndicatorList(results);
            getMvpView().hideLoading();
        }
     })

Answer 1

rx-java 只需调用 .flatMap() 即可轻松链接异步调用.

mIndicatorTable.sync()
        .flatMap(new Function<Boolean, SingleSource<MobileServiceList<IndicatorModel>>>() {
            @Override
            public SingleSource<MobileServiceList<IndicatorModel>> apply(Boolean aBoolean) throws Exception {
                return mIndicatorTable.table().read(null);
            }
        })
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(...)

但是你想在另一个线程中继续调用 sync() 并且 RecycleView 只能从主线程访问。这就是 rx-android 派上用场的地方。 .subscribeOn(Schedulers.io()) 在另一个线程中实例化流，.observeOn(AndroidSchedulers.mainThread()) 确保这一行下面的语句(这里，.subscribe(...)) 在主线程中执行。