when using callbackFlow I need to clean up resources but when the job is canceled awaitClose isn't ...

# coroutines

[Luis Munoz]

when using callbackFlow I need to clean up resources but when the job is canceled awaitClose isn't being called. How do I invoke awaitClose on a cancellation of the job

fun listen(port: Int = 1337, nThreads: Int = 2) =  callbackFlow<String> {

                val ch = b.bind(port).coAwait { }
                for (obj in coReceiveChannel) {

                    send(obj)
                }

                  awaitClose {
                    coReceiveChannel.cancel()
                    ch.close()
                   }
            
      } // end of listen
      
job = listen().onEach { println(it) }.launchIn(scope)
        
job.cancel()   // doesn't call awaitClose

[Luis Munoz]

never mind came up with this:

[Luis Munoz]

coReceiveChannel.consumeAsFlow().onEach{
    offer(it)
}.onCompletion { close() }

[Zach Klippenstein (he/him) [MOD]]

It’s probably because awaitClose never gets a chance to even be executed before cancelling. for (x in channel) will loop until the channel is closed, so if that channel is never closed, but you cancel the flow, then the for will throw the CancellationException and awaitClose will never be executed. You could just wrap your loop in a try/finally to get the behavior you want. awaitClose is only intended for the use case where your flow has nothing else to do but sit around and wait to be cancelled.

[streetsofboston]

Also, it seems in your use-case, you don’t need a callbackFlow { … }. A plain flow { … } suffices.

And then wrap the for … loop inside a try - finally block as @Zach Klippenstein (he/him) [MOD] suggests.

[elizarov]

try/finally is the way to go.

[Luis Munoz]

thank you