Hi i am new in coroutines 🙂 i have a simply question.. from viewModel i get data from server with retrofit, viewModel -> useCase -> Repository... retrofit. When create coroutine i use the classic viewModelScope.launch { ... in some book i read that for network i have to use or better to use

viewModelScope.launch( <http://Dispatchers.IO|Dispatchers.IO>)

is it correct ?

If i have this method

fun wait(){
    val timeout= TimeHelper.getNow() + timeoutInterval // <- part of class constructor
    if(processor.getLag() > 0){
        val currentTIme = TimeHelper.getNow() 

        if(currentTime >= timeout){ System.exit(42) }
        Thread.sleep(someTime)
    }
}

And want to use a coroutine

runBlocking

and

delay

, would it be something like

fun wait() = runBlocking{
    launch { delayMethod() }
    if(processor.getLag() > 0){
        val currentTIme = TimeHelper.getNow() 

        if(currentTime >= timeout){ System.exit(42) }
    }
}

suspend fun delayMethod(){
  delay(someTime)
}

What’s currently the best way to check for cancellation inside a suspend fun ?

coroutineScope { ensureActive() }

?

i' am studing Flow but i don/t have example of real use now.. ok i can use some public api but i need some continous stream to try operator, zip, merge.. etc. How can use or find some data stream.. such as a socket?

could you annotate a method that should not be called within a coroutine so that you get a compile-time error?

Hi everyone. If I would write this method:

suspend fun timeDifference() : Int {
   val x = System.currentTimeMillis()
   delay(5.toDuration(DurationUnit.SECONDS)
   val y = System.currentTimeMillis()
   return TimeUnit.SECONDS.ofSecond(y - x)
}

and then write a test for this:

@Test
fun test() = runTest {
   val result = timeDifference()
   advanceUntilIdle()
   printLn(result)
}

Is it true that the output would be inconsistent? I tried it a lot of times and it doesn’t print the “predicted result” (in this case 5 seconds) but rather some other number.

Hi, just starting to learn Mutex in Kotlin. I have a situation where I have multiple instances of a class that will work on multiple threads. And the work is a heavy computation work that takes more than 10 seconds. I want the threads to add items to the same SynchronizedList using Mutex. The work, i.e. compute() method will be a suspendable function working as a coroutine. Should I create a single, shared Mutex instance for all instances that do the heavy work? I found an example for a single instance doing work using coroutines and the Default Dispatcher, but I am not so sure about multiple instances. Any help would be appreciated.

Hi! I have an inline function that is binding the state to a lifecycle owner.

inline fun <A> Flow<A>.bindState(
    lifecycleOwner: LifecycleOwner,
    crossinline bind: suspend (A) -> Unit,
) =
    lifecycleOwner.lifecycleScope.launchWhenStarted {
        collect(bind)
    }

Recently I've updated the coroutines library to 1.6.4 (from 1.5.2) and looks like

Flow.collect(crossinline action: suspend (value: T) -> Unit)

doesn't work with the newer version of coroutines dependencies.

Flow.collect()

now takes a

FlowCollector

functional interface argument instead of a

suspend (T) -> Unit

. Any suggestions of how I can still keep this as an inline function?

I'm trying to run a simple program that makes an HTTP call using retrofit but for some reason it is still running even after the call is finished and I can't figure out why It finishes after about a minute, but the call is done after a few seconds

fun main() = runBlocking {
    val service = createFooService()
    val response = withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
        service.foo()
    }
    println(response)
}

interface FooService {
    @POST("foo")
    suspend fun foo(): FooResponse
}

dependencies {
    implementation("com.jakewharton.retrofit:retrofit2-kotlinx-serialization-converter:0.8.0")
    implementation("com.squareup.okhttp3:okhttp:4.10.0")
    implementation("com.squareup.retrofit2:retrofit:2.9.0")
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core-jvm:1.6.4")
    implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.4.1")
}

Hi all! I've a question more philosophy than coroutines.. Having functions that does not are declared as suspend but inside them there is a scope.launch is a good practice or is better to do the scope.launch{} outside of the function but on each method invocation so each one can decide the scope to run it? For example:

fun start() {
        coroutineScope.launch {
            socketIoServer.startAsync()
        }
    }

//Another class
socket.start()

or

fun start() {
   socketIoServer.startAsync()
 }
//Another class

scope.launch{ socket.start() }

I have a launch {}, which inside i execute some blocking code which is listening for a single message from a pipe. Is there a way to notify/wait when the code will be suspended at the moment it started listening on the pipe? Thanks in advance for any help.

I got a coroutine test, where I use the

TestScope

provided inside the

runTest{}

function to pass it to a

launchIn

function to create a hot StateFlow. Problem is, the test this way never ends since the coroutineScope always has something to do therefore

runTest

times out and fails my test. I’ve “fixed” this by adding the line

this.coroutineContext.cancelChildren()

as the last line of my test after everything I’ve asserted was correct and so on. Is this considered okay or should I really be careful with something like this? Feels like I’m doing something in a way which doesn’t solve the root cause but patches it, but could be wrong.

Trying to test a

okhttp3.Authenticator

which unfortunately isn’t suspending and I am in there using

runBlocking

to bridge this gap. My problem is as such: In the body of that authenticator, everything is wrapped with a

runBlocking

but that one in turn calls a suspending function from another service. That function in turn, calls other suspending functions which at some point are using the CoroutineContext given from TestScope.backgroundScope (Or even TestScope itself). This means that then runBlocking is waiting for that to run, and the test is hanging waiting for

runBlocking

to finish at the same time, not giving me an opportunity to call

runCurrent()

or something like that to make the coroutine actually run. I guess this brings me to the question, should I be going with an UncofinedTestDispatcher here? I am kinda scared of using that usually since I don’t have good control of what’s going on, but I don’t know if that’s the more “appropriate” way to run tests with

runTest {}

. I usually default to not using UnconfinedTestDispatcher unless really needed, is this a bad habit?

Hi folks, I have a question as a relatively newbie in the coroutine world. I'm trying to refactor this piece of code:

val oas = specs.map {
    async {
      try {
        client.get(it).body<OpenApiSpec>() // Ktor HTTP client
      } catch (e: ClientRequestException) {
        null
      }
    }
  }.awaitAll()

I'd like to abstract away the exception handling, so I've tried something like this:

suspend inline fun <reified T> HttpResponse.bodyOrNull(): T? = try {
  body()
} catch (e: ClientRequestException) {
  null
}

val oas = specs.map {
  async {
    client.get(it).bodyOrNull<OpenApiSpec>()
  }
}.awaitAll()

But instead of swallowing the exceptions and returning null as in the example above, I get the exception bubbling up. I suspect I'm not yet fully understanding the exception handling model for coroutines, could anybody help shed some light? Thanks in advance. (Btw, I obtain the same result if I wrap the try/catch with

coroutineScope

)

when writing a custom Dispatcher, is there a way to tell IntelliJ to not show warnings about

Possibly blocking call in non-blocking context could lead to thread starvation

Dispatchers.IO somehow manages to make such warnings go away, I want to make those warnings go away for my custom dispatcher too

launch(Dispatchers.CUSTOM) {
   println(Thread.currentThread())
   Thread.sleep(5_000) <<-- warning: Possibly blocking call ...
   delay(5_000)
   println("done!!!")
}

Hi! I have a question about coroutines support while targeting Kotlin/JS platform. I just saw https://github.com/Kotlin/kotlinx.coroutines/issues/3433#issuecomment-1340924559 recently, so it means coroutines in JavaScript have been abandoned in general, and it's better to use plain `Promise`s?

1. Is anyone running coroutine async stack traces in production? 2. Do coroutine async stacktraces work for js?

How can I test timings of a

Flow

? I am trying to do with turbine and `TestCoroutineScheduler.currentTime`:

runTest {
  val myFlow: Flow<Int> = flow<Int> {
    (1..5).forEach {
      delay(100)
      emit(it)
    }
  }
  myFlow.test {
    repeat(5) {
      awaitItem()
      println(testScheduler.currentTime)
    }
    awaitComplete()
  }
}

The described approach does not work: For each run my test prints different/random values, e.g. "300,500,500,500,500 " etc. I have tried using the

myFlow.flowOn()

operator with both test dispatchers available.

I’m a bit confused as how to download a file with retrofit (okhttp) and coroutines so I declare the api

interface SpeedApi {
    @Streaming
    @GET
    suspend fun download(@Url url: String): Response<ResponseBody>
}

val response = speedApi.download("...")
response.body.use {
   it.byteStream().copyTo(...)
}

works, but the stream copying makes that blocking, and therefore not cancellable

should I roll my own copying like with manual cancellation check like this?

body.use {
    it.byteStream().use { iss ->
        var bytesCopied: Long = 0
        val buffer = ByteArray(DEFAULT_BUFFER_SIZE)
        var bytes = iss.read(buffer)
        while (bytes >= 0 && coroutineContext.isActive) { <-------------
            ...
            bytesCopied += bytes
            bytes = iss.read(buffer)
        }
        bytesCopied
    }
}

I do get that cancellation is cooperative etc etc — just looking for a idiomatic-retrofit way

Use https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/run-interruptible.html

We have a case of a coroutine that stops running in a Ktor server we would like help with debugging. This happens sporadically on some of our machines. The machine doesn’t seem to lack resources (CPU / memory).

class KafkaConsumer {

   private val scope = CoroutineScope(<http://Dispatchers.IO|Dispatchers.IO>)

   fun onMessage(message: KafkaMessage<K, V>?) {
      print("before launch") // this always shows
      scope.launch() {
        print("in coroutine") // this stops showing sometimes
        process() // Async operation that might take a second or two but also with some small runBlocking inside
      }
   }

}

Inside of

process

we make use again of the IO dispatcher to make an HTTP call. Are there debugging tools / techniques we could use? Would using a separate dispatcher inside

process

help with isolating the problem? Thanks

I'm on JVM, I have two coroutines that needs to access a mutable state exclusively. I can choose between using

synchronized

or a

Mutex

. Both work. Which one should I use? On one side,

synchronized

can block the thread of a coroutine but it's much faster, on the other side

Mutex

never blocks the thread but it's slower. Is there a preferred option, in general? Thanks

If i have a SharedFlow of sealed class, can i subscribe to just specific event? If so how?

Still after years of using coroutines, sometimes my brain suddenly questions itself. So probably a dumb question:

withContext(<http://Dispatchers.IO|Dispatchers.IO>) { new { ... } }

I have DB code from Exposed (

new

entity creation) that is not suspending and has this parameter:

init: T.() -> Unit

, which as you can see does not allow suspension. If I now really need to call a suspended function, I would usually do so before the

new

block and reference the result. Now I thought, what happens if I just call

runBlocking

with my suspending code inside

new

? Will it “block more” this way, or be the same since the

new

block has no suspending capabilities anyway?

I'm using a try/catch on a function call from a library which is doing some heavy nested and recursive coroutines with a bunch of callbacks mixed from our project and the library project and am getting this error. Is there any way for me to figure out what coroutine this is and what class is the owner of it?

Creates a supervisor job object in an active state. Children of a supervisor job can fail independently of each other Found the above statement on the doc of supervisor job, It is quite confusing for me. Like the supervisor job is used in the context of child coroutine. "Children of a supervisor job" where in practice we use a supervisor job on the child coroutine

Why isn't this used?

withContext(CoroutineExceptionHandler { ... }) {

}

I feel like it's a very confusing system as it is now. If I had to explain it to someone else I would say that uncaught exceptions first try to go up to parent job somehow. Otherwise they look for a CoroutineExceptionHandler falling back on system unhandled exception handler. With runBlocking be a special case that ignores CoroutineExceptionHandler despite being "root". Supervisor scope being another special case where all direct children are "root". Finally you can: CoroutineScope(CoroutineExceptionHandler {}) but it will be overwritten if you provide one to launch. What is the reason it doesn't just use the context's CoroutineExceptionHandler and have scope constructors add a default handler that delegates to system unhandled? (so every context has a CoroutineExceptionHandler and a Job) Also is withContext(currentCoroutineContext()) the same as coroutineScope at the end of the day?