Quick ? on `BroadcastChannel<T>.asFlow()` ...

# coroutines

[taer]

Quick ? on BroadcastChannel<T>.asFlow() The docs state: `

If the flow consumer fails with an exception, subscription is cancelled.

I have a

broadcastChannel.asFlow().onCompletion{ printSomething } .map {  codeThatThrowsException }.catch { print MessageAboutError } . collect { collector}

The onCompletion fires, then the catch. Afterwards, the collect never sees a new message.. I just added the onComplete to validate. I expected the .catch to handle the error so the flow doesn't get canceled. I can easily fix the codeThatThrowsException but want to catch unexpected errors in the future. It looks like the downstream catch doesn't get a chance to handle the error prior to the upstream channel getting unsubscribed from

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

catch will catch errors flowing downstream, and can stop them from flowing further downstream, emit other items downstream, etc. You could use catch, for example, to “resubscribe” upstream when an error occurs.

[taer]

why does the upstream get canceled in this case?

[taer]

or rather, whats the means to catch errors, log them, and just not kill the flow?

[taer]

putting a token catch block in front of the map that throws doesn't fire as expected. Do I just need to try/catch the code in the "codeThatThrowsExceptions"?

[octylFractal]

if you need to catch errors without killing the flow, you need to try/catch that code

[taer]

ok.. That's the conclusion I'm coming to. I guess the exception bubbles both ways.. towards upstream in their emit call.. And then downstream for catch to operate on

[taer]

just validated w/ a simple non-channel flow

[octylFractal]

there is also retry which will do a re-collect on error, but the thing is that doing it outside of the operation means that all the catch/retry operator knows is that something threw an exception, and stopped all collection. retry will call the internal collect(FlowCollector) again with the same setup, but that may not result in perfect resumption of your code where it was, or you might miss some values

[taer]

yeah.. was hoping for a catch that eats the exception, and lets the collector not know.

[taer]

The retry is pretty magical. But in this case, the flow is hot, so there's not really much in terms of a re-do

[taer]

I did a quick non-channel flow and had the same results. Learning more and more. 🙂

[octylFractal]

it might be helpful to know that these two functions are essentially the same: https://gist.github.com/octylFractal/8b4bb4b6d133142908d7c6bd8ca75024

essentially the catch operator cannot see the body of a flow {} or map {}, so it cannot catch the operation and resume right after it -- there is no way for catch to "wrap" the error I have introduced here. instead, catch can only get the error after the entire method has be removed from the call stack, and it also tracks each emission to check if it came from downstream

[octylFractal]

retry is effectively wrapping a do { ...} while(precondition) around the whole outer try-catch

[taer]

interesting.. That actually helps a lot

[taer]

Thanks!

[taer]

So a map in that gist would be just a function call that gets wrapped inside the call the println in your example

[octylFractal]

yes

[octylFractal]

if you added a map { "Label: $it" } after the catch, it would essentially be println("Label: Something?"), etc.

[octylFractal]

if you add it before the catch, it's like:

val value = "Label: Something?"
try {
  println(value)
} catch (e) {
  downstream = e
  throw e
}
val value2 = "Label: " + if (Math.random() < 1.0) {
            error("Boo!")
        } else { "Another thing!" }
try {
  println(value2)
} catch (e) {
  downstream = e
  throw e
}

i.e., the error does get caught by the outer thing properly, but it still doesn't allow each map to be individually wrapped by catch