title: How to ensure that a Flux with infinite stream completes in a given time?
body: <p>I have a <code>Flux</code> generator (from project reactor) for stream that delays 1 second to emit each element. And because the stream is infinite I also decide how many elements I would like to take using the <code>take()</code> method. I want to test that such a method can be executed in a given time. I tried to create my solution <a href="https://stackoverflow.com/a/44676423/2096986">inspired by this answer</a> but without success.</p>
  <p>This is the <code>Flux</code> method</p>
  <pre><code>import reactor.core.publisher.Flux;
  import java.time.Duration;
  @Slf4j
  public class FluxAndMonoStreams {
      public Flux&lt;Double&gt; createFluxDoubleWithDelay(long numberOfElements, long delaySec) {
          return Flux.interval(Duration.ofSeconds(delaySec))
                  .map(l -&gt; l.doubleValue())
                  .take(numberOfElements)
                  .log();
      }
  }
  </code></pre>
  <p>and this is the unit test. I have tried the approaches using <code>.thenConsumeWhile(value -&gt; true)</code>, <code>.expectNextCount(4)</code>, <code>.expectNext(0.0, 1.0, 2.0, 3.0)</code> after waiting 4 seconds on the <code>.thenAwait(Duration.ofSeconds(4))</code>. But none of them works.</p>
  <pre><code>import org.junit.jupiter.api.Test;
  import reactor.core.publisher.Flux;
  import reactor.core.scheduler.Scheduler;
  import reactor.core.scheduler.Schedulers;
  import reactor.test.StepVerifier;
  import java.time.Duration;
  import java.util.concurrent.atomic.AtomicInteger;
  public class FluxAndMonoStreamsTest {
      FluxAndMonoStreams myFluxAndMonoStreams = new FluxAndMonoStreams();
      @Test
      void testCreateFluxStreamWithMapAndVerifyDelay() {
          Flux&lt;Double&gt; streamLongFlux = myFluxAndMonoStreams
                  .createFluxDoubleWithDelay(4, 1);
  
          Scheduler scheduler = Schedulers.newSingle(&quot;test&quot;);
          AtomicInteger incrementer = new AtomicInteger();
  
          StepVerifier
                  .withVirtualTime(() -&gt; streamLongFlux
                          .subscribeOn(scheduler)
                          .doOnNext(value -&gt; incrementer.incrementAndGet())
                  )
                  .expectSubscription()
                  .thenAwait(Duration.ofSeconds(4))
                  // .thenConsumeWhile(value -&gt; true)
                  // .expectNextCount(4)
                  // .expectNext(0.0, 1.0, 2.0, 3.0)
                  .verifyComplete()
          ;
      }
  }
  </code></pre>
  <p>I am receiving this error which says that I am not consuming values <code>1.0</code>, <code>2.0</code>, ... Basically I want to make sure that I consume only the values <code>0.0, 1.0, 2.0, 3.0</code> within 4 seconds. How do I do that?</p>
  <pre><code>12:37:31.853 [Test worker] DEBUG reactor.util.Loggers - Using Slf4j logging framework
  12:37:31.912 [test-1] INFO reactor.Flux.Take.1 - onSubscribe(FluxTake.TakeSubscriber)
  12:37:31.915 [test-1] INFO reactor.Flux.Take.1 - request(unbounded)
  12:37:32.917 [parallel-1] INFO reactor.Flux.Take.1 - onNext(0.0)
  12:37:32.921 [parallel-1] INFO reactor.Flux.Take.1 - cancel()
  
  expectation &quot;expectComplete&quot; failed (expected: onComplete(); actual: onNext(0.0))
  java.lang.AssertionError: expectation &quot;expectComplete&quot; failed (expected: onComplete(); actual: onNext(0.0))
  </code></pre>
  <p><strong>Additional explain</strong>: Just to clarify one more time the behavior that I want. If I uncomment <code> .expectNext(0.0, 1.0, 2.0, 3.0)</code> and replace 4 seconds to 2 seconds in <code>.thenAwait(Duration.ofSeconds(2))</code>, the test should fail. But it is not.</p>