java - 如何在 RxJava 中将 observable 包装在另一个 observable 中？

Question

我想将改造 api 调用包装在另一个方法中，我可以在其中另外显示/隐藏加载程序、检查网络等。当我的 api 返回可观察时，我最终的方式如下:

private <T> Observable<T> request(final Observable<T> apiCall, final ViewManager viewManager) {
    return Observable.create(new Action1<Emitter<T>>() {
        @Override
        public void call(final Emitter<T> emitter) {
            if (!NetworkUtils.isConnected(context)) {
                emitter.onError(new ConnectException("network not connected"));
                return;
            }
            viewManager.showLoader();
            apiCall.subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(new Observer<T>() {
                        @Override
                        public void onCompleted() {
                            viewManager.hideLoader();
                            emitter.onCompleted();
                        }

                        @Override
                        public void onError(Throwable e) {
                            viewManager.hideLoader();
                            emitter.onError(e);
                        }

                        @Override
                        public void onNext(T response) {
                            emitter.onNext(response);
                        }
                    });
        }
    }, Emitter.BackpressureMode.BUFFER);
}

这是处理问题的标准方法吗？如何将一个可观察值包装在另一个可观察值中？谁能指导一下吗？

Answer 1

响应式扩展的惯用方式是使用组合，这是 RX 的强大功能之一。

首先让我们使用运算符定义所需的行为，您想要的是这样的:

apiCall
  .observeOn(AndroidSchedulers.mainThread())
  .startWith(Observable.defer(() -> {
      if (!NetworkUtils.isConnected(context)) {
          return Observable.error(new ConnectException("network not connected"));
      } else {
          return Observable.empty();
      }
  }))
  .doOnSubscribe(() -> viewManager.showLoader())
  .doOnCompleted(() -> viewManager.hideLoader())
  .doOnError(throwable -> viewManager.hideLoader());

现在，为了将其组合到任何网络 apiCall Observable，您可以使用 compose() 运算符并将此逻辑封装到 Transformer 中:

class CustomTransformer<T> implements Observable.Transformer<T, T> {

    private final ViewManager viewManager;
    private final Context context;

    CustomTransformer(ViewManager viewManager, Context context) {
        this.viewManager = viewManager;
        this.context = context;
    }

    @Override
    public Observable<T> call(Observable<T> apiCall) {
        return apiCall
                .observeOn(AndroidSchedulers.mainThread())
                .startWith(Observable.defer(() -> {
                    if (!NetworkUtils.isConnected(context)) {
                        return Observable.error(new ConnectException("network not connected"));
                    } else {
                        return Observable.empty();
                    }
                }))
                .doOnSubscribe(() -> viewManager.showLoader())
                .doOnCompleted(() -> viewManager.hideLoader())
                .doOnError(throwable -> viewManager.hideLoader());
        ;
    }
}

然后你可以将它与任何网络Observable组合:

someRetrofitQuery
   .compose(new CustomTransformer<>(viewManager, context))
    ...
   .subscribe();