Hi folks I have a Flow that was created using `BroadcastChan

Question

Hi folks I have a Flow that was created using `BroadcastChannel openSubscriber consumeAsFlow ` and I use at the flow `launchIn Scope ` that s return a Job but I think that when a cancel this job the Channel is not being canceled when I use something like `Flow first` the channel is cancelled How can I cancel a Flow that canceling it will cancel the Channel too

octylFractal · Answer

you ll have to manually glue that together because by default I think it cancels the subscription which is reasonable

octylFractal · Answer

ah no

octylFractal · Answer

`BroadcastChannel asFlow ` is probably what you should be doing if you want that semantic

octylFractal · Answer

hmm no that also is 3 If the flow consumer fails with an exception subscription is cancelled

SrSouza · Answer

So something like this should do the unsubscription right ```val channel = ConflatedBroadcastChannel lt Int gt GlobalScope launch for i in 1 20 channel send i delay 500 val job = channel asFlow onEach println it launchIn GlobalScope delay 5000 job cancel CancellationException delay 5000 ```

octylFractal · Answer

it will cancel the subscription but not the `channel` itself

SrSouza · Answer

I was testing something like this and the `c` channel was not getting closed for receiving but when I use something like `first ` he get closed ```val c = channel openSubscription val job = c consumeAsFlow onEach println it launchIn GlobalScope delay 5000 println c isClosedForReceive job cancel CancellationException println c isClosedForReceive delay 5000 ```

octylFractal · Answer

you would need to check for cancellation in the flow it does not check for it itself

SrSouza · Answer

I m asking this for not have a Channel leak here Because I test it and the Flow do not run a after the `job cancel`

SrSouza · Answer

About this snippet above the `BroadcastChannel asFlow ` when using `job cancel ` cancel the subscription and the `openSubscription consumeAsFlow ` don t

octylFractal · Answer

sorry to be misleading I think they re nearly the same

octylFractal · Answer

yea okay I see what s happening you re checking it way too fast

octylFractal · Answer

if I do ``` val job = c consumeAsFlow onEach println it launchIn GlobalScope delay 5000 println c isClosedForReceive job cancel CancellationException println c isClosedForReceive delay 1000 println c isClosedForReceive ``` then the last one prints `true`

octylFractal · Answer

`job cancel` doesn t wait for the other end to receive the message

octylFractal · Answer

it seems to be a coinflip on when it becomes visible probably due to memory semantics of the JVM

SrSouza · Answer

I think it is unsubscribing but the Channel is not close yet be cause the Flow get closed before consume the last value from the Channel

SrSouza · Answer

With this snippet ```val c = channel openSubscription val job = c consumeAsFlow onEach println it launchIn GlobalScope delay 5000 println c isClosedForReceive job cancel CancellationException println c isClosedForReceive delay 2000 println c isClosedForReceive delay 2000 c consumeAsFlow onEach println it launchIn GlobalScope delay 1000 println c isClosedForReceive ```

SrSouza · Answer

I got this result ```1 2 3 4 5 6 7 8 9 10 false false false false 11 true```

octylFractal · Answer

yes that seems right

octylFractal · Answer

kind of weird that canceling the receive end doesn t drop all the items

octylFractal · Answer

seems like this might be a bug I m honestly not sure

octylFractal · Answer

I think <https github com Kotlin kotlinx coroutines issues 1265> might be related