Is there a reason <trySendBlocking()> is JVM only? How can I achieve this behavior in Kotlin/Native?...

# coroutines

[Jeff Lockhart]

Is there a reason trySendBlocking() is JVM only? How can I achieve this behavior in Kotlin/Native? My use case is adapting a callback API to Flow, without potentially dropping emissions as trySend() could.

val flow = callbackFlow {
    val job = addListener {
        trySendBlocking(it) // unavailable in Kotlin/Native
    }
    awaitClose {
        removeListener(job)
    }
}

[Jeff Lockhart]

I suppose I could buffer the flow:

val flow = callbackFlow {
    val job = addListener {
        trySend(it)
    }
    awaitClose {
        removeListener(job)
    }
}.buffer()

Helps avoid dropping emissions (up to the buffer capacity). But is there a way to do this blocking the send?

[Orhan Tozan]

I think you are looking for send() instead of trySend()?

[Jeff Lockhart]

send() is a suspend function and there's no coroutine in the listener callback, which is why I'm looking for the blocking equivalent trySendBlocking().

I suppose I could also launch a coroutine within the ProducerScope provided by the callbackFlow() builder for send() to suspend in.

val flow = callbackFlow {
    val job = addListener {
        launch { send(it) }
    }
    awaitClose {
        removeListener(job)
    }
}

[Orhan Tozan]

What is your usecase?

[Joffrey]

If you want to block, why not runBlocking { send(it) } ?

[bezrukov]

launching a coroutine for sending may lead to broken order of these events (e,g, if multithreaded dispatcher is used).
You can use

.buffer(capacity = Channel.UNLIMITED)

if you don't want to drop emission when exceeding buffer capacity

[bezrukov]

iirc, runBlocking is also JVM only

[Joffrey]

No it's also on KNative: https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/run-blocking.html

[Jeff Lockhart]

My use case is for converting some database change callback APIs to Kotlin Flows in a KMM shared module.

That's a great point about launched coroutines not guaranteeing ordering.

runBlocking() is available in Kotlin/Native. I looked at the JVM trySendBlocking() implementation, which uses runBlocking() internally. I copied the function to my KMM shared module and it has all the dependencies for Kotlin/Native, only warns about using InternalCoroutinesApi with ChannelResult. So doesn't seem to be a reason this function isn't available in Kotlin/Native.

@Throws(InterruptedException::class)
public fun <E> SendChannel<E>.trySendBlocking(element: E): ChannelResult<Unit> {
    trySend(element).onSuccess { return ChannelResult.success(Unit) }
    return runBlocking {
        val r = runCatching { send(element) }
        if (r.isSuccess) ChannelResult.success(Unit)
        else ChannelResult.closed(r.exceptionOrNull())
    }
}

[Orhan Tozan]

@Jeff Lockhart that's not what I really meant with an usecase. We need to know why you want to do that. What are you trying to achieve? Like a more high level example of what you want

[Jeff Lockhart]

I want to use Kotlin Flow APIs instead of callback APIs with the various uses of the database throughout our app. I'm writing extensions to provide these APIs. The use cases could be many different things from listening for database changes, live queries, external sync changes, etc. The APIs are all similar and converting them to Flows all use the same pattern.

[Orhan Tozan]

@Jeff Lockhart Ok, let me try it in another way: why do you not use trySend()?

[Jeff Lockhart]

I want to avoid trySend() potentially dropping emissions if the channel is full. trySendBlocking() blocks if this would happen with trySend().

[Orhan Tozan]

Why do you want to avoid potentially dropping emissions?

[Jeff Lockhart]

Because they may be emissions that collectors should handle, for example a database change that a subscriber should notify the user about, where not just the last change, but all changes should be handled.

[Orhan Tozan]

Ok thanks that clears it up, that gives me a higher view of your usecase

[Orhan Tozan]

I think if that is what you are trying to achieve, your channel should have an infinite buffer?

[Jeff Lockhart]

That may be the best option to guarantee delivery. Also avoids blocking as well.

It does seem like trySendBlocking() could be added to Kotlin/Native though. I created a YouTrack issue.

[Orhan Tozan]

I think you want to stay away from blocking methods as much as possible

[Jeff Lockhart]

The deprecated sendBlocking() function is available in Kotlin/Native, which functions similarly, only doesn't catch exceptions to return as a result.