Testing Time-Based Reactor Core Streams With Virtual Time

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 how Reactor Core provides an abstraction to time itself to enable the testing of functions that depend on passage of time.

For certain operators of Reactor Core, time is an important consideration — for example, a variation of an "interval" function that 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 the 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 — its virtual time-based scheduler, which provides a neat way to test these kinds of operations in a deterministic way.

Let me show it off in two ways — an explicit way that should make the actions of the 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 the "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 that 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.