Learning RxJava - Second Edition

Fixed-size buffering

The simplest overload for buffer() accepts a count argument as the buffer size and groups emissions in the batches of the specified size. If we wanted to batch up emissions into lists of eight elements, we can do that as follows:

import io.reactivex.rxjava3.core.Observable;public class Ch7_01 {    public static void main(String[] args) {        Observable.range(1, 50)                  .buffer(8)                  .subscribe(System.out::println);    }}

The output is as follows:

[1, 2, 3, 4, 5, 6, 7, 8][9, 10, 11, 12, 13, 14, 15, 16][17, 18, 19, 20, 21, 22, 23, 24][25, 26, 27, 28, 29, 30, 31, 32][33, 34, 35, 36, 37, 38, 39, 40][41, 42, 43, 44, 45, 46, 47, 48][49, 50]

As you can see, if the buffer size does not cleanly divide the total number of emissions, the remaining ...