collect methods for ReceiveChannel.

[konrad-kaminski]

Let's say I have a component, which reads a list of users from a database. A natural API could look like this:

interface UserRepository {
  suspend fun findUsers(): ReceiveChannel<User>
}

This way I can do sth with every user as they come. However, in some cases I'd like to get all users as a collection (e.g. a list). It would be convenient to have extension functions, which can receive all elements from a ReceiveChannel and create a list/set/map/sequence.

In the attached code you can find four of these. They are based on similar methods of Flux from Project Reactor. I created only basic variants of these methods, but if needed, I can create more involved ones (e.g. this version of collectMap).

The four extension functions signatures are:

suspend fun <E> ReceiveChannel<E>.collectList(): List<E>

suspend fun <K, E> ReceiveChannel<E>.collectMap(keyExtractor: (E) -> K): Map<K, E>

suspend fun <E> ReceiveChannel<E>.collectSequence(): Sequence<E>

suspend fun <E> ReceiveChannel<E>.collectSet(): Set<E>

[fvasco]

Hi,
some personal considerations.

For uniformity with kotlin.collections names the names should be: toList, toSequence, toSet.

collectMap interface is different from map extension function interface, in my personal opinion should be removed in favour of something like channel.map{ key to value}.toMap(), where:

fun <T, R> ReceiveChannel<T>.map(transform: (T) -> R): ReceiveChannel<R>

suspend fun <K, V> ReceiveChannel<Pair<K, V>>.toMap(): Map<K, V>

Finally collectSequence (toSequence) is eager, but

Sequence type that represents lazily evaluated collections.

http://kotlinlang.org/api/latest/jvm/stdlib/kotlin.sequences/index.html

[konrad-kaminski]

I've actually given some thoughts to the issue of naming these methods. I considered toList,etc. (in fact they are also used by Rx1/Rx2), but decided to use collect naming scheme. The reason is that these methods actually collect events into some data structure. toList would rather suggest that you want to convert one structure type into another (as it happens with kotlin.collections).

suspend fun <K, V> ReceiveChannel<Pair<K, V>>.toMap(): Map<K, V> seems fine to me and I can add it quite easily (though as collectMap for consistency).

As for the fun <T, R> ReceiveChannel<T>.map(transform: (T) -> R): ReceiveChannel<R> it's a bit different thing. We generally have to think about introducing similar operators to the ones available for CompletableFuture or various reactive implementations, and map is only one of them. Furthermore the question is do we want the transform function to be coroutine or not? What do you think @elizarov ?

collectSequence must create an eager sequence - it's not possible to create a lazy sequence backed by a coroutine (unless it's a restricted suspending function) - if we want this kind of structure we actually have to use... ReceiveChannel. :)

[cypressious]

If lazy collectSequence cannot be achieved, maybe it should be left out? A user cann still call collectList().toSequence() which would be the same thing except now it's clear that the operation is eager.

[konrad-kaminski]

@cypressious The reason I introduced collectSequence was exactly to avoid such constructions as collectList().toSequence().

[elizarov]

Support for channel-transformation function was already in my to-do list for quite a while and I'm glad it is not just a theoretical curiosity of mine, but indeed a practical issue. The naming of these functions is a tough call. Naming various filter and map function is no-brainer, but conversion functions offer some more food for thought, given that they work a bridge from asynchronous (and backpressure-capable) channels to the fully materialised collections.

The reason for thought is not just their names and semantics, but their piece in the larger scheme of things. Currently, Kotlin has Collection hierarchy (with List, Set, etc) for eager collections, and Sequence for lazily evaluated ones. Their APIs are carefully aligned and the source code of their many functions is actually auto-generated from a single template. We are also working on persistent collections as a part of kotlinx project. They would have the similar set of operations and we are looking for ways to unify their behaviour and code, too.

Now, Channel is yet another (4th) instance of the similar concept.

Actually, we might, one day, introduce concepts (or type classes) and a first-class Kotlin entity, thus shifting the task of generating the code for all those filter, map, etc functions using a defined template from the source-code generation tool to the compiler itself. This requires some care in terms of naming and semantics of functions that we make available now.

[fvasco]

toList would rather suggest that you want to convert one structure type into another

Not quite, collectList has the same behaviour of Sequence<T>.toList()

collectSequence must create an eager sequence

fun <T : Any> ReceiveChannel<T>.toSequence() = generateSequence { runBlocking { receiveOrNull() } }

[elizarov]

I'm sorry I'm dragging feet on this PR. It is also related to #88 and is a part of a large piece of work to provide the same set of operations for ReceiveChannel that are currently available for Sequence in Kotlin standard library. I really want to do this all in one big chunk as a part of one of the upcoming release.

[elizarov]

Method to save channel to various collections are now a part of pr-88 branch. The naming question between toXXX and collectXXX is still an open one. See discussion in #88