urgh, I'm drowning in a lack of understanding of F...

# coroutines

[groostav]

urgh, I'm drowning in a lack of understanding of Flow's vs RecieveChannels. I think what I want is a SharedFlow that does infact terminate. I've read both elizarov's medium articles. Can anyone straighten me out?

I have a function that produces a stream of data. it supports cancellation but most use cases shouldnt cancel it (eg: if I adapted this thing to a Flow() and somebody called first() , we should not cancel the remaining elements. The remaining elements should instead be produced and simply left un-consumed). You do have to kick it off. If you forget about it in most cases it'l complete normally but it some it will hang open, like a resource that needs to be disposed.

Should I make this thing into a ReceiveChannel or a Flow?

[Joffrey]

Sorry to hear that 😞 Do you have any specific question? (the edited question is good, thanks)

[groostav]

yeah sorry i accidentally hit enter before i finished typing

[groostav]

it seems like a really important question around "is this concept better expressed as a (~Cold) Flow or a (~Hot) Channel" is: does it need disposal semantics? if yes: implement as channel, if no: implement as a flow. Is it that simple?

[Joffrey]

What is your expectations regarding consumers of this stream? Do you have multiple consumers at the same time? If yes, should each of the consumers receive all elements, or do you expect elements to be dispatched among the consumers (one element goes to one consumer only)?

With ReceiveChannel, each element goes to a single consumer. If you have 2 concurrent consumers, each of them will receive approximatively half of the elements. So that might not be suitable for you.

With a ShareFlow , you can sort of make it terminate by sending a sentinel value, and using takeWhile after sharing the flow. Collectors will then stop. To support cancellation of upstream, you could use SharingStarted.WhileSubscribed when starting your shared flow, which cancels the upstream when there are 0 subscribers.

[andylamax]

I accidentally hit enter before I finish typing

That is slack experience at its finest

[andylamax]

In case of multiple consumers, what should the behaviour be when one consumer calls first and the other consumer calls collect/receive after quite a while??

[groostav]

Its safe to say its only 1 consumer

[andylamax]

If its one consumer, I see no reason to emit other elements after cancelation. Coz no other party would be interested. No?

[groostav]

What I mean by cancellation is cancelling the producer, since this is going to be a hot channel or a hot flow. The producer is intrinsically hot, i just don't want a call to first() to kill the producer. The producer is actually a process, so killing it would end up as kill -9 subprocess, which just, even if your not interested in its output, doesn't seem great.

[Tijl]

if you call first() and then later again (or collect or another terminator) then you have more than one consumer

[groostav]

yeah so I was using consumeAsFlow() vs receiveAsFlow() , where first() was throwing FlowAborted at the flow and the channel implementing it was marked as consumed.

[groostav]

if I take a step back, what I really want to do is implement a method suspend fun next(): T and have access to all the fancy combinators one would expect, like map and filter . From this angle, the concept of "is this flow finished" or "do you have multiple consumers" or "which is the start element" are fairly elegantly answered by the Iterator interface:
• "is this flow finished", => ask hasNext()
• "do you have multiple consumers" => I dont know, but if you want a fan-out build it yourself
• "which is the start element" => I don't know, if you wanted the start element you should've buffered it yourself.

[groostav]

mmm but I guess the "which is the start element" is a fairly racy question, and thus I need some kind of synchronization device on that first element

[Tijl]

SharedFlow

[groostav]

designing stream APIs is hard...

[groostav]

who knew? 😆