Make subscribe methods suspending

[joffrey-bion]

Please describe the problem/use case that this feature is meant to solve/enable
Sometimes the user may want to send a message to the server that is known to trigger some messages on a subscription.
In this case, we need to make sure that we have actually subscribed before sending the message.

One such example is when using a short-lived susbcription to implement a request-response mechanism.

Describe the solution you'd like
The ideal code for it would be:

val sub = session.subscribe(receiveDestination, deserializer) // suspends until actual subscription happens
session.send(sendDestination, request, serializer)
val response = sub.first()

The problem is that Krossbow only supports cold flows that subscribe upon collection, so the subscription only happens when first() is called here.

Describe alternatives you've considered

The current workaround is using UNDISPATCHED to try and narrow down the window that can cause problems, and add some delay to ensure we give enough time for the subscription to happen.
But it is not 100% guaranteed to work:

coroutineScope {
    val sub = async(start = CoroutineStart.UNDISPATCHED) {
        session.subscribe(receiveDestination, deserializer).first()
    }
    delay(30) // ensures the subscription happened
    session.convertAndSend(sendDestination, request, serializer)
    val response = sub.await()
}

Context
Kotlin version: 1.4.0
Kotlin target: all targets