DZone
Thanks for visiting DZone today,
Edit Profile
  • Manage Email Subscriptions
  • How to Post to DZone
  • Article Submission Guidelines
Sign Out View Profile
  • Post an Article
  • Manage My Drafts
Over 2 million developers have joined DZone.
Log In / Join
Refcards Trend Reports
Events Video Library
Refcards
Trend Reports

Events

View Events Video Library

Zones

Culture and Methodologies Agile Career Development Methodologies Team Management
Data Engineering AI/ML Big Data Data Databases IoT
Software Design and Architecture Cloud Architecture Containers Integration Microservices Performance Security
Coding Frameworks Java JavaScript Languages Tools
Testing, Deployment, and Maintenance Deployment DevOps and CI/CD Maintenance Monitoring and Observability Testing, Tools, and Frameworks
Culture and Methodologies
Agile Career Development Methodologies Team Management
Data Engineering
AI/ML Big Data Data Databases IoT
Software Design and Architecture
Cloud Architecture Containers Integration Microservices Performance Security
Coding
Frameworks Java JavaScript Languages Tools
Testing, Deployment, and Maintenance
Deployment DevOps and CI/CD Maintenance Monitoring and Observability Testing, Tools, and Frameworks

Last call! Secure your stack and shape the future! Help dev teams across the globe navigate their software supply chain security challenges.

Modernize your data layer. Learn how to design cloud-native database architectures to meet the evolving demands of AI and GenAI workloads.

Releasing software shouldn't be stressful or risky. Learn how to leverage progressive delivery techniques to ensure safer deployments.

Avoid machine learning mistakes and boost model performance! Discover key ML patterns, anti-patterns, data strategies, and more.

Related

  • Setting Up Data Pipelines With Snowflake Dynamic Tables
  • Harnessing Real-Time Insights With Streaming SQL on Kafka
  • Thread-Safety Pitfalls in XML Processing
  • Java Stream API: 3 Things Every Developer Should Know About

Trending

  • Revolutionizing Financial Monitoring: Building a Team Dashboard With OpenObserve
  • AWS to Azure Migration: A Cloudy Journey of Challenges and Triumphs
  • Why Database Migrations Take Months and How to Speed Them Up
  • Designing a Java Connector for Software Integrations
  1. DZone
  2. Coding
  3. JavaScript
  4. Testing Time-Based Reactor Core Streams With Virtual Time

Testing Time-Based Reactor Core Streams With Virtual Time

Take a look at Reactor Core's virtual time-based scheduler and the StepVerifier class to see how you can use them for time-based testing.

By 
Biju Kunjummen user avatar
Biju Kunjummen
·
Sep. 21, 17 · Tutorial
Likes (3)
Comment
Save
Tweet
Share
12.4K Views

Join the DZone community and get the full member experience.

Join For Free

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.

Stream (computing)

Published at DZone with permission of Biju Kunjummen, DZone MVB. See the original article here.

Opinions expressed by DZone contributors are their own.

Related

  • Setting Up Data Pipelines With Snowflake Dynamic Tables
  • Harnessing Real-Time Insights With Streaming SQL on Kafka
  • Thread-Safety Pitfalls in XML Processing
  • Java Stream API: 3 Things Every Developer Should Know About

Partner Resources

×

Comments

The likes didn't load as expected. Please refresh the page and try again.

ABOUT US

  • About DZone
  • Support and feedback
  • Community research
  • Sitemap

ADVERTISE

  • Advertise with DZone

CONTRIBUTE ON DZONE

  • Article Submission Guidelines
  • Become a Contributor
  • Core Program
  • Visit the Writers' Zone

LEGAL

  • Terms of Service
  • Privacy Policy

CONTACT US

  • 3343 Perimeter Hill Drive
  • Suite 100
  • Nashville, TN 37211
  • support@dzone.com

Let's be friends: