Hi guys, is this statement true, specifically regrading the part where they say: "The suspend functions that directly suspend a coroutine are in the standard library"?

Marking a function

suspend

gives it the capability to suspend the coroutine, but it will only actually suspend the coroutine if it internally calls another suspend function that suspends the coroutine. The suspend functions that directly suspend a coroutine are in the standard library. Among these are

suspendCoroutine()

and

suspendCancellableCoroutine()

. You will not often use these. They are most commonly used to convert non-coroutine APIs into suspend functions.

Some of the commonly used functions that indirectly suspend a coroutine are

withContext()

,

delay()

,

Job.join()

and

Deferred.await()

.

I found it in this Stackoverflow post.

I have a kotlin flow

flow { /* do some IO and emits some results */ }
  .collect { /* do some post processing */ }

If I introduce some retry logic using delay() into the body of the flow, it deadlocks if delay() gets called. In particular, it deadlocks after delay gets called, I think the first time I do some IO after delay returns.

flow { /* do some IO, retry if necessary, and emit some results */ }
  .collect { /* do some post processing */ }

But if I tell the flow to flow on some dispatcher, then it suddenly works again

flow { /* do some IO, retry if necessary, and emit some results */ }
  .flowOn(Dispatchers.Unconfined)
  .collect { /* do some post processing */ }

Why would that be the case?

I guess the reason for stateIn having an initialValue parameter is that it's needed for regular Flows, but for my case I am always using it after combining or mapping StateFlows. It's not a big deal when the output is a primitive type but when it's something more complicated creating an initial value that I know is never going to be used isn't great. Maybe there should be a version of those functions that outputs a StateFlow instead of Flow so it doesn't need to be converted with stateIn?

Can anyone help me?

// Data Layer
val locationUpdatesFlow = callbackFlow {
   // Registered Location Callback
   awaitClose { 
      // Cleanup 
   }
}.shareIn(
    scope = applicationScope, 
    sharingStarted = SharingStarted.WhileSubscribed(),
)

The reason, Why I am using SharedFlow in Data layer is, This flow can be collected in background also (by some other services....) while we collecting via UI. So, it will be performant that, the

locationUpdatesFlow

is SHARED. (I think so). Also, I need, this flow is running, if at least one subscriber is available. Otherwise, Kill this Flow. For this reason, I use

SharingStarted.WhileSubscribed()

Now In UI Layer (ViewModel)

val locationsState = DataLayer.locationUpdatesFlow
     .stateIn(
        scope = viewModelScope,
        sharingStarted = SharingStarted.WhileSubscribed(5000),
        initialValue = null
     )

The reason, Why I am using STATEFLOW here is, I need the location value in my UI as observable value In Screen, I use

viewModel.locationsState.collectAsStateWithLifecycle()

The Problem, I faced is below, 1. DataLayer.locationUpdatesFlow is NOT started....... While my UI is UP and RUNNING Is there any problem, that two nested

SharableFlow

(One on another) - (StateFlow on SharedFlow, in this case) Can any one explain this, And how to resolve this? (Also considering my Data Layer usecases/needs)

private suspend fun DefaultClientWebSocketSession.send(outgoingSubmission: OutgoingSubmission) {
    when (outgoingSubmission) {
        is OutgoingSubmission.Text -> {
            val outgoingEnvelope = OutgoingEnvelope.HandleMessage(
                ...
            )
            send(OutgoingEnvelope.HandleMessage.serializer(), outgoingEnvelope) <-------
        }
        ...
    }
}

detekt

flags this code with

suspend function uses CoroutineScope as receiver. [SuspendFunWithCoroutineScopeReceiver]

is this a false positive? what's wrong with it? Sure ktor's

DefaultClientWebSocketSession

is a coroutine scope but im pretty sure it's not meant to launch sends into it?

When an exception occurs in a coroutine, does it propagate the exception to the parent first, or does the cancellation propagate to its children first? I tested the code below, and the termination order always changes depending on the position of the

delay()

functions. So, I couldn’t tell which happens first, exception propagation or cancellation propagation. I realized that depending on the coroutine’s progress, it’s impossible to know which one will terminate first. But, I’m still curious whether, at the exact moment an error occurs, the coroutine internally attempts to propagate cancellation to its children first or propagate the exception to its parent first.

fun main(): Unit = runBlocking {
    val job1 = launch(CoroutineName("Coroutine1")) {

        val job2 = launch(CoroutineName("Coroutine2")) {
            val job4 = launch(CoroutineName("Coroutine4")) {
                println("Coroutine4 run")
                delay(600)
            }
            job4.invokeOnCompletion {
                println("Coroutine4 end")
            }
            val job5 = launch(CoroutineName("Coroutine5")) {
                println("Coroutine5 run")
                delay(600)
            }
            job5.invokeOnCompletion {
                println("Coroutine5 end")
            }
            println("Coroutine2 run")
        }
        job2.invokeOnCompletion {
            println("Coroutine2 end")
        }

        val job3 = launch(CoroutineName("Coroutine3")) {
            val job6 = launch(CoroutineName("Coroutine6")) {
                delay(600)
                println("Coroutine6 run")
            }
            job6.invokeOnCompletion {
                println("Coroutine6 end")
            }
            println("Coroutine3 run")
            delay(500)
            throw Exception() // here! exception occurs!
        }

        job3.invokeOnCompletion {
            println("Coroutine3 end")
        }
        println("Coroutine1 run")
    }

    job1.invokeOnCompletion {
        println("Coroutine1 end")
    }

}

I'm not sure where to ask this, so apologies if this is not the right channel. I'm asking here because my Googling suggested that this issue may somehow be coroutine related. I'm trying to debug some code in a suspend function (which happens to be part of a kotest spec), and I need to see the value of a variable,

messages

. However, IntelliJ is saying "'messages was optimised out". This is despite the fact that my code uses this the variable after the breakpoint.

val messages = msg.messages()
println(messages)                  // breakpoint is on this line
messages.size shouldBe 1

Is there any way to tell it to not optimize out a variable?

Hi all kodee greetings , I have a coroutine that has few children coroutines and suspend method calls. How to rethrow an exception thrown by one of the suspending method calls? Code example in 🧵

Came across some code in a PR that seems weird to me. Let me know if it's just me or if this is idiomatic coroutines code.

class MyCoolHandler{ 
var mContinuation: Continuation<Foo>? = null
suspend fun waitingForFoo(): Foo = suspendCoroutine { continuation -> 
// code
mContinuation = continuation
}
// some other class functions
}

mutableXOf

are not thread safe right? I have a mutable set which i populate from within multiple coroutines (thousands) and I am missing some items after my code is executed. I wrapped the

.add

calls from a single thread context scope and it seems to be working fine now. are there any thread safe versions of mutable set of? or better yet, are there any recommended approaches to this?

Say I want to run 2 api calls in parallel, then map their results together, and then save that to db. Question is, should I try to scope down the "

coroutineScope { .. }

length, or is it okay to have ot extend to the full function body? Whats more idiomatic?

When writing a suspend function, do we make any assumptions about whether or not that function is already being executed in a given scope? Say I'm writing a function which needs to do some file reading and then perform some UI stuff. Should I explicitly use

withContext

for every switch between UI and background? Currently I have:

suspend fun openView(path: Path): MyView {
    // Load data

    return withContext(Dispatchers.Swing) {
        // Create UI View
    }
}

Is there a more idiomatic way to achieve the result of asynchronously loading a bunch of file data from a list of paths? I've been using this:

val data: List<MyDataClass> = withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
    filePaths.map { path ->
        async {
            path.inputStream().use {
                // Read in file data as instance of MyDataClass
            }
        }
    }.awaitAll()
}

Hi guys, can someone comment on how true this comment is https://www.reddit.com/r/Kotlin/comments/1kko2s7/comment/mrytehj/?utm_source=share&u[…]m=web3x&utm_name=web3xcss&utm_term=1&utm_content=share_button? The commenter talked about why they wouldn't use ktor/coroutine, one of the issue they mentioned is that debugger doesn't follow the coroutine, but the carrier thread, which I have experienced before, another issue is that parameter will be marked as "optimised out", also the fact that code is "colored" with coroutine. I can't confirm about other issues that was mentioned, so I'm looking for everyone's opinion. Thanks in advance.

This is probably a stupid question, but is there anyway to wrap up this:

sealed interface FooResult
data object FooCancelled : FooResult
data class FooSuccess : FooResult

suspend fun foo(): FooResult = TODO()

val deferred = async { foo() }
deferred.cancel()
val result = deferred.await()

so that the last call will never throw

CancellationException

, instead returning

FooCancelled

? Without a try / catch (or equivalent) around

deferred.await()

?

Is it possible to have a MutableStateFlow that continuously produces values? I would like to zip it with another flow as I don't want the resulting flow to be limited by how often my MutableStateFlow gets mutated.

val foo = MutableStateFlow(0)
val bar = flow { ... }
foo.zip(bar) { ... } // I don't want the resulting flow to be limited by how often foo is updated.

Hey everyone, when I'm converting a callback based API to a

Flow

using

callbackFlow

, do I have to call

close()

after

trySend(...)

when the call back only emits once?

Don't know if this is the right channel, but does something an async task queue exist where I provide a coroutine scope in which callbacks can be scheduled to be executed sequentially or simultaneously?

Hi guys. Does anybody have an experience in creating observability for active coroutines at runtime (PROD)? I need to know how many active coroutines are in the thread pool. I see there is an

kotlin-coroutines-debug

library, which solves my requirements. However, I'm not sure about performance impact + if it is suitable to native builds (according to the docs - not really)... I would be very grateful for advice. :)

Is it possible to disable the automatic dump of coroutines at the end of tests when debug probes are enabled? I'd like to control the exact time at which it happens.

How would I encode interruptible code vs. non-interruptible code? Aka ask some Job nicely to cancel, but the job itself can say no, or await until an interruptible section has been reached.

I'm trying to figure out why this

first()

doesn't get cancelled by the timeout

Given the following snippet:

object MyObject {
    var x = 1
    init {
        GlobalScope.launch {
            x++
        }
    }
}

suspend fun main() = coroutineScope {
    println(MyClass.x.toString()) // output: 1
    delay(100L)
    println(MyClass.x.toString()) // output: 2
}

How to make

MyClass.x.toString()

print 2 without needing to have a delay statement? How should I change the init block?

Do all scopes get cancelled on it's own, when the application is terminated? I have a scope, that has basically same lifetime as the application. I'm wondering if I should bother closing it. I guess it will get canceled, on JVM shutdown or something?

DebugProbes.install()

is very slow! Here, it takes between ~1.5s and ~3s depending on the run, and it's called once for each Gradle module

What is the most idiomatic way of having a

MutableSharedFlow

(that acts as a trigger to refresh stuff), that has a default value (so syncs get triggered right away on start)? I find manually throwing in `trigger.emit(Unit)`after everyone is subscriber a bit awkwards Unless there is an API I'm not aware of, probably the best would be to have own subclass right? Like this maybe?

class TriggerFlow : Flow<Unit> {
    private val _trigger = MutableSharedFlow<Unit>(
        replay = 1
    )
    
    init {
        _trigger.tryEmit(Unit)
    }
    
    suspend fun trigger() {
        _trigger.emit(Unit)
    }
    
    override suspend fun collect(collector: FlowCollector<Unit>) {
        _trigger.collect(collector)
    }
}

How does this look? Is the

tryEmit

in init cool?

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!

are there any solutions for resumable coroutines that persist over application restarts? I saw this from nextjs yesterday and I would love to have something like this* on my backend *this = having coroutines run and not having to worry about server restarts (ie have

delay()

for some days and then continue the coroutine)

I'm aware of the

runTest

workaround for

withTimeout

using

withContext(Dispatchers.Default.limitedParallelism(1))

, but is there a workaround for

Flow.timeout

? I can

flowOn

, but I'd rather not force my entire Flow to a specific dispatcher if possible.

I've got a big JSON file that I want to read; for each entry in an array that is read lazily, I want to make a request to an external system; however, I don't want to send each request at once but use a maximum of x requests at once; I also don't want to evaluate more objects than I can currently send; is there an easy way to do that?