Taking first element of a GroupedObservable with RxJava

Question

I'm pretty new to RxJava.

I'm trying to group elements by a given key and take the first value of each key.

For instance, group these names by their first letter and print the first one for each group:

Observable<String> names = Observable.just(
                             "John", "Steve", "Ruth",
                             "Sam", "Jane", "James");

Here's what I tried so far :

1st try:

names.groupBy(s -> s.charAt(0))
     .flatMap(grp -> grp.first())
     .subscribe(s -> System.out.println(s));

FAILED!: all the names are printed.

2nd try:

names.groupBy(s -> s.charAt(0))
     .subscribe(grp -> grp
                         .first()
                         .subscribe(System.out::println));

FAILED AGAIN!: all the names are still printed...

I managed to get the expected result by reducing each GroupedObservable :

1st try, with reduce:

names.groupBy(s -> s.charAt(0))
     .flatMap(grp -> grp.reduce((a, b) -> a))
     .subscribe(s -> System.out.println(s));

2nd try, with reduce:

names.groupBy(s -> s.charAt(0))
     .subscribe(grp -> grp
                         .reduce((a, b) -> a)
                         .subscribe(System.out::println));

But this solution is not ideal. In a real world situation, this would involve some expensive computation and the reduce() operator would take a lot longer to compute (computing all the values) than the first() operator (computing only 1 value by key).

Answer 1

This is due to the property of groupBy: if a group gets cancelled (no longer observed), a new group with the same key may get started. Here first unsubscribes after the first element and the subsequent element with the same key recreates the group.

You can prevent this via some publish trickery:

Observable<String> names = Observable.just(
        "John", "Steve", "Ruth",
        "Sam", "Jane", "James");

names.groupBy(s -> s.charAt(0))
.flatMap(grp -> grp.publish(o -> o.first().concatWith(o.ignoreElements())))
.subscribe(s -> System.out.println(s));

Here, I share the group and create a special output; a concatenation of the first element with the end of the whole group (ignoring elements).