RxJava：如何處理zip運算符的錯誤？

Question

我正在使用RxJava和RxAndroid與Retrofit2。

Observable<ResponseOne> responseOneObservable = getRetrofitClient().getDataOne() 
.subscribeOn(Schedulers.io()) 
.observeOn(AndroidSchedulers.mainThread()); 

Observable<ResponseTwo> responseTwoObservable = getRetrofitClient().getDataTwo() 
.subscribeOn(Schedulers.io()) 
.observeOn(AndroidSchedulers.mainThread()); 

在上面的兩個Observer上使用如下所示的zip操作符。

Observable<ArrayList<TestData>> testDataObservable = Observable.zip(responseOneObservable, responseTwoObservable, new Func2<ResponseOne, ResponseTwo, ArrayList<TestData>>() { 
      @Override 
       public ArrayList<TestData> call(ResponseOne responseOne, ResponseTwo responseTwo) { 
        ArrayList<TestData> testDataList = new ArrayList(); 
         // Add test data from response responseOne & responseTwo 
        return testDataList; 
      } 
    }) 
    .subscribeOn(Schedulers.io()) 
    .observeOn(AndroidSchedulers.mainThread()) 
    .subscribe(new Subscriber<ArrayList<TestData>>() { 

     @Override 
     public void onNext(ArrayList<TestData> testDataList) { 

     } 

     @Override 
     public void onCompleted() { 
      Log.d(TAG, "onCompleted"); 
     } 

     @Override 
     public void onError(Throwable t) { 
      Log.d(TAG, "onError Throwable: " + t.toString()); 
     } 
    }); 

如果存在任何誤差在responseOneObservable和responseTwoObservable改型HTTP調用期間發生那麼它會直接調用的testDataObservable訂戶的onError方法。

即使兩個observable中的任何一個獲得了成功響應，我也想繼續使用zip運算符的call方法。

如何使用zip運算符處理錯誤響應？

Answer 1

您可以使用onErrorResumeNext返回一些觀察或onErrorReturn一些默認值返回zip，如：

Observable.zip(
    responseOneObservable 
     .onErrorReturn(new Func1<Throwable, ResponseOne>() { 
     @Override 
     public ResponseOne call(final Throwable throwable) { 
      return new ResponseOne(); 
     } 
    }), 
    responseTwoObservable 
     .onErrorReturn(new Func1<Throwable, ResponseTwo>() { 
     @Override 
     public ResponseTwo call(final Throwable throwable) { 
      return new ResponseTwo(); 
     } 
    }), 
    ... 

更多信息，請參見onError handling。

Answer 2

您應該使用onErrorResumeNext來覆蓋單個壓縮的觀察值，以指示他們在發生錯誤時發出默認項目。

參考Error-Handling-Operators

Answer 3

您可以返回從以下兩種觀測的默認響應與onErrorResumeNext運營商。

Observable<ResponseOne> responseOneObservable = getRetrofitClient().getDataOne() 
    .onErrorResumeNext(throwable -> {/*some default value*/}) 
    .subscribeOn(Schedulers.io()) 
    .observeOn(AndroidSchedulers.mainThread()); 

Observable<ResponseTwo> responseTwoObservable = getRetrofitClient().getDataTwo() 
    .onErrorResumeNext(throwable -> {/*some default value*/}) 
    .subscribeOn(Schedulers.io()) 
    .observeOn(AndroidSchedulers.mainThread()); 

另見Error handling in RxJava - Dan Lew