Grouping is a powerful operation that can be achieved using RxKotlin. This operation allows you to group emissions based on their property. Say, for example, you have an Observable / Flowable emitting integer numbers (Int), and, as per your business logic, you have some separate code for even and odd numbers and want to handle them separately. Grouping is the best solution in that scenario.

Let's take an example:

    fun main(args: Array<String>) { 
      val observable = Observable.range(1,30) 
 
      observable.groupBy {//(1) 
        it%5 
      }.blockingSubscribe {//(2) 
        println("Key ${it.key} ") 
        it.subscribe {//(3) 
            println("Received $it") 
        } 
      } 
    } 

In this example, I've grouped emissions based on their remainder when divided by 5, so, basically, there should be 5 groups (0 through 4). On comment (1) of this example, we used the groupBy operator and passed a predicate to it, upon which the grouping should be performed. The groupBy operator takes the result of the predicate to group emissions.

On comment (2) of this example, we used the blockingSubscribe operator to subscribe to the newly created Observable<GroupedObservable<K, T>> instance. We could also use the simple subscribe operator; however, as we are printing the output to the console, by using subscribe, everything will look like a mess. Mainly because the subscribe operator doesn't wait for the given task on emission to complete before taking the next emission. On the other hand, blockingSubscribe will make the program wait until it completes processing an emission, before proceeding to a new one.

The groupBy operator returns Observable that emits GroupedObservable, containing our groups; so, inside blockingSubscribe, we need to subscribe to the emitted GroupedObservable instance. On comment (3), we did the same, after printing the key of the emitted GroupedObservable instance.

The output is as follows: