Hi, my colleagues and me had a discussion around a...

# coroutines

[Lukas K-G]

Hi,
my colleagues and me had a discussion around a callback flow that was not cancellation-cooperative:

internal fun ScheduledAdItem.adItemStatusAsFlow() = callbackFlow {
    trySendBlocking(adItemStatus)
    val statusListener = AdItemStatusListener { _, status -> trySendBlocking(status) }
    addStatusListener(statusListener)
    awaitClose { removeStatusListener(statusListener) }
}

We figured that this is likely due to the trySendBlocking which does a runBlocking.

We then pivoted to the following:

internal fun ScheduledAdItem.adItemStatusAsFlow() = callbackFlow {
    send(adItemStatus)
    val statusListener = AdItemStatusListener { _, status -> launch { send(status) }}
    addStatusListener(statusListener)
    awaitClose { removeStatusListener(statusListener) }
}

Which we then figured that it might not guarantee the order of events if run on a coroutine dispatcher with multiple threads.

What would be the recommended way to implement a cancellation cooperative callback flow with guaranteed event order?

Thanks in advance!

[kevin.cianfarini]

Channel.send is fair, meaning that the first coroutine to invoke send will be the first to actually send the element, regardless of if its suspended or not. https://kotlinlang.org/docs/channels.html#channels-are-fair

[kevin.cianfarini]

So, it should preserve order.

[Lukas K-G]

thanks for the fast response. 🙇
but what about the launch itself? wouldn't this suspend and potentially be not fair on a say default coroutine dispatcher?

[Lukas K-G]

such as in this example?

[kevin.cianfarini]

It's possible I suppose. You could use launch(start=UNDISPATCHED) to ensure the same fairness guarantees as the channel offers.

[Lukas K-G]

Ah, interesting, I didn't know about that. Would this be the recommended way to do it?

[kevin.cianfarini]

It depends on your use case. I have very rarely used undispatched, but it comes in handy sometimes.

[ephemient]

why trySendBlocking in a launch anyway? if you're in a suspend scope, just use trySend or send itself

[ephemient]

but yeah launch depends on the dispatcher and doesn't guarantee fairness in general

[Lukas K-G]

The trySendBlocking is not in a launch, only the send is.

[Lukas K-G]

(or am I missing something? 😅)

[kevin.cianfarini]

If I had to guess, the callback for AddItemStatusListener is synchronous

[kevin.cianfarini]

So a launch is necessary.

[Lukas K-G]

AddItemStatusListener is synchronous, sorry for the missing information.

[ephemient]

oh I mixed up the two versions in the original post then

[ephemient]

the Undiapatched dispatcher does start running the coroutine in the caller, so it should do the trick for this case

[ephemient]

however it also means the coroutine will resume in the context of wherever resumed it, which could be some random thread

[Lukas K-G]

Any thoughts on cancellation in this case? The launch would no longer suspend but the send would still, so should be good?

[kevin.cianfarini]

I think generally this is made more complicated by running it on a mulithreaded dispatcher. If you can avoid that things would be simpler.

[Lukas K-G]

Hmm, yea, we would run it on the main dispatcher anyways most likely but it doesn't feel good to define this in a way it only works if called from the correct context.

[Vilgot Fredenberg]

the Undiapatched dispatcher does start running the coroutine in the caller, so it should do the trick for this case

however it also means the coroutine will resume in the context of wherever resumed it, which could be some random thread

But start = CoroutineStart.UNDISPATCHED should work without that caveat?

[ephemient]

oh oops

[ephemient]

yes, I meant the Unconfined dispatcher. the Undispatched start mode does not do that

[ephemient]

somehow I missed the start= in the reply and then forgot the tiny difference in name

[ephemient]

sorry, maybe I need more coffee first

[Vilgot Fredenberg]

It's a really unfortunate naming collision

[franztesca]

The correct solution IMHO if you want to retain all the events from the callback is to use trySend (not blocked!) to send the event to the callbackFlow and make the underlying channel buffered (UNLIMITED!) with the .buffer operator after callbackFlow

[Lukas K-G]

Thanks for all the input, this is what we settled for in the end.
It seems like this is not cooperative for cancellations as well so we added a yield after collection of the flow to ensure we don't process any values in case the coroutine was cancelled.

[Lukas K-G]

(we didn't set the buffer to UNLIMITED though cause we know that the number of callbacks is << 64 which is the default buffer size)