Add await() extension on Rector Mono type

[elizarov]

It should simply be an inline alias to Publisher.awaitSingle(). But special care shall be taken about Mono<Void> types that never emit any value. They need awaitCompletion() which is missing altogether now.

[jorge-recio]

I would like to work on this issue.
If suspend fun Mono<T>.await(): T is implemented as an inline alias of suspend fun Publisher.awaitSingle(): T and Mono.await() is called the result will be a TypeCastException: null cannot be cast to non-null type. What should be the expected behavior?

About the implementation of suspend fun Mono<T>.awaitCompletion(): Unit can be implemented using a new suspend fun Publisher<T>.awaitCompletion(single: Boolean): Unit with single = true. But, can this feature be implemented as part of this issue or needs to be done in a different one?

[elizarov]

They both need to be implemented together, but they need careful design on their behaviour in various edge-cases. It is not clear at this moment what should happen when Mono completes without emitting a value and how common this behavior is in the real code.

[wickedev]

Since this issue hasn't progressed for a year, I've written PoC-level code. Need more cases then this?

suspend inline fun <reified T> Mono<T>.await(): T {
    if (null is T) {
        // for Mono<T?>
        return awaitFirstOrNull() as T
    }

    return when (T::class) {
        // for Mono<Unit> or Mono<Void>
        Unit::class, Void::class -> {
            @Suppress("UNCHECKED_CAST", "ReactiveStreamsSubscriberImplementation")
            suspendCancellableCoroutine { cont ->
                cont as CancellableContinuation<T?>
                subscribe(object : Subscriber<T?> {
                    override fun onSubscribe(s: Subscription) {
                        cont.invokeOnCancellation {
                            s.cancel()
                        }
                        s.request(1)
                    }

                    override fun onNext(t: T?) {
                        error("Mono<Void> or Mono<Unit> cannot emit value")
                    }

                    override fun onError(t: Throwable) {
                        cont.resumeWithException(t)
                    }

                    override fun onComplete() {
                        cont.resume(null)
                    }
                })
            }
        }
        // for Mono<T>
        else -> awaitSingle() as T
    }
}

object MonoAwaitTest : Spek({
    test("mono await") {
        runBlocking {
            val m1: Mono<Int> = Mono.just(100)
            val r1: Int = m1.await()
            r1.should.be.equal(100)

            val m2: Mono<String> = Mono.just("value")
            val r2: String = m2.await()
            r2.should.be.equal("value")

            val m3: Mono<String?> = Mono.justOrEmpty("String?")
            val r3: String? = m3.await()
            r3.should.be.equal("String?")

            val m4: Mono<String?> = Mono.justOrEmpty(null)
            val r4: String? = m4.await()
            r4.should.be.equal(null)

            val m5: Mono<Unit> = Mono.empty()
            val r5: Unit = m5.await()
            r5.should.be.equal(null)

            val m6: Mono<Void> = Mono.empty()
            val r6: Void = m6.await()
            r6.should.be.equal(null)
        }
    }
})

[dkhalanskyjb]

It seems from the documentation (https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Mono.html) that Mono<T> not returning a value most be fairly common. For example, the existence of Mono<T>.defaultIfEmpty(defaultValue: T): Mono<T> already implies strongly that not returning a value is not limited to Mono<Void>, and this operator is used somewhat commonly: https://github.com/search?l=Java&q=Mono+defaultIfEmpty&type=Code. Thus, it looks to me like using Publisher.awaitSingle() would not reflect the intended semantics correctly; instead, Mono seems to be an equivalent for RxJava's Maybe and should probably be handled in the same way.