RxJava 将两种不同类型的 observable 组合成第三种类型的 observable

Question

我有两个服务器端点。 端点 1 接受用户名和密码并返回一个 id。 端点 2 采用 that 并返回一个 UserProfile 对象。

我正在尝试编写一个将用户名和密码作为参数并返回 Observable<User> 的方法（其中 User 是一个对象，其中一个字段是端点 1 的 id，另一个字段是 UserProfile） .

我已经弄清楚如何调用我的第一个端点并将生成的 id 作为第二个端点的输入，但是我一直坚持如何使用来自第二个端点的响应和来自第一个端点的响应创建 User 对象并返回它。

这是不正确的，但这是我能够进步的最远：

    public Observable<UserProfile> login(String u, String p) {
        return mNetworkApi.establishSessionViaCreds(u, p)
            .flatMap(s -> mNetworkApi.getUserProfile(s));
    }

这个方法接受两个字符串并输出一个Observable<UserProfile>。我需要它来输出一个 Observable。可以通过获取第一个 observable（String）的输出和第二个 observable（UserProfile）的输出来创建用户。但是我该怎么做呢？下面是我想要的方法签名，但没有正确的代码来创建 Observable。

public Observable<User> login(String username, String password) {
    mNetworkApi.doLogin(username, password)
        flatMap((Func1<String, Observable<?>>) s -> mNetworkApi.getUserProfile(s));
    /..Code to transform observable.../
    return ...;
}

有没有一种方法可以从具有两种不同类型（字符串和用户配置文件）的可观察对象中创建具有第三种类型（用户）的可观察对象？

编辑 为了解释我遇到的问题更好。第一个 observable 返回一个字符串。该字符串用作下一个 observable 的输入，但是在第二个 observable 运行后，我需要从第一个 observable 访问该字符串，并且无法再获得对它的引用。

编辑 这是我能得到的最接近的结果，但由于“id”无法访问，它仍然无法工作。如果我可以获得对 id 的引用，那么这将完美运行，但我调用 new User(id, userProfile)，id 变量是不可访问的。

    @Override
public Observable<User> loginUser(String username, String password) {
    Observable<String> observableUsername = Observable.just(username);
    Observable<String> observablePassword = Observable.just(password);

    Observable<User> userObs = Observable.zip(observableUsername, observablePassword,
            (user, pass) ->
                mNetworkApi.establishSessionViaCreds(user, pass)
                    .flatMap(id -> mNetworkApi.getUserProfile(id))
                        .map(userProfile -> new User(id, userProfile))
    ).flatMap(new Func1<User, Observable<?>>() {
        @Override
        public Observable<?> call(User user) {
            return Observable.just(new User(id, userProfile));
        }
    });

    return userObs;
}

编辑

Observable<User> userObs = Observable.zip(observableUsername, observablePassword,
            (user, pass) ->
                    mNetworkApi.establishSessionViaCreds(user, pass)
                            .flatMap(id -> mNetworkApi.getUserProfile(id)
                                    .map(new Func1<UserProfile, User>() {
                                        @Override
                                        public User call(UserProfile userProfile) {
                                            return Observable.just(new User(id, userProfile));
                                        }
                                    })
                            )
    );

Answer 1

有一个 flatMap 的覆盖，它允许您指定 Func2 以合并 Func1 的原始输入和输出。对于您的情况，它看起来像：

public Observable<User> login(String u, String p) {

    return networkApi
            .establishSessionViaCreds(u, p)
            .flatMap(
                    // here is the normal Func1
                    new Func1<String, Observable<UserProfile>>() {
                        @Override
                        public Observable<UserProfile> call(String id) {
                            return networkApi.getUserProfile(id);
                        }
                    },
                    // here is the extra Func2 which allows the merge
                    new Func2<String, UserProfile, User>() {
                        @Override
                        public User call(String id, UserProfile userProfile) {
                            return new User(id, userProfile);
                        }
                    }
            );
}

或者，如果您使用的是 lambda：

public Observable<User> login(String u, String p) {

    return networkApi
            .establishSessionViaCreds(u, p)
            .flatMap(
                    id -> { return networkApi.getUserProfile(id); },
                    User::new
            );
}

Answer 2

你需要Observable.zip:

Observable<String> username = ...;
Observable<String> password = ...;

Observable<User> userObs = Observable.zip(username, password,
  (user,pass) ->
    mNetworkApi.doLogin(user, pass)
               .flatMap(id -> mNetworkApi.getUserProfile(id))
)
.flatMap(obs -> obs);

请注意，最后一行是必需的，因为 zip 是 A + B -> C，但在本例中，我们使用的是 A + B -> Observable<C>，我们需要解压缩生成的 Observable。