ReceiveChannel.asFlow operator

[qwwdfsad]

A missing piece of the Flow puzzle is a connection between regular ReceiveChannel and Flow (note that we already have a pretty non-obvious ReceieveChannel.asFlux) with a proper migration path for channel operators.

For example, all operators on top of the channels are deprecated, but users have no clear migration path from their channel operators sequence to flow.

The main focus here is to provide clear consumption semantics and decide whether we want to introduce such primitive at all

[zach-klippenstein]

I've run up against this a few times, where I know I will only ever have a single collector (so the lack of multicasting is not an issue), but I still want to use the operators (or integrate with an API that requires Flows). This comes up the most often in unit tests. Another use case I have is that I want to use a regular (non-broadcast) Channel to send, because I don't want ArrayBroadcastChannel's "drop items when not subscribed" behavior, but I would still potentially like to be able to expose a Flow as the API for consistency and convenient access to operators (square/workflow#463).

If such an operator were to be added to the library, I think it would need to be either clearly named and/or clearly documented that multiple concurrent collectors would not get the same items. As long as that is documented, I think it would be fine as long as #1261 lands, since a lot of use cases would probably want to multicast the channel before doing anything else with it.

[elizarov]

The proposal is to name it consumeAsFlow and add additional precaution that repeated attempts to collect from the resulting flow produce IllegalStateException just in case. See

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

Lines 17 to 29 in bc34a2e

	/**
	* Represents the given receive channel as a hot flow and [consumes][ReceiveChannel.consume] the channel
	* on the first collection from this flow. The resulting flow can be collected just once and throws
	* [IllegalStateException] when trying to collect it more than once.
	*
	* ### Cancellation semantics
	* 1) Flow consumer is cancelled when the original channel is cancelled.
	* 2) Flow consumer completes normally when the original channel completes (~is closed) normally.
	* 3) If the flow consumer fails with an exception, channel is cancelled.
	*
	*/
	@FlowPreview
	public fun <T> ReceiveChannel<T>.consumeAsFlow(): Flow<T> = object : Flow<T> {

[fvasco]

The proposal is to name it consumeAsFlow

Should BroadcastChannel<T>.asFlow() be renamed to BroadcastChannel<T>.subscribeAsFlow()?

BroadcastChannel looks slightly different, should it be: openSubscription().consumeAsFlow()?

[elizarov]

@fvasco For BroadcastChannel a simple asFlow seems appropriate because there is good match between the source type (BroadcastChannel) and a target type (Flow) and representing the former as the latter is quite a natural transformation. It is only ReceiveChannel that requires a special attention to the fact that unlike all the regular flows, the plain channel can be received (consumed) just once.

[fvasco]

You are right, @elizarov,
I confused this use case with the example #1326 (comment) , in that case asFlow drops messages (and so on... see other issue).