Reactor Core implements the Reactive Streams specification and deals with handling a (potentially unlimited) stream of data. If it interests you, do check out the excellent documentation it offers. Here I am assuming some basic familiarity with the Reactor Core libraries Flux and Mono types and will cover Reactor Core provides an abstraction to time itself to enable testing of functions which depend on passage of time.
For certain operators of Reactor-core, time is an important consideration – for eg, a variation of “interval” function which emits an increasing number every 5 seconds after an initial “delay” of 10 seconds:
val flux = Flux
.interval(Duration.ofSeconds(10), Duration.ofSeconds(5))
.take(3)Testing such a stream of data depending on normal passage of time would be terrible, such a test would take about 20 seconds to finish.
Reactor-Core provides a solution, an abstraction to time itself – Virtual time based Scheduler, that provides a neat way to test these kinds of operations in a deterministic way.
Let me show it in two ways, an explicit way which should make the actions of Virtual time based scheduler very clear followed by the recommended approach of testing with Reactor Core.
import org.assertj.core.api.Assertions.assertThat
import org.junit.Test
import reactor.core.publisher.Flux
import reactor.test.scheduler.VirtualTimeScheduler
import java.time.Duration
import java.util.concurrent.CountDownLatch
class VirtualTimeTest {
@Test
fun testExplicit() {
val mutableList = mutableListOf<Long>()
val scheduler = VirtualTimeScheduler.getOrSet()
val flux = Flux
.interval(Duration.ofSeconds(10), Duration.ofSeconds(5), scheduler)
.take(3)
val latch = CountDownLatch(1)
flux.subscribe({ l -> mutableList.add(l) }, { _ -> }, { latch.countDown() })
scheduler.advanceTimeBy(Duration.ofSeconds(10))
assertThat(mutableList).containsExactly(0L)
scheduler.advanceTimeBy(Duration.ofSeconds(5))
assertThat(mutableList).containsExactly(0L, 1L)
scheduler.advanceTimeBy(Duration.ofSeconds(5))
assertThat(mutableList).containsExactly(0L, 1L, 2L)
latch.await()
}
}1. First the scheduler for “Flux.interval” function is being set to be the Virtual Time based Scheduler.
2. The stream of data is expected to be emitted every 5 seconds after a 10 second delay
3. VirtualTimeScheduler provides an “advanceTimeBy” method to advance the Virtual time by a Duration, so the time is being first advanced by the delay time of 10 seconds at which point the first element(0) is expected to be emitted
4. Then it is subsequently advanced by 5 seconds twice to get 1 and 2 respectively.
This is deterministic and the test completes quickly. This version of the test is ugly though, it uses a list to collect and assert the results on and a CountDownLatch to control when the test terminates. A far cleaner approach for testing Reactor-Core types is using the excellent StepVerifier class and a test which makes use of this class looks like this:
import org.junit.Test
import reactor.core.publisher.Flux
import reactor.test.StepVerifier
import reactor.test.scheduler.VirtualTimeScheduler
import java.time.Duration
class VirtualTimeTest {
@Test
fun testWithStepVerifier() {
VirtualTimeScheduler.getOrSet()
val flux = Flux
.interval(Duration.ofSeconds(10), Duration.ofSeconds(5))
.take(3)
StepVerifier.withVirtualTime({ flux })
.expectSubscription()
.thenAwait(Duration.ofSeconds(10))
.expectNext(0)
.thenAwait(Duration.ofSeconds(5))
.expectNext(1)
.thenAwait(Duration.ofSeconds(5))
.expectNext(2)
.verifyComplete()
}
}This new test with StepVerifier reads well with each step advancing time and asserting on what is expected at that point.
| Published on Java Code Geeks with permission by Biju Kunjummen, partner at our JCG program. See the original article here: Testing time based reactor core streams with Virtual time Opinions expressed by Java Code Geeks contributors are their own. |
