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