Does anybody know good ways to avoid following issue.

fun main() = runBlocking {
    val f = channelFlow {
        launch {
            while (true) {
                offer(1)
                delay(500)
            }
        }

        awaitClose {
            println("f1 closed")
        }
    }

    val b = f.broadcastIn(this).apply {
        /**
         * If uncomment the following code, Exception below will be raised.
         *
         * java.lang.IllegalStateException: Another handler was already registered 
         *
         * You cannot onClose callback twice for same SendChannel.
         *
         * There is already onClose callback in this code: awaitClose callback inside channelFlow builder
         *
         */
        // invokeOnClose { println("b1 closed") }
    }

    launch {
        b.openSubscription().consumeAsFlow().take(5).collect {
            println(it)
        }
        b.cancel()
    }.join()
}

As I commented in the inline comment,

invokeOnClose

on the BroadcastChannel which is issued by broadcatIn terminal function on the Flow instance created by channelFlow builder would throws

java.lang.IllegalStateException Another handler was already registered

. Though It is possible to understand soon with this short sample code, it's sometimes hard to know whether the Flow instance is created by channelFlow when you

broadcastIn

on it.

Don't use

invokeOnClose

, use

onCompletion

on the

Flow

.

Ah, I see. Thanks.