Does `BroadcastChannel#openSubscription.consumeAsF...

# coroutines

[Abhishek Bansal]

Does BroadcastChannel#openSubscription.consumeAsFlow() completes when collected? If not how do I go about cancelling subscription?
I was trying to do this in my class so that client does not have to worry about cancelling the subscription

private fun getResponseChannel(): Flow<ContentResponse> {
        val channel = responseChannel.openSubscription()
        return channel
            .consumeAsFlow()
            .onCompletion {
                Timber.d("Cancelled")
                channel.cancel()
            }
    }

But it seems that onCompletion never gets called

[dekans]

It completes when channel is closed

[Abhishek Bansal]

okay.. thanks for your reply. My use case is that I wish to achieve RxJava’s PublishSubject like behavior. I wish to have a broadcast channel where multiple subscribers can come and go dynamically. is this possible with this setup at all?
I cannot close original channel I just wish to close new subscription when its done.

[streetsofboston]

Yup, this is possible and your code snippet looks good.

IIRC, onCompletion is called only when the upstream Flow (which basically is the broadcast channel) completes (error or plain closes).
You canceling the CoroutineScope in which you call collect on the Flow returned by getResponseChannel ('un-subscribing' in Rx terms), won't cause a call to onCompletion.