Added extensions for virtual time testing

src/main/kotlin/reactor/kotlin/test/StepVerifierExtensions.kt

@@ -16,12 +16,14 @@
 
 package reactor.kotlin.test
 
+import org.reactivestreams.Publisher
 import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.test.StepVerifier
 import reactor.test.StepVerifier.Assertions
 import reactor.test.StepVerifier.LastStep
 import reactor.test.StepVerifierOptions
+import reactor.test.scheduler.VirtualTimeScheduler
 import java.time.Duration
 import kotlin.reflect.KClass
 
@@ -128,4 +130,38 @@ fun <T> Mono<T>.test(n: Long): StepVerifier.FirstStep<T> = StepVerifier.create(t
  */
 fun <T> Mono<T>.test(options: StepVerifierOptions): StepVerifier.FirstStep<T> = StepVerifier.create(this, options)
 
+/**
+ * Extension for testing the supplied [Publisher] with [StepVerifier] API, using a [VirtualTimeScheduler].
+ *
+ * @see [StepVerifier.withVirtualTime]
+ * @author Mikael Elm
+ */
+fun <T> (() -> Publisher<T>).testUsingVirtualTime(): StepVerifier.FirstStep<T> =
+ StepVerifier.withVirtualTime { invoke() }
+
+/**
+ * Extension for testing the supplied [Publisher] with [StepVerifier] API, using a [VirtualTimeScheduler].
+ *
+ * @see [StepVerifier.withVirtualTime]
+ * @author Mikael Elm
+ */
+fun <T> (() -> Publisher<T>).testUsingVirtualTime(n: Long): StepVerifier.FirstStep<T> =
+ StepVerifier.withVirtualTime({ invoke() }, n)
 
+/**
+ * Extension for testing a [Publisher] with [StepVerifier] API, using a [VirtualTimeScheduler].
+ *
+ * @see [StepVerifier.withVirtualTime]
+ * @author Mikael Elm
+ */
+fun <T> Publisher<T>.testUsingVirtualTime(vtsLookup: () -> VirtualTimeScheduler, n: Long): StepVerifier.FirstStep<T> =
+ StepVerifier.withVirtualTime({ this }, vtsLookup, n)
+
+/**
+ * Extension for testing a [Publisher] with [StepVerifier] API, using a [VirtualTimeScheduler].
+ *
+ * @see [StepVerifier.withVirtualTime]
+ * @author Mikael Elm
+ */
+fun <T> Publisher<T>.testUsingVirtualTime(options: StepVerifierOptions): StepVerifier.FirstStep<T> =
+ StepVerifier.withVirtualTime({ this }, options)

src/test/kotlin/reactor/kotlin/test/StepVerifierExtensionsTests.kt

@@ -17,7 +17,12 @@
 package reactor.kotlin.test
 
 import org.junit.Test
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
 import reactor.kotlin.core.publisher.toMono
+import reactor.test.StepVerifierOptions
+import reactor.test.scheduler.VirtualTimeScheduler
+import java.time.Duration
 
 class StepVerifierExtensionsTests {
 
@@ -53,4 +58,53 @@ class StepVerifierExtensionsTests {
  .verifyError<IllegalStateException>()
  }
 
+ @Test
+ fun `testUsingVirtualTime()`() {
+ { Mono.just("foo").delayElement(Duration.ofDays(1)) }
+ .testUsingVirtualTime()
+ .thenAwait(Duration.ofDays(1))
+ .expectNext("foo")
+ .expectComplete()
+ .verify(Duration.ofSeconds(1))
+ }
+
+ @Test
+ fun `testUsingVirtualTime() requesting n elements`() {
+ { Flux.just("foo", "bar").delayElements(Duration.ofDays(1)) }
+ .testUsingVirtualTime(1)
+ .thenAwait(Duration.ofDays(1))
+ .expectNext("foo")
+ .expectNoEvent(Duration.ofDays(7))
+ .thenRequest(1)
+ .thenAwait(Duration.ofDays(1))
+ .expectNext("bar")
+ .expectComplete()
+ .verify(Duration.ofSeconds(1))
+ }
+
+ @Test
+ fun `testUsingVirtualTime() passing scheduler and requesting n elements`() {
+ val vts = VirtualTimeScheduler.create()
+ Flux.just("foo", "bar").delayElements(Duration.ofDays(1), vts)
+ .testUsingVirtualTime({ vts }, 1)
+ .thenAwait(Duration.ofDays(1))
+ .expectNext("foo")
+ .expectNoEvent(Duration.ofDays(7))
+ .thenRequest(1)
+ .thenAwait(Duration.ofDays(1))
+ .expectNext("bar")
+ .expectComplete()
+ .verify(Duration.ofSeconds(1))
+ }
+
+ @Test
+ fun `testUsingVirtualTime() passing options`() {
+ val vts = VirtualTimeScheduler.create()
+ Mono.just("foo").delayElement(Duration.ofDays(1), vts)
+ .testUsingVirtualTime(StepVerifierOptions.create().virtualTimeSchedulerSupplier { vts })
+ .thenAwait(Duration.ofDays(1))
+ .expectNext("foo")
+ .expectComplete()
+ .verify(Duration.ofSeconds(1))
+ }
 }