Hands-On Design Patterns with Kotlin

Batching

Sometimes, slowing the producer is not possible. So, are we back to the original problem, of either dropping some events or running out of memory? Luckily, Rx still has a few tricks up its sleeve. It is often more efficient to process data in batches. We've already discussed such a case in the previous chapter. For that, we can specify buffer() for our subseriber.

Buffer has three flavors. The first one is batch-per-size:

val latch = CountDownLatch(1)val o = Observable.intervalRange(8L, 15L, 0L, 100L, TimeUnit.MILLISECONDS)o.buffer(3).subscribe({    println(it)}, {}, { latch.countDown()})latch.await()

It outputs the following:

[8, 9, 10][11, 12, 13][14, 15, 16][17, 18, 19][20, 21, 22]

The second is the batch-per-time interval. Imagine ...