Add ReceiveChannel.toList(destination)

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.

Author: dkhalanskyjb

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

@@ -194,6 +194,43 @@ public suspend fun <E> ReceiveChannel<E>.toList(): List<E> = buildList {
     }
 }
 
+/**
+ * Consumes the elements of this channel into the given [destination] mutable list.
+ *
+ * This function will attempt 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.
+ * - If new elements do arrive and the channel is not eventually closed, [toList] will use more and more memory
+ *   until exhausting it.
+ *
+ * If the channel is [closed][SendChannel.close] with a cause, [toList] will rethrow that cause.
+ * However, the [destination] list is left in a consistent state containing all the elements received from the channel
+ * up to that point.
+ *
+ * The operation is _terminal_.
+ * This function [consumes][ReceiveChannel.consume] all elements of the original [ReceiveChannel].
+ *
+ * Example:
+ * ```
+ * val values = listOf(1, 5, 2, 9, 3, 3, 1)
+ * // start a new coroutine that creates a channel,
+ * // sends elements to it, and closes it
+ * // once the coroutine's body finishes
+ * val channel = produce {
+ *    values.forEach { send(it) }
+ * }
+ * val destination = mutableListOf<Int>()
+ * channel.toList(destination)
+ * check(destination == values)
+ * ```
+ */
+public suspend inline fun <T> ReceiveChannel<T>.toList(destination: MutableList<T>) {
+    consumeEach {
+        destination.add(it)
+    }
+}
+
 @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 testEmptyToListWithDestination() = runTest {
+        val initialList = listOf(-1, -2, -3)
+        val destination = initialList.toMutableList()
+        emptyList<Nothing>().asReceiveChannel().toList(destination)
+        assertEquals(initialList, destination)
+    }
+
+    @Test
+    fun testToListWithDestination() = runTest {
+        val initialList = listOf(-1, -2, -3)
+        val destination = initialList.toMutableList()
+        testList.asReceiveChannel().toList(destination)
+        assertEquals(initialList + testList, destination)
+    }
+
+    @Test
+    fun testToListWithDestinationOnFailedChannel() = runTest {
+        val initialList = listOf(-1, -2, -3)
+        val destination = initialList.toMutableList()
+        val channel = Channel<Int>(10)
+        val elementsToSend = (1..5).toList()
+        elementsToSend.forEach {
+            val result = channel.trySend(it)
+            assertTrue(result.isSuccess)
+        }
+        channel.close(TestException())
+        assertFailsWith<TestException> {
+            channel.toList(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)