Why does Thread.sleep() trigger the subscription to Flux.interval()?

user_3790512 0
Viewed 922

If, in a main() method, I execute this

Flux.just(1,2)
    .log()
    .subscribe();

I get this in the console:

[ INFO] (main) | onSubscribe([Synchronous Fuseable] FluxArray.ArraySubscription)
[ INFO] (main) | request(unbounded)
[ INFO] (main) | onNext(1)
[ INFO] (main) | onNext(2)
[ INFO] (main) | onComplete()

If instead of just() I use the interval() method:

Flux.interval(Duration.ofMillis(100))
    .take(2)
    .log()
    .subscribe();

the elements are not logged, unless I add Thread.sleep() which gives me:

[ INFO] (main) onSubscribe(FluxTake.TakeSubscriber)
[ INFO] (main) request(unbounded)
[ INFO] (parallel-1) onNext(0)
[ INFO] (parallel-1) onNext(1)
[ INFO] (parallel-1) onComplete()

The question is: why do I require to pause a thread to actually trigger the subscription ?

project-reactor
3 Answers

You need to wait on main thread and let the execution complete. Your main thread terminates before next element is generated. Your first element is generated after 100 ms so you need to wait/block the main thread. Try this:

CountDownLatch latch = new CountDownLatch(1);
Flux.interval(Duration.ofMillis(100))
        .take(2)
        .doOnComplete(latch::countDown)
        .log()
        .subscribe();
latch.await(); // wait for doOnComplete

CountDownLatch:

A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.

user_60993470

Flux.interval(..) emits items on a parallel, freeing up your calling thread. This is why your program exits. What you should do is:

Flux.interval(Duration.of(duration))
    .take(n)
    .doOnNext(this::logElement)
    .blockLast();

This will block the calling thread till doOnNext() emits the last time (which should be the second item form the upstream).

user_67144260

Flux.interval(Duration) produces a Flux that is infinite and emits regular ticks from a clock.

So your first example has no concurrency, everything happens on the same thread. The subscription and all events must complete before the method and process ends.

By using Flux.interval you have added concurrency and asynchronous behaviour into your example. Interval is a regular clock like a metronome. You subscribe but immediately complete and the process will usually finish before any work (onNext) is done.

user_15426670
Related