Как приостановить поток, пока коллектор не достроен? ⇐ Android

[Anonymous]

Я хочу использовать Flow для реализации модели производителя и потребителя. Когда состояние не израсходовано всеми потребителями (действие не закончилось), производитель блокируется.

Код: Выделить всё

fun main() {
val flow = MutableSharedFlow(
replay = 0,
extraBufferCapacity = 0,
onBufferOverflow = BufferOverflow.SUSPEND
)
runBlocking {
launch {
flow.collect {
println("collector1: start collect ${it}")
delay(1000)
println("collector1: collect finish ${it}")
}
}
launch {
(3..4).forEach {
println("emit ${it}")
flow.emit(it)
}
}
}
}

Я получил:

Код: Выделить всё

emit 3
collector1: start collect 3
emit 4
collector1: collect finish 3
collector1: start collect 4
collector1: collect finish 4

Но я ожидаю

Код: Выделить всё

emit 3
collector1: start collect 3
collector1: collect finish 3
emit 4
collector1: start collect 4
collector1: collect finish 4

Дальше, когда имеется несколько потребителей:

Код: Выделить всё

fun main() {
val flow = MutableSharedFlow(
replay = 0,
extraBufferCapacity = 0,
onBufferOverflow = BufferOverflow.SUSPEND
)
runBlocking {
launch {
flow.collect {
println("collector1: start collect ${it}")
delay(1000)
println("collector1: collect finish ${it}")
}
}
launch {
flow.collect {
println("collector2: start collect ${it}")
delay(2000)
println("collector2: collect finish ${it}")
}
}
launch {
(3..4).forEach {
println("emit ${it}")
flow.emit(it)
}
}
}
}

Я ожидаю

Код: Выделить всё

emit 3
collector1: start collect 3
collector2: start collect 3
collector1: collect finish 3
collector2: collect finish 3
emit 4
collector1: start collect 4
collector2: start collect 4
collector1: collect finish 4
collector2: collect finish 4

Я пробовал заблокировать или использовать канал, но они не смогли решить эту проблему. Разве мне не следует использовать поток для решения такого рода проблем?

Подробнее здесь: https://stackoverflow.com/questions/792 ... t-finished