Flux.handle example

FluxHandleExample.java

import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

public class FluxHandleExample {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        CompletableFuture<Boolean> completableFuture = new CompletableFuture<>();

        Flux.just(0, 1, 2, 3)
                .log("ONE")
                .<Integer>handle((integer, synchronousSink) -> synchronousSink.next(integer * 10))
                .delayElements(Duration.ofSeconds(3))
                .log("TWO")
                .doOnComplete(() -> completableFuture.complete(true))
                .subscribe();

        completableFuture.get();
    }
}