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

Question

我想用另一种方法包装改造 api 调用，我可以在其中另外 show/hide 加载程序、检查网络等。作为我的 api returns 可观察到的，我结束的方式如下：

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);
}

这是处理问题的标准方法吗？如何将一个 Observable 包裹在另一个 Observable 中？谁能指导一下？

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();

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

How to wrap observable inside another observable in RxJava?

java

android

observable

rx-java

rx-android