Reactor-core: Mono.thenMany lost Publisher data when subscribe and run forever when blockLast

[rj-hwang]

Actual behavior

Focus on code comments:

@Test
fun thenManyLostPublisherData() {
  Mono.empty<Int>()
    .thenMany<Int> { Flux.just(1, 2) }
    .subscribe { println(it) }  // not output item 1, 2
}

@Test
fun thenManyRunForever() {
  Mono.empty<Int>()
    .thenMany<Int> { Flux.just(1, 2) }
    .blockLast() // run forever
}

Source

I use these code to avoid above bugs:

@Test
fun useFlatMapIterableInsteadThenMany() {
  Mono.empty<Int>()
    .then(Flux.just(1, 2).collectList())
    .flatMapIterable { it.asIterable() }
    .subscribe { println(it) } // output item 1, 2
}
@Test
fun useFlatMapIterableInsteadThenMany2() {
  Mono.empty<Int>()
    .then(Flux.just(1, 2).collectList())
    .flatMapIterable { it.asIterable() }
    .doOnNext { println(it) } // output item 1, 2
    .blockLast()
}

Reactor Core version

3.1.8.RELEASE

JVM version (e.g. java -version)

java version "1.8.0_152"
Java(TM) SE Runtime Environment (build 1.8.0_152-b16)
Java HotSpot(TM) 64-Bit Server VM (build 25.152-b16, mixed mode)

[simonbasle]

As answered on StackOverflow

I tested the same code translated to Java, which worked, indicating that it had something to do with the Kotlin itself.

You are using Kotlin's "lambda as a last parameter" short-form syntax. The thing is, if you look at the thenMany method signature, it doesn't accept a Function, but a Publisher.

So why is that lambda accepted, and what does it represent?

It seems to be in fact interpreted as a Publisher (as it has only 1 method, subscribe(Subscriber))!

Replace the { } with ( ) and everything will be back to normal.

[simonbasle]

@rj-hwang As a side note, don't do tests with subscribe and println. Familiarize yourself with the StepVerifier class, which allow you to provide expectation of what the sequence should look like (a bit like assertions for reactive streams).

Your test examples would become:

@Test
fun thenManyLostPublisherData() {
  StepVerifier.create(Mono.empty<Int>()
    .thenMany<Int> { Flux.just(1, 2) }
  )
  .expectNext(1, 2)
  .expectComplete()
  .verify(); // don't forget this or the test will verify nothing and just pass
  //verify naturally blocks until the sequence could be asserted, no more tests
  //that complete too early because the sequence is asynchronous...
}

@Test
fun thenManyRunForever() {
  StepVerifier.create(Mono.empty<Int>()
    .thenMany<Int> { Flux.just(1, 2) }
    )
    .expectNext(1, 2)
    .verifyComplete(); //combines .expectComplete() and .verify()
    //.blockLast() // no need for blockLast() with the StepVerifier
}

[rj-hwang]

Thanks point it out. I wasn't aware of that. The original is so.