Hey guys. At the moment I'm processing events in t...

# coroutines

[Ivan Pavlov]

Hey guys.
At the moment I'm processing events in the following way:

val channel = BroadcastChannel<Event>(10)
val receiveChannel = channel.openSubscription()
val flow = receiveChannel.consumeAsFlow()

flow.onEach(someProcessor::doOnEach)
    .catch { log.error(it) { "Error while processing event" } }
    .launchIn(coroutineScope)

In my case I have to just skip an event if an exception is thrown in someProcessor.doOnEach(event: Event) and continue processing of next events.
I can achieve what I want with

flow.onEach {
  try {
    someProcessor.doOnEach(it)
  } catch (e: Throwable) {
     //log
  }
}

but I feel like I'm missing something. Is there a better way to do this?

[octylFractal]

why does that feel like it's missing something? seems like a good way to handle it

[Ivan Pavlov]

Because it looks like there should be an operator which can do that for me, like catchAndResume 🙂

[Dominaezzz]

There is a thread about this catch and resume functionality. I'll share it if I can find it. But the gist of it was Flow doesn't really handle exceptions like that, as it wouldn't be "transparent" as the docs put it.

[Dominaezzz]

Turns out there's been loads of threads. This was the one I was referring to https://kotlinlang.slack.com/archives/C1CFAFJSK/p1568900693092300