Introduce merge operator

Fixes #1491

Author: qwwdfsad

binary-compatibility-validator/reference-public-api/kotlinx-coroutines-core.txt

@@ -926,7 +926,9 @@ public final class kotlinx/coroutines/flow/FlowKt {
 	public static final fun map (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun mapLatest (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun mapNotNull (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun merge (Ljava/lang/Iterable;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun merge (Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
+	public static final fun merge ([Lkotlinx/coroutines/flow/Flow;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun observeOn (Lkotlinx/coroutines/flow/Flow;Lkotlin/coroutines/CoroutineContext;)Lkotlinx/coroutines/flow/Flow;
 	public static final synthetic fun onCompletion (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function2;)Lkotlinx/coroutines/flow/Flow;
 	public static final fun onCompletion (Lkotlinx/coroutines/flow/Flow;Lkotlin/jvm/functions/Function3;)Lkotlinx/coroutines/flow/Flow;

kotlinx-coroutines-core/common/src/flow/internal/Merge.kt

@@ -75,3 +75,23 @@ internal class ChannelFlowMerge<T>(
     override fun additionalToStringProps(): String =
         "concurrency=$concurrency, "
 }
+
+internal class ChannelLimitedFlowMerge<T>(
+    private val flows: Iterable<Flow<T>>,
+    context: CoroutineContext = EmptyCoroutineContext,
+    capacity: Int = Channel.BUFFERED
+) : ChannelFlow<T>(context, capacity) {
+    override fun create(context: CoroutineContext, capacity: Int): ChannelFlow<T> =
+        ChannelLimitedFlowMerge(flows, context, capacity)
+
+    override fun produceImpl(scope: CoroutineScope): ReceiveChannel<T> {
+        return scope.flowProduce(context, capacity, block = collectToFun)
+    }
+
+    override suspend fun collectTo(scope: ProducerScope<T>) {
+        val collector = SendingCollector(scope)
+        flows.forEach { flow ->
+            scope.launch { flow.collect(collector) }
+        }
+    }
+}

kotlinx-coroutines-core/common/src/flow/operators/Merge.kt

@@ -81,6 +81,42 @@ public fun <T> Flow<Flow<T>>.flattenConcat(): Flow<T> = flow {
     collect { value -> emitAll(value) }
 }
 
+/**
+ * Merges the given flows into a single flow without preserving an order of elements.
+ * All flows are merged concurrently, without limit on the number of simultaneously collected flows.
+ *
+ * ### Operator fusion
+ *
+ * Applications of [flowOn], [buffer], [produceIn], and [broadcastIn] _after_ this operator are fused with
+ * its concurrent merging so that only one properly configured channel is used for execution of merging logic.
+ */
+@ExperimentalCoroutinesApi
+public fun <T> Iterable<Flow<T>>.merge(): Flow<T> {
+    /*
+     * This is a fuseable implementation of the following operator:
+     * channelFlow {
+     *    forEach { flow ->
+     *        launch {
+     *            flow.collect { send(it) }
+     *        }
+     *    }
+     * }
+     */
+    return ChannelLimitedFlowMerge(this)
+}
+
+/**
+ * Merges the given flows into a single flow without preserving an order of elements.
+ * All flows are merged concurrently, without limit on the number of simultaneously collected flows.
+ *
+ * ### Operator fusion
+ *
+ * Applications of [flowOn], [buffer], [produceIn], and [broadcastIn] _after_ this operator are fused with
+ * its concurrent merging so that only one properly configured channel is used for execution of merging logic.
+ */
+@ExperimentalCoroutinesApi
+public fun <T> merge(vararg flows: Flow<T>): Flow<T> = flows.asIterable().merge()
+
 /**
  * Flattens the given flow of flows into a single flow with a [concurrency] limit on the number of
  * concurrently collected flows.

kotlinx-coroutines-core/common/test/flow/operators/MergeTest.kt

@@ -0,0 +1 @@
+/* * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license. */package kotlinx.coroutines.flowimport kotlinx.coroutines.*import kotlin.test.*import kotlinx.coroutines.flow.merge as originalMergeabstract class MergeTest : TestBase() {    abstract fun <T> Iterable<Flow<T>>.merge(): Flow<T>    @Test    fun testMerge() = runTest {        val n = 100        val sum = (1..n).map { flowOf(it) }            .merge()            .sum()        assertEquals(n * (n + 1) / 2, sum)    }    @Test    fun testSingle() = runTest {        val flow = listOf(flowOf(), flowOf(42), flowOf()).merge()        val value = flow.single()        assertEquals(42, value)    }    @Test    fun testNulls() = runTest {        val list = listOf(flowOf(1), flowOf(null), flowOf(2)).merge().toList()        assertEquals(listOf(1, null, 2), list)    }    @Test    fun testContext() = runTest {        val flow = flow {            emit(NamedDispatchers.name())        }.flowOn(NamedDispatchers("source"))        val result = listOf(flow).merge().flowOn(NamedDispatchers("irrelevant")).toList()        assertEquals(listOf("source"), result)    }    @Test    fun testIsolatedContext() = runTest {        val flow = flow {            emit(NamedDispatchers.name())        }        val result = listOf(flow.flowOn(NamedDispatchers("1")), flow.flowOn(NamedDispatchers("2")))            .merge()            .flowOn(NamedDispatchers("irrelevant"))            .toList()        assertEquals(listOf("1", "2"), result)    }}class IterableMergeTest : MergeTest() {    override fun <T> Iterable<Flow<T>>.merge(): Flow<T> = originalMerge()}class VarargMergeTest : MergeTest() {    override fun <T> Iterable<Flow<T>>.merge(): Flow<T> = originalMerge(*toList().toTypedArray())}