Closing SubscriptionReceiveChannel inside consumeEach throws NPE

[patrickjuchli]

I have an NPE when closing a SubscriptionReceiveChannel of a BroadcastChannel inside a consumeEach block. For example:

fun main(args: Array<String>) = runBlocking<Unit> {
    val channel = BroadcastChannel<Int>(1)
    launch(CommonPool) {
        for (x in 1..5) {
            delay(100L)
            channel.send(x)
        }
        channel.close()
    }
    val sub = channel.openSubscription()
    sub.consumeEach {
        println(it)
        if (it == 2) {
            sub.close()
            // Adding return@consumeEach here doesn't help
        }
    }
    println("Done!")
}

consumeEach should probably check whether a channel is closed, I was surprised when I encountered this. In the meantime, this workaround helps:

for (value in sub) {
    println(value)
    if (value == 2) {
        sub.close()
        break // This is necessary
    }
}    

[elizarov]

Fixed in develop branch; added your example as a test.

[elizarov]

Released in version 0.18