Learning RxJava - Second Edition

Time-based buffering

You can use the buffer(long timespan, TimeUnit unit) operator at fixed time intervals by providing timespan and unit values. To buffer emissions into a list at 1-second intervals, you can run the following code:

import io.reactivex.rxjava3.core.Observable;import java.util.concurrent.TimeUnit;public class Ch7_05 {    public static void main(String[] args) {        Observable.interval(300, TimeUnit.MILLISECONDS)                  .map(i -> (i + 1) * 300)                  .buffer(1, TimeUnit.SECONDS)                  .subscribe(System.out::println);        sleep(4000);    }}

Note that we are making the source emit every 300 milliseconds, and each resulting buffered list will likely contain three or four emissions due to the 1-second interval cutoff. The output will be as follows:

[300, 600, ...