Enforce exception transparency in SafeFlow

[qwwdfsad]

We can enforce it by caching and checking any exception that was thrown by downstream.emit.
The trick is to have (almost) the same performance as before while giving up on tail-call.
For that, we can use a trick with "reusable continuation" as in TakeFlowCollector.

Enforcing this invariant is crucial, otherwise, it may lead to really weird interactions, for example, consider the following pipeline:

val flow = flow {
    try {
        emit(1)
    } catch (e: Throwable) {
        emit(2)
    }
}

// Somewhere on consumer side
flow.collect {
    println("Processed '$it' and expecting no more elements")
    throw CancellationException()
}

collect lambda will be invoked twice even though it threw an exception on the first emit