Fail-fast when trying to consume a channel multiple times

[elizarov]

Channels are designed to be used in fan-out fashion or in a pipeline fashion. Pipeline is conveniently performed with various operators like filter, map, etc (see PR #88) which are modeled after a Sequence. However, a sequence is a multi-shot abstraction, for the following is perfectly file:

val s: Sequence<T> = ...
s.filter { ... }.forEach { it.doSomething() }
s.map { ... }.forEach { it.doSomethingElse() }

However, a similar code will not work in the same way with a ReceiveChannel. ReceiveChannel is more like Iterator, than it is like a Sequence. Each element of the ReceiveChannel can be consumed only once. It is expected that this might be a common source of errors and misunderstanding, so the proposal is to make all the standard operators "fail-fast" when the channel is already being consumed by another operator. So, in the above example, if the types of s was to be replaced with ReceiveChannel<T>, then the invocation of s.map (which comes after s.filter) shall fail immediately.

P.S. It is a separate issue whether a truly Sequence-like (Rx-like) higher-order abstraction shall be introduced for channels.

[ZakTaccardi]

just want to confirm that the following would still work:

val channel: ReceiveChannel<T> = ...

channel.filter { .. }
    .map { .. }
    .consumeEach { .. }

[bdavisx]

I'm confused by the description, will consumeEach only be able to be called once? If so how would you do a fan-out?

[elizarov]

@bdavisx Thank you for observation. This indeed can be confusing and we should think on how to make it less confusing. They idea is that for fan-out you don't consume the channel. Instead, you simply invoke receive() to receive elements from it as you need.

[bdavisx]

Thanks and yes, very confusing because the current guide/examples fan out section doesn't use receive(), it uses multiple calls to channel.consumeEach().

[bdavisx]

I changed the doc to use an iterator instead of consumeEach() - and in the end it looks like the iterator uses receive(). I think this is a better example, but I'm not sure.

#406

[jcornaz]

Will a forEach extension function be provided in order to easily iterate over all elements, but not consume the channel?

[qwwdfsad]

@jcornaz probably no, because it will be an additional source of confusion.
Users (who didn't read the documentation carefully) may start to pick "random" method. because it's not clear from the name whether consumeEach and forEach have any semantic difference.

Usually you need consumeEach, so it's a reasonable "default" for the library

[jcornaz]

Isn't the confusion already here? I don't think it is particularity clearer that for(elt in channel) has semantic difference with channel.consumeEach.

However I agree that user may see the forEach in the auto-completion and use it without understanding the danger of not consuming the channel...

[elizarov]

The difference between "consuming" and "iterating in fan-out fashion" is indeed confusing and more design effort is needed to make it more explicit.

[dave08]

In regular collections we're all used to forEach being terminal and onEach not, why have consume in the first place and not reuse familiar concepts (apart from the suspending nature of consume which is anyways in the function's signature...)? Or maybe there's another aspect here that I didn't get?

[elizarov]

Initially consumeEach was called forEach, but we had to rename it for better integration with Rx where there is already a forEach member which accepts non suspending code block which means that you cannot easily do nice Kotlin coroutines stuff there, e.g. this does not work:

observable.forEach {
    delay(1000)
    println("Received $it")
}

But this works:

observable.consumeEach {
    delay(1000)
    println("Received $it")
}

We can revert this decision, rename it back to forEach, and find a different name for extensions on various reactive classes and interfaces.

[efemoney]

How about a receiveEach to augment consumeEach?. That might solidify the mental model of consume vs receive. Personally would've prefer forEach and onEach but they are taken like @elizarov says.

[qwwdfsad]

Status update: won't be fixed.

There are valid use-cases with a fan-out style (thus concurrent consumeEach calls) and this change will add a subtle difference between iteration and consumption.
Additionally, tracking a closed state is a way too intrusive change for channels