Как закрыть канал после завершения всех сопрограмм производителя?

Question

Рассмотрим следующий код:

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
    val channel = Channel<String>()
    launch {
        channel.send("A1")
        channel.send("A2")
        log("A done")
    }
    launch {
        channel.send("B1")
        log("B done")
    }
    launch {
        for (x in channel) {
            log(x)
        }
    }
}

fun log(message: Any?) {
    println("[${Thread.currentThread().name}] $message")
}

Исходная версия имеет такую ​​сопрограмму-получатель:

launch {
        repeat(3) {
            val x = channel.receive()
            log(x)
        }
    }

Он ожидает только 3 сообщения в канале. Если я изменю его на первую версию, мне нужно будет закрыть канал после завершения всех сопрограмм производителя. Как я могу это сделать?

Answer 1

Возможное решение - создать работу, которая будет ждать всех channel.send() закончить и позвонить channel.close() в invokeOnCompletion этой работы:

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.*

fun main() = runBlocking<Unit> {
    val channel = Channel<String>()
    launch {
      launch {
          channel.send("A1")
          channel.send("A2")
          log("A done")
      }
      launch {
          channel.send("B1")
          log("B done")
      }
    }.invokeOnCompletion {
        channel.close()
    }
    launch {
        for (x in channel) {
            log(x)
        }
    }
}

fun log(message: Any?) {
    println("[${Thread.currentThread().name}] $message")
}

Answer 2

Основываясь на принятом ответе, вы можете написать расширение, инкапсулирующее область действия открытого канала:

      import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.flow.receiveAsFlow
import kotlinx.coroutines.flow.toList

fun main() = runBlocking {
    val channel = Channel<String>()
    launch {
        produce(channel) {
            send("foo")
            send("bar")
            send("foobar")
        }
    }
    val asList = channel.receiveAsFlow()
        .toList()
    println(asList)
}

fun <T> CoroutineScope.produce(channel: Channel<T>, block: suspend Channel<T>.() -> Unit) =
        channel.run {
            launch(context = coroutineContext) { block() }
        }
        .invokeOnCompletion { channel.close() }