Канал выхода из вентилятора / вентилятора в закрытии

Я производю предметы, потребляя из нескольких совместных процедур и отбрасывая назад в resultChannel. Производитель закрывает свой канал после последнего элемента.

Код никогда не заканчивается как resultChannel никогда не закрывается. Как обнаружить и правильно закончить итерацию так hasNext() return false ?

 val inputData = (0..99).map { "Input$it" } val threads = 10 val bundleProducer = produce<String>(CommonPool, threads) { inputData.forEach { item -> send(item) println("Producing: $item") } println("Producing finished") close() } val resultChannel = Channel<String>(threads) repeat(threads) { launch(CommonPool) { bundleProducer.consumeEach { println("CONSUMING $it") resultChannel.send("Result ($it)") } } } val iterator = object : Iterator<String> { val iterator = resultChannel.iterator() override fun hasNext() = runBlocking { iterator.hasNext() } override fun next() = runBlocking { iterator.next() } }.asSequence() println("Starting interation...") val result = iterator.toList() println("finish: ${result.size}") 

Вы можете запустить сопрограмму, которая ждет, когда потребители закончат, а затем resultChannel .

Во-первых, перепишите код, запускающий потребителей для сохранения Job s:

 val jobs = (1..threads).map { launch(CommonPool) { bundleProducer.consumeEach { println("CONSUMING $it") resultChannel.send("Result ($it)") } } } 

Затем запустите другую сопрограмму, которая закрывает канал, когда все Job будут выполнены:

 launch(CommonPool) { jobs.forEach { it.join() } resultChannel.close() }