When using a `callbackFlow` is it necessary to cal...

# coroutines

[Tash]

When using a callbackFlow is it necessary to call close(exception) if an operation inside it throws? i.e.

callbackFlow {
   try {
      val result = somethingThatThrows() // Try-catch or just let it be?
      offer(result)
   } catch (exception) {
      close(exception)
   }
   
   awaitClose { ... } 
}

[Dominaezzz]

It.... depends. Your example doesn't look like it requires callbackFlow .
If the exception means that offer /`send` will stop being called, then yes, you must call close.

[Tash]

Gotcha. That was a pseudo-example. The real use case would look something like:

fun processedEvents(coordinator: EventCoordinator): Flow<Event> {
    return callbackFlow {
        val eventListener = object: EventListener {
            override fun onEvent(event: Event) {
                // Can throw
                val processedEvent = coordinator.processEvent(event)
                
                offer(processedEvent)
            }
        }

        coordinator.addEventListener(eventListener)

        awaitClose { coordinator.removeEventListener(eventListener) }
    }
}

[Tash]

If processEvent(...) throws in that example, the listener should be removed, i.e. awaitClose block must run

So, sounds like needs to be surrounded with try/catch + close 🤔

[Dominaezzz]

Ah, in that case you're right it needs to be surrounded.
You might want to do this to make sure you're not catching exception thrown from offer.

val r = runCatching { coordinator.processEvent(event) }
r.fold({ offer(it) }, { close(it) })

[Dominaezzz]

Or even better, I think you should do this.

fun processedEvents(coordinator: EventCoordinator): Flow<Event> {
    return callbackFlow {
        val eventListener = object: EventListener {
            override fun onEvent(event: Event) {
                offer(event)
            }
        }
        coordinator.addEventListener(eventListener)
        awaitClose { coordinator.removeEventListener(eventListener) }
    }.map { coordinator.processEvent(it) }
}

[Dominaezzz]

"better" is debatable, depending on how pure coordinator.processEvent is.

[Tash]

woah, using map is kinda nice. yeah, coordinator.processEvent internally calls a getter and maps the incoming argument to some new value. it does not inflict any side-effects.

[Tash]

so it sounds like awaitClose is triggered only when `close`/`cancel` have been explicitly called. i.e., letting the processEvent exception propagate would not have caused awaitClose to be called

[Dominaezzz]

It would, because the channel is closed when the downstream collector (i,e map) cancels/finishes collection (i,e throws exception)

[Tash]

oh yes, gotcha. i meant, without either the map or try/catch , awaitClose would not get called if processEvent threw

[Tash]

this is great, thanks so much