kotlin - 如何 'zip'两个或更多协程 channel ？

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