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的条件运算符,但我找不到如何启动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);