将 Single's<User> 的 Flowable 转换为 List<User> 的 Flowable

Question

我有这个 Rx 流，用于将 UserScoreTO 列表分组到单个用户对象（多对一关系）中。

public void execute() {
    getUsers()
            .flatMap(list -> Flowable.fromIterable(list))
            .groupBy(userScoreTO -> userScoreTO.id)
            .flatMap(groups -> Flowable.fromCallable(() -> groups.collect(User::new, (user, userscore) -> {
                user.id = userscore.id;
                user.name = userscore.name;
                user.totalScore += userscore.score;
            }))).subscribe(userSingle -> userSingle.subscribe(new SingleObserver<User>() {
        @Override
        public void onSubscribe(Disposable d) {
            System.out.println("onSubscribe");
        }

        @Override
        public void onSuccess(User user) {
            System.out.println(user);
        }

        @Override
        public void onError(Throwable e) {
            System.out.println(e);

        }
    }));

}

如你所见，为了消费这个 flowable，我订阅了它，消费了它发出的一些 Single's。所以我也订阅了那张单曲。虽然这甚至可以工作，但有点烦人......我只想订阅一个并消费一组用户......

几天前我问了另一个问题about this same code。该类的完整代码在那里。

Answer 1

将flatMap 放在末尾可以消除对嵌套subscribe 的需要。

例子：

getUsers()
    .flatMap(list -> Flowable.fromIterable(list))
    .groupBy(userScoreTO -> userScoreTO.id)
    .flatMap(groups -> 
        Flowable.fromCallable(() -> 
            groups.collect(User::new, (user, userscore) -> {
                user.id = userscore.id;
                user.name = userscore.name;
                user.totalScore += userscore.score;
            }
    )))
    .flatMap(it -> it.toFlowable()) // <-- unwrap the singles
    .subscribe(user -> System.out.println(user));