RxJava流：有條件的運算符和錯誤處理

Question

我是RxJava的新手，並嘗試將我的頭包裹在一個更復雜的登錄邏輯中，該登錄邏輯有三個異步方法進行處理。對我來說，這是「如果我得到這個東西轉化爲RxJava，任何東西（tm）是可能的」 :)

所以我想要做的是以下幾點：

Call A -> (Process A) -> Call B with results of A -> (Process B) -\ 
        \            -> Combine and Subscribe 
        \-> Call C with results of A -> (Process C) -/ 

現在的問題是， Call C分支只能在特定條件下執行，否則不能（Combine and Subscribe然後可以從該分支接收到NULL值，這是可以的）。

而且，錯誤處理是不平凡：雖然Call A和Call C（如果運行的話）需要通過onError提供他們的錯誤給最終用戶，Call B的‘成功’是寧可選的，可以忽略不計在失敗的情況下。

這是我想出了這麼遠，而忽略了「C」分支還是：

mApi.callA(someArgs) 
      // a transition operator to apply thread schedulers 
      .compose(applySchedulers()) 
      // from rxlifecycle-components, unsubscribes automatically 
      // when the activity goes down 
      .compose(bindToLifecycle()) 
      // possibly other transformations that should work on the (error) 
      // states of the first and the following, chained API calls 
      .flatMap(response -> processA(response)) 
      .flatMap(response -> mApi.callB(response.token)) 
      .flatMap(response -> processB(response)) 
      // redirects HTTP responses >= 300 to onError() 
      .lift(new SequenceOperators.HttpErrorOperator<>()) 
      // checks for application-specific error payload and redirects that to onError() 
      .lift(new SequenceOperators.ApiErrorOperator<>()) 
      .subscribe(this::allDone this::failure); 

我看了看周圍的Wiki for conditional operators，但我不能找到一個提示如何開球Call C科。另外，我不確定我的SequenceOperators是否可以按照這種方式工作，也就是說可以放在鏈中的所有請求之後，還是我需要其中的幾個，每個放置在觸發新的flatMap（）運算符後的中間Call。

有人能指出我正確的方向嗎？

謝謝！

Answer 1

您應該使用Zip操作:)的結果應該是這樣的：

mApi.callA(someArgs) 
     // ... 
     .flatMap(response -> processA(response)) 
     .flatMap(response -> { 
       return Observable.zip(
        callB(response), 
        callC(response), 
        (rA,rB) -> { 
          // or just return a new Pair<>(rA, rB) 
          return combineAPlusB(rA,rB) 
        } 
      ) 
     }) 
     // ... 
     .subscribe(this::allDone this::failure);