链接两个 Observable 以返回另一个

Question

我有两个名为 A 和 B 的 observable。它们中的每一个都对不同的 REST 服务执行请求，因此它们扩展了一个不同的模型，如上所示。他们中的任何一个执行的请求都可能失败。现在，我需要能够链接它们并返回一个 ModelC 对象。所以，伪编码流应该是这样的：

A 执行请求，如果失败则执行某些操作，否则将其结果（responseModelA）传递给 B，以便它可以执行另一个涉及使用部分 responseModelA 的 REST 请求。如果 B 失败，则发生某些事情，如果没有，则将其响应 (responseModelB) 与 responseModelA（手动设置 POJO 字段）结合起来创建 ModelC，这是订阅者应该在其 call()上作为参数接收的内容> 方法。

这是否可以远程使用 rxJava 进行编码？我非常坚持这一点，所以我愿意接受任何建议。

谢谢。

Answer 1

这假设您已经创建了一个 REST 请求接口，该接口返回类似于以下的 Observables（使用 Retrofit 这很容易）：

interface RestApi {
    Observable<ModelA> getModelA();
    Observable<ModelB> getModelB(int modelBId);
}

class ModelA {
    int modelBId;
    Object fieldA;
}

class ModelB {
    Object fieldB;
}

class ModelC {
    Object fieldFromA;
    Object fieldFromB;

    public ModelC(Object fieldFromA, Object fieldFromB) {
        this.fieldFromA = fieldFromA;
        this.fieldFromB = fieldFromB;
    }
}

要使 ModelB 请求依赖于 ModelA 请求的结果，可以使用 .flatMap 将一个 Observable 的结果转换为另一个 Observable。

然后，要创建 ModelC，请使用 .map 从 ModelA 和 ModelB 中选择所需的字段并返回结果。

RestApi restApi;

Observable<ModelC> observeModelC() {
    return restApi
            .getModelA()
            .flatMap(new Func1<ModelA, Observable<ModelC>>() {
                @Override
                public Observable<ModelC> call(final ModelA modelA) {
                    // Use the modelBId from modelA to get ModelB.
                    return restApi
                            .getModelB(modelA.modelBId)
                            // Combine A & B to create C
                            .map(new Func1<ModelB, ModelC>() {
                                @Override
                                public ModelC call(ModelB modelB) {
                                    return new ModelC(modelA.fieldA, modelB.fieldB);
                                }
                            });
                }
            });
}

您的订阅者将如下所示：

observeModelC()
        .subscribe(new Observer<ModelC>() {
            @Override
            public void onCompleted() {
                // All done.
            }

            @Override
            public void onError(Throwable e) {
                // All errors from any API request will end up here.
                // For Retrofit, cast e to RetrofitError for
                // detailed error info.
            }

            @Override
            public void onNext(ModelC modelC) {
                // Yeah! - Use modelC.
            }
        });

Answer 2

Rx.NET 中的顺序组合是由SelectMany 实现的（我认为它在Rx.Java 中是FlatMap？）。 Rx.NET 也应用了 fail-fast 语义，因此 SelectMany 可能就是你所需要的。

(C#)

IObservable<ModelC> query = A().FlatMap(B, (a, b) => C(a, b));

query.Subscribe(success => DoSomethingSuccessful(), error => DoSomethingElse(error));

A 和 B 是返回 observables 的函数，C 是返回 ModelC 的函数。

但是，如果您想在 A 或 B 失败时执行特定操作，那么 Rx.NET 中的一个快速解决方案是使用 Do 运算符。

IObservable<ModelC> query = A().Do(_ => {}, AFailed)
                               .FlatMap(B.Do(_ => {}, BFailed), (a, b) => C(a, b));

query.Subscribe(success => DoSomethingSuccessful(), UltimateFailure);

其中 AFailed 和 BFailed 是返回 void 的方法，它们接受单个 Exception 参数。

抱歉，我不懂 Java，但也许这会引导您找到正确的解决方案。