@Test
    fun testLevel1() {
        val result = Mono.just("")
            .map { "123" }
            .block()
        assertEquals("123", result)
    }

fun nonBlockingMethod1sec(data: String)
    = data.toMono().delayElement(Duration.ofMillis(1000))
    @Test
    fun testLevel2() {
        val result = nonBlockingMethod1sec("Hello world")
            .flatMap { nonBlockingMethod1sec(it) }
            .block()
        assertEquals("Hello world", result)
    }

fun collectTasks() = (0..99)
    @Test
    fun testLevel3() {
        val result = nonBlockingMethod1sec("Hello world")
            .flatMap { businessContext ->
                collectTasks()
                    .toFlux()
                    .map {
                        businessContext + it
                    }
                    .collectList()
            }
            .block()!!
        assertEquals(collectTasks().toList().size, result.size)
    }

fun collectTasks() = (0..100)
    @Test
    fun testLevel4() {
        val result = nonBlockingMethod1sec("Hello world")
            .flatMap { businessContext ->
                collectTasks().toFlux()
                    .flatMap {
                        Mono.deferContextual { reactiveContext ->
                            val hash = businessContext + it + reactiveContext["requestId"]
                            hash.toMono()
                        }
                    }.collectList()
            }
            .contextWrite { it.put("requestId", UUID.randomUUID().toString()) }
            .block()!!
        assertEquals(collectTasks().toList().size, result.size)
    }

fun collectTasks() = (0..1000)
    fun doSomethingNonBlocking(data: String)
        = data.toMono().delayElement(Duration.ofMillis(1000))
    fun doSomethingBlocking(data: String): String {
        Thread.sleep(1000); return data
    }
    val pool = Schedulers.newBoundedElastic(10, Int.MAX_VALUE, "test-pool")
    private val logger = getLogger()
    @Test
    fun testLevel5() {
        val counter = AtomicInteger(0)
        val result = nonBlockingMethod1sec("Hello world")
            .flatMap { _ ->
                collectTasks().toFlux()
                    .parallel()
                    .runOn(pool)
                    .flatMap {
                        Mono.deferContextual { _ ->
                            doSomethingNonBlocking(it.toString())
                                .doOnRequest { logger.info("Added task in pool ${counter.incrementAndGet()}") }
                                .doOnNext { logger.info("Non blocking code finished ${counter.get()}") }
                                .map { doSomethingBlocking(it) }
                                .doOnNext { logger.info("Removed task from pool ${counter.decrementAndGet()}") }
                        }
                    }.sequential()
                    .collectList()
            }
            .block()!!
        assertEquals(collectTasks().toList().size, result.size)
    }

val result = nonBlockingMethod1sec("Hello world")
            .flatMap { _ ->
                collectTasks().toFlux()
                    .parallel(1)
                    .runOn(pool, 1)
                    .flatMap({
                        Mono.deferContextual { _ ->
                            doSomethingNonBlocking(it.toString())
                        }
                    }, false, 1, 1)
                    .sequential()
                    .collectList()
            }
            .block()!!