Add ReceiveChannel.consumeTo(destination) (#4520)

Use case: collecting elements up until the point the channel is
closed without losing the elements when `toList` when the exception
is thrown.

This function is similar to `Flow<T>.toList(destination)`,
which we already have, so the addition makes sense from the
point of view of consistency as well.

* Remove the assumed pitfall from `consumeEach` documentation

The way to trigger the pitfall that used to be described can be
boiled down to this pattern:
```kotlin
run {
    channel.consumeEach {
        if (channel.isEmpty) {
            // do something
            return@run
        } else {
            // do something else
        }
    }
}
```

However, here, `isEmpty` is already introducing a race condition,
so `consumeEach` itself does not cause any additional issues.

This does not seem like a pitfall in realistic scenarios after all.
`consumeEach` can perform an early return and thus erase
the elements present in the channel, but this also goes for
elements that possibly entered the channel long ago; without
explicitly checking if the channel is empty, there is no way to
distinguish these two scenarios.

Author: dkhalanskyjb

kotlinx-coroutines-core/api/kotlinx-coroutines-core.api

@@ -776,6 +776,7 @@ public final class kotlinx/coroutines/channels/ChannelsKt {
 	public static final fun consume (Lkotlinx/coroutines/channels/ReceiveChannel;Lkotlin/jvm/functions/Function1;)Ljava/lang/Object;
 	public static final fun consumeEach (Lkotlinx/coroutines/channels/BroadcastChannel;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun consumeEach (Lkotlinx/coroutines/channels/ReceiveChannel;Lkotlin/jvm/functions/Function1;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static final fun consumeTo (Lkotlinx/coroutines/channels/ReceiveChannel;Ljava/util/Collection;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
 	public static final fun consumes (Lkotlinx/coroutines/channels/ReceiveChannel;)Lkotlin/jvm/functions/Function1;
 	public static final fun consumesAll ([Lkotlinx/coroutines/channels/ReceiveChannel;)Lkotlin/jvm/functions/Function1;
 	public static final synthetic fun count (Lkotlinx/coroutines/channels/ReceiveChannel;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;

kotlinx-coroutines-core/api/kotlinx-coroutines-core.klib.api

@@ -1082,6 +1082,7 @@ final suspend fun <#A: kotlin/Any, #B: kotlin.collections/MutableCollection<in #
 final suspend fun <#A: kotlin/Any, #B: kotlinx.coroutines.channels/SendChannel<#A>> (kotlinx.coroutines.channels/ReceiveChannel<#A?>).kotlinx.coroutines.channels/filterNotNullTo(#B): #B // kotlinx.coroutines.channels/filterNotNullTo|filterNotNullTo@kotlinx.coroutines.channels.ReceiveChannel<0:0?>(0:1){0§<kotlin.Any>;1§<kotlinx.coroutines.channels.SendChannel<0:0>>}[0]
 final suspend fun <#A: kotlin/Any> (kotlinx.coroutines.channels/ReceiveChannel<#A>).kotlinx.coroutines.channels/receiveOrNull(): #A? // kotlinx.coroutines.channels/receiveOrNull|receiveOrNull@kotlinx.coroutines.channels.ReceiveChannel<0:0>(){0§<kotlin.Any>}[0]
 final suspend fun <#A: kotlin/Any?, #B: #A> (kotlinx.coroutines.flow/Flow<#B>).kotlinx.coroutines.flow/reduce(kotlin.coroutines/SuspendFunction2<#A, #B, #A>): #A // kotlinx.coroutines.flow/reduce|reduce@kotlinx.coroutines.flow.Flow<0:1>(kotlin.coroutines.SuspendFunction2<0:0,0:1,0:0>){0§<kotlin.Any?>;1§<0:0>}[0]
+final suspend fun <#A: kotlin/Any?, #B: kotlin.collections/MutableCollection<#A>> (kotlinx.coroutines.channels/ReceiveChannel<#A>).kotlinx.coroutines.channels/consumeTo(#B): #B // kotlinx.coroutines.channels/consumeTo|consumeTo@kotlinx.coroutines.channels.ReceiveChannel<0:0>(0:1){0§<kotlin.Any?>;1§<kotlin.collections.MutableCollection<0:0>>}[0]
 final suspend fun <#A: kotlin/Any?, #B: kotlin.collections/MutableCollection<in #A>> (kotlinx.coroutines.channels/ReceiveChannel<#A>).kotlinx.coroutines.channels/toCollection(#B): #B // kotlinx.coroutines.channels/toCollection|toCollection@kotlinx.coroutines.channels.ReceiveChannel<0:0>(0:1){0§<kotlin.Any?>;1§<kotlin.collections.MutableCollection<in|0:0>>}[0]
 final suspend fun <#A: kotlin/Any?, #B: kotlin.collections/MutableCollection<in #A>> (kotlinx.coroutines.flow/Flow<#A>).kotlinx.coroutines.flow/toCollection(#B): #B // kotlinx.coroutines.flow/toCollection|toCollection@kotlinx.coroutines.flow.Flow<0:0>(0:1){0§<kotlin.Any?>;1§<kotlin.collections.MutableCollection<in|0:0>>}[0]
 final suspend fun <#A: kotlin/Any?, #B: kotlin/Any?, #C: kotlin.collections/MutableMap<in #A, in #B>> (kotlinx.coroutines.channels/ReceiveChannel<kotlin/Pair<#A, #B>>).kotlinx.coroutines.channels/toMap(#C): #C // kotlinx.coroutines.channels/toMap|toMap@kotlinx.coroutines.channels.ReceiveChannel<kotlin.Pair<0:0,0:1>>(0:2){0§<kotlin.Any?>;1§<kotlin.Any?>;2§<kotlin.collections.MutableMap<in|0:0,in|0:1>>}[0]

kotlinx-coroutines-core/common/src/channels/Channels.common.kt

@@ -5,6 +5,9 @@
 package kotlinx.coroutines.channels
 
 import kotlinx.coroutines.*
+import kotlinx.coroutines.flow.consumeAsFlow
+import kotlinx.coroutines.flow.toCollection
+import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.selects.*
 import kotlin.contracts.*
 import kotlin.jvm.*
@@ -148,23 +151,23 @@ public inline fun <E, R> ReceiveChannel<E>.consume(block: ReceiveChannel<E>.() -
  *
  * In this example, several coroutines put elements into a single channel, and a single consumer processes the elements.
  * Once it finds the elements it's looking for, it stops [consumeEach] by making an early return.
- *
- * **Pitfall**: even though the name says "each", some elements could be left unprocessed if they are added after
- * this function decided to close the channel.
- * In this case, the elements will simply be lost.
- * If the elements of the channel are resources that must be closed (like file handles, sockets, etc.),
- * an `onUndeliveredElement` must be passed to the [Channel] on construction.
- * It will be called for each element left in the channel at the point of cancellation.
  */
 public suspend inline fun <E> ReceiveChannel<E>.consumeEach(action: (E) -> Unit): Unit =
     consume {
         for (e in this) action(e)
     }
 
 /**
- * Returns a [List] containing all the elements sent to this channel, preserving their order.
+ * [Consumes][consume] the elements of this channel into a list, preserving their order.
+ *
+ * This is a convenience function equivalent to calling [consumeAsFlow] followed by [kotlinx.coroutines.flow.toList].
+ * It is useful for testing code that uses channels to observe the elements the channel contains at the end of the test.
  *
- * This function will attempt to receive elements and put them into the list until the channel is
+ * There is no way to recover channel elements if the channel gets closed with an exception
+ * or to apply additional transformations to the elements before building the resulting collection.
+ * Please use [consumeAsFlow] and [kotlinx.coroutines.flow.toCollection] for such advanced use-cases.
+ *
+ * [toList] attempts to receive elements and put them into the list until the channel is
  * [closed][SendChannel.close].
  * Calling [toList] on channels that are not eventually closed is always incorrect:
  * - It will suspend indefinitely if the channel is not closed, but no new elements arrive.
@@ -173,8 +176,12 @@ public suspend inline fun <E> ReceiveChannel<E>.consumeEach(action: (E) -> Unit)
  *
  * If the channel is [closed][SendChannel.close] with a cause, [toList] will rethrow that cause.
  *
- * The operation is _terminal_.
- * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ * Since this function is implemented using [consume], it is _terminal_,
+ * meaning that [toList] will [cancel][ReceiveChannel.cancel] the channel before exiting.
+ * A [toList] call can finish before the sender closes the channel
+ * if it gets cancelled while waiting for the next element,
+ * or if [MutableList.add] fails with an exception.
+ * In both cases, the exception will be used for cancelling the channel and then rethrown.
  *
  * Example:
  * ```
@@ -189,11 +196,72 @@ public suspend inline fun <E> ReceiveChannel<E>.consumeEach(action: (E) -> Unit)
  * ```
  */
 public suspend fun <E> ReceiveChannel<E>.toList(): List<E> = buildList {
-    consumeEach {
-        add(it)
-    }
+    consumeEach(::add)
 }
 
+/**
+ * [Consumes][consume] the elements of this channel into the provided mutable collection.
+ *
+ * This is a convenience function equivalent to calling [consumeAsFlow]
+ * followed by [kotlinx.coroutines.flow.toCollection].
+ * Please use [consumeAsFlow] directly in scenarios where elements should undergo additional transformations
+ * before being added to the resulting collection.
+ *
+ * [consumeTo] attempts to receive elements and put them into the collection until the channel is
+ * [closed][SendChannel.close].
+ *
+ * If the channel is [closed][SendChannel.close] with a cause, [consumeTo] will rethrow that cause.
+ * However, the elements already received up to that point will remain in the collection.
+ *
+ * Since this function is implemented using [consume], it is _terminal_,
+ * meaning that [consumeTo] will [cancel][ReceiveChannel.cancel] the channel before exiting.
+ * A [consumeTo] call can finish before the sender closes the channel
+ * if it gets cancelled while waiting for the next element,
+ * or if [MutableCollection.add] fails with an exception.
+ * In both cases, the exception will be used for cancelling the channel and then rethrown.
+ *
+ * The intended use case for this function is collecting the remaining elements of a closed channel
+ * and processing them in a single batch.
+ *
+ * Example:
+ * ```
+ * val doContinue = AtomicBoolean(true)
+ *
+ * // Start the sender
+ * val channel = produce {
+ *     var i = 0
+ *     while (doContinue.load()) {
+ *         send(i++)
+ *         delay(10.milliseconds)
+ *         if (i == 42) break
+ *     }
+ * }
+ *
+ * // Start the consumer
+ * launch {
+ *     // Read elements until we suddenly decide to stop
+ *     // or until the channel is closed.
+ *     while (Random.nextInt(100) != 0) {
+ *         val nextElement = channel.receiveCatching()
+ *         if (nextElement.isClosed) return@launch
+ *         println("Received ${nextElement.getOrNull()}")
+ *     }
+ *     delay(100.milliseconds)
+ *     doContinue.store(false)
+ *     delay(100.milliseconds)
+ *     val remainingElements = mutableListOf<Int>()
+ *     try {
+ *         channel.consumeTo(remainingElements)
+ *     } finally {
+ *         println("Remaining elements: $remainingElements")
+ *     }
+ * }
+ * ```
+ */
+public suspend fun <E, C: MutableCollection<E>> ReceiveChannel<E>.consumeTo(collection: C): C =
+    consumeEach(collection::add).let { collection }
+
+
 @PublishedApi
 internal fun ReceiveChannel<*>.cancelConsumed(cause: Throwable?) {
     cancel(cause?.let {

kotlinx-coroutines-core/common/test/channels/ChannelsTest.kt

@@ -85,14 +85,13 @@ class ChannelsTest: TestBase() {
     }
 
     @Test
-    fun testEmptyList() = runTest {
+    fun testEmptyToList() = runTest {
         assertTrue(emptyList<Nothing>().asReceiveChannel().toList().isEmpty())
     }
 
     @Test
     fun testToList() = runTest {
         assertEquals(testList, testList.asReceiveChannel().toList())
-
     }
 
     @Test
@@ -104,6 +103,39 @@ class ChannelsTest: TestBase() {
         }
     }
 
+    @Test
+    fun testEmptyConsumeToWithDestination() = runTest {
+        val initialList = listOf(-1, -2, -3)
+        val destination = initialList.toMutableList()
+        emptyList<Nothing>().asReceiveChannel().consumeTo(destination)
+        assertEquals(initialList, destination)
+    }
+
+    @Test
+    fun testConsumeToWithDestination() = runTest {
+        val initialList = listOf(-1, -2, -3)
+        val destination = initialList.toMutableList()
+        testList.asReceiveChannel().consumeTo(destination)
+        assertEquals(initialList + testList, destination)
+    }
+
+    @Test
+    fun testConsumeToWithDestinationOnFailedChannel() = runTest {
+        val initialList = listOf(-1, -2, -3)
+        val destination = initialList.toMutableList()
+        val channel = Channel<Int>(10)
+        val elementsToSend = (1..5)
+        elementsToSend.forEach {
+            val result = channel.trySend(it)
+            assertTrue(result.isSuccess)
+        }
+        channel.close(TestException())
+        assertFailsWith<TestException> {
+            channel.consumeTo(destination)
+        }
+        assertEquals(initialList + elementsToSend, destination)
+    }
+
     private fun <E> Iterable<E>.asReceiveChannel(context: CoroutineContext = Dispatchers.Unconfined): ReceiveChannel<E> =
         GlobalScope.produce(context) {
             for (element in this@asReceiveChannel)