如何在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); 
} 

这是一个处理问题的标准方式？你如何在另一个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可观的，你可以使用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();