Как «застегнуть» два или более каналов сопрограммы?

Question

Итак, в RxJava мы могли бы просто сделать:

Observable.zip(someObservable, anotherObservable, BiFunction { a, b -> //do something }.subscribe { // do something }

Как мы можем сделать то же самое с Kotlin Coroutine Channels?

Answer 1

не идеальное решение, но, похоже, работает

@ExperimentalCoroutinesApi
private fun <T, R> CoroutineScope.zipChannels(
    channel1: ReceiveChannel<T>,
    channel2: ReceiveChannel<T>,
    zip: (T, T) -> R
): ReceiveChannel<R> = produce {
    val iterator1 = channel1.iterator()
    val iterator2 = channel2.iterator()
    while (iterator1.hasNext() && iterator2.hasNext()) {
        val value1 = iterator1.next()
        val value2 = iterator2.next()
        send(zip(value1, value2))
    }
    channel1.cancel()
    channel2.cancel()
    close()
}

Обновление

Также есть устаревший оператор zip