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