Skip to content

Add ReceiveChannel.toList(destination) #4520

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 3 commits into
base: develop
Choose a base branch
from
Open

Add ReceiveChannel.toList(destination) #4520

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
038c6b7
Add ReceiveChannel.toList(destination)
dkhalanskyjb Sep 10, 2025
2d8f52b
Keep only one overload
dkhalanskyjb Sep 10, 2025
6039cf8
Fixup
dkhalanskyjb Sep 10, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion kotlinx-coroutines-core/api/kotlinx-coroutines-core.api
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -820,7 +820,9 @@ public final class kotlinx/coroutines/channels/ChannelsKt {
public static synthetic fun takeWhile$default (Lkotlinx/coroutines/channels/ReceiveChannel;Lkotlin/coroutines/CoroutineContext;Lkotlin/jvm/functions/Function2;ILjava/lang/Object;)Lkotlinx/coroutines/channels/ReceiveChannel;
public static final fun toChannel (Lkotlinx/coroutines/channels/ReceiveChannel;Lkotlinx/coroutines/channels/SendChannel;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
public static final fun toCollection (Lkotlinx/coroutines/channels/ReceiveChannel;Ljava/util/Collection;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
public static final fun toList (Lkotlinx/coroutines/channels/ReceiveChannel;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
public static final fun toList (Lkotlinx/coroutines/channels/ReceiveChannel;Ljava/util/List;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
public static final synthetic fun toList (Lkotlinx/coroutines/channels/ReceiveChannel;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
public static synthetic fun toList$default (Lkotlinx/coroutines/channels/ReceiveChannel;Ljava/util/List;Lkotlin/coroutines/Continuation;ILjava/lang/Object;)Ljava/lang/Object;
public static final fun toMap (Lkotlinx/coroutines/channels/ReceiveChannel;Ljava/util/Map;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
public static final synthetic fun toMap (Lkotlinx/coroutines/channels/ReceiveChannel;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
public static final synthetic fun toMutableList (Lkotlinx/coroutines/channels/ReceiveChannel;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
Expand Down
1 change: 1 addition & 0 deletions kotlinx-coroutines-core/api/kotlinx-coroutines-core.klib.api
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1142,6 +1142,7 @@ final suspend fun kotlinx.coroutines/yield() // kotlinx.coroutines/yield|yield()
final suspend inline fun <#A: kotlin/Any?, #B: kotlin/Any?> (kotlinx.coroutines.flow/Flow<#A>).kotlinx.coroutines.flow/fold(#B, crossinline kotlin.coroutines/SuspendFunction2<#B, #A, #B>): #B // kotlinx.coroutines.flow/fold|[email protected]<0:0>(0:1;kotlin.coroutines.SuspendFunction2<0:1,0:0,0:1>){0§<kotlin.Any?>;1§<kotlin.Any?>}[0]
final suspend inline fun <#A: kotlin/Any?> (kotlinx.coroutines.channels/BroadcastChannel<#A>).kotlinx.coroutines.channels/consumeEach(kotlin/Function1<#A, kotlin/Unit>) // kotlinx.coroutines.channels/consumeEach|[email protected]<0:0>(kotlin.Function1<0:0,kotlin.Unit>){0§<kotlin.Any?>}[0]
final suspend inline fun <#A: kotlin/Any?> (kotlinx.coroutines.channels/ReceiveChannel<#A>).kotlinx.coroutines.channels/consumeEach(kotlin/Function1<#A, kotlin/Unit>) // kotlinx.coroutines.channels/consumeEach|[email protected]<0:0>(kotlin.Function1<0:0,kotlin.Unit>){0§<kotlin.Any?>}[0]
final suspend inline fun <#A: kotlin/Any?> (kotlinx.coroutines.channels/ReceiveChannel<#A>).kotlinx.coroutines.channels/toList(kotlin.collections/MutableList<#A> = ...): kotlin.collections/List<#A> // kotlinx.coroutines.channels/toList|[email protected]<0:0>(kotlin.collections.MutableList<0:0>){0§<kotlin.Any?>}[0]
final suspend inline fun <#A: kotlin/Any?> (kotlinx.coroutines.flow/Flow<#A>).kotlinx.coroutines.flow/collect(crossinline kotlin.coroutines/SuspendFunction1<#A, kotlin/Unit>) // kotlinx.coroutines.flow/collect|[email protected]<0:0>(kotlin.coroutines.SuspendFunction1<0:0,kotlin.Unit>){0§<kotlin.Any?>}[0]
final suspend inline fun <#A: kotlin/Any?> (kotlinx.coroutines.flow/Flow<#A>).kotlinx.coroutines.flow/collectIndexed(crossinline kotlin.coroutines/SuspendFunction2<kotlin/Int, #A, kotlin/Unit>) // kotlinx.coroutines.flow/collectIndexed|[email protected]<0:0>(kotlin.coroutines.SuspendFunction2<kotlin.Int,0:0,kotlin.Unit>){0§<kotlin.Any?>}[0]
final suspend inline fun <#A: kotlin/Any?> (kotlinx.coroutines.flow/SharedFlow<#A>).kotlinx.coroutines.flow/count(): kotlin/Int // kotlinx.coroutines.flow/count|[email protected]<0:0>(){0§<kotlin.Any?>}[0]
Expand Down
16 changes: 9 additions & 7 deletions kotlinx-coroutines-core/common/src/channels/Channels.common.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -162,7 +162,8 @@ public suspend inline fun <E> ReceiveChannel<E>.consumeEach(action: (E) -> Unit)
}

/**
* Returns a [List] containing all the elements sent to this channel, preserving their order.
* Consumes the elements of this channel into the given [destination] mutable list.
* If none is provided, a new [ArrayList] will be created.
*
* This function will attempt to receive elements and put them into the list until the channel is
* [closed][SendChannel.close].
Expand All @@ -172,6 +173,8 @@ public suspend inline fun <E> ReceiveChannel<E>.consumeEach(action: (E) -> Unit)
* 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].
Expand All @@ -183,16 +186,13 @@ public suspend inline fun <E> ReceiveChannel<E>.consumeEach(action: (E) -> Unit)
* // sends elements to it, and closes it
* // once the coroutine's body finishes
* val channel = produce {
* values.forEach { send(it) }
* values.forEach { send(it) }
* }
* check(channel.toList() == values)
* ```
*/
public suspend fun <E> ReceiveChannel<E>.toList(): List<E> = buildList {
consumeEach {
add(it)
}
}
public suspend inline fun <T> ReceiveChannel<T>.toList(destination: MutableList<T> = ArrayList()): List<T> =
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, we can't check that the return value will be used now. Maybe unsplit it back, and do not return List on one of the destination overload?

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've attempted this, but cross-referencing overloads clutters the documentation quite a bit, and also, Flow.toList already follows the single-overload approach: public suspend fun <T> Flow<T>.toList(destination: MutableList<T> = ArrayList()): List<T>, so it's more internally consistent to have a single overload.

I'm wondering how real the risk of calling toList() on a channel and forgetting to obtain the resulting list is. Do you have an opinion on this?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We could in theory check it on flow if people do this mistake...

However, having the return value also allows chaining, and that tips the scale for me.

So let's keep your current version.

Copy link
Contributor

@murfel murfel Sep 10, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I wonder if people mistakenly do flow.toList() instead of flow.collect() to trigger flow execution for its side effects.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@sandwwraith, what do you think, is it more idiomatic to have fun X.toList(): List<T> + fun X.toList(destination: MutableList<T>): Unit or just fun X.toList(destination: MutableList<T> = ArrayList<T>): List<T> from the point of view of having an excessive return value?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it is better to have it as in the stdlib now: fun X.toList(): List<T> + @Ignorable fun <C: Collection/MutableList> X.toCollection/MutableList(destination: C): C. Generic argument in the second function allows you to collect result to ArrayList or HashSet, for example.

consumeEach(destination::add).let { destination }

@PublishedApi
internal fun ReceiveChannel<*>.cancelConsumed(cause: Throwable?) {
Expand All @@ -201,3 +201,5 @@ internal fun ReceiveChannel<*>.cancelConsumed(cause: Throwable?) {
})
}

@Deprecated("Preserving binary compatibility, was stable", level = DeprecationLevel.HIDDEN)
public suspend fun <T> ReceiveChannel<T>.toList(): List<T> = toList(ArrayList())
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Newline police

36 changes: 34 additions & 2 deletions kotlinx-coroutines-core/common/test/channels/ChannelsTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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)
Expand Down