How to merge several objects into a single one after execute the groupBy operation?

Question

I have the class below that I've created to illustrate my doubt.

After doing the initial transformations on my flowable I have:

UserScoreTO{id=1, name='john', score=4}
UserScoreTO{id=1, name='john', score=5}
UserScoreTO{id=1, name='john', score=1}
UserScoreTO{id=2, name='paul', score=4}
UserScoreTO{id=2, name='paul', score=2}
UserScoreTO{id=3, name='mark', score=1}
UserScoreTO{id=3, name='mark', score=7}

I want to combine UserScoreTO objects with the same id into a Flowable that emits one single object for each group, that contains the user information and the sum of the scores.

So the result will be flowable that emits:

User (1, "john", 10);
User (2, "paul", 6);
User (3, "mark", 8);

I want to do this with RxJava if possible (I know I could achieve same result doing something with HashMaps).

package examples.rxjava;

import java.util.Arrays;
import java.util.List;

import io.reactivex.Flowable;

import static java.lang.System.out;

public class TestUsers {

    public static void main(String[] args) {
        new TestUsers().execute();
    }


    public void execute() {
        getUsers()
                .flatMap(list -> Flowable.fromIterable(list))
                .groupBy(userScoreTO -> userScoreTO.id).subscribe(group -> group.subscribe(out::println));

    }





    Flowable<List<UserScoreTO>> getUsers() {
        return Flowable.fromCallable(
                () -> Arrays.asList(
                        new UserScoreTO(1, "john", 4),
                        new UserScoreTO(1, "john", 5),
                        new UserScoreTO(1, "john", 1),

                        new UserScoreTO(2, "paul", 4),
                        new UserScoreTO(2, "paul", 2),

                        new UserScoreTO(3, "mark", 1),
                        new UserScoreTO(3, "mark", 7))
        );


    }

    private class User {
        private int id;
        private String name;
        private int totalScore;

        public User(int id, String name, int totalScore) {
            this.id = id;
            this.name = name;
            this.totalScore = totalScore;
        }

        @Override
        public String toString() {
            return "User{" +
                    "id=" + id +
                    ", name='" + name + '\'' +
                    ", totalScore=" + totalScore +
                    '}';
        }
    }

    private class UserScoreTO {
        private int id;
        private String name;
        private int score;


        public UserScoreTO(int id, String name, int score) {
            this.id = id;
            this.name = name;
            this.score = score;
        }

        @Override
        public String toString() {
            return "UserScoreTO{" +
                    "id=" + id +
                    ", name='" + name + '\'' +
                    ", score=" + score +
                    '}';
        }
    }

}

Answer 1

The code you provided already does exactly what you want. groupBy will group UserScoreTO objects with the same id into the same GroupedFlowable object.

Edit:

Okay I see, maybe the following does what you want a little bit better

  public void execute() {
getUsers()
    .flatMap(Flowable::fromIterable)
    .groupBy(userScoreTO -> userScoreTO.id)
    .map(group -> group.reduce(new User(group.getKey(), "", 0),
        (user, userScoreTO) -> {
          user.name = userScoreTO.name;
          user.totalScore += userScoreTO.score;
          return user;
        }))
    .subscribe(userSingle -> userSingle.subscribe(System.out::println));

}

What you want is to aggregate the UserScoreTo into the same User object by applying the reduce operator.

Answer 2

After doing some research and doing a lot of trial-and-error code, I found this [collect][1] operator that did the trick:

It creates a new User object (User::new) as an initial supplier that holds data from all items from a group. So I set the properties and increment the score, returning it at the end of the group iteration.

As I'm new to RxJava I don't know if this is the best approach or if the code can be simplified, but the resulting code is:

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

                }
            });

        });

The output:

onSubscribe
onSubscribe
onSubscribe
User{id=1, name='john', totalScore=10}
User{id=2, name='paul', totalScore=6}
User{id=3, name='mark', totalScore=8}