Hi, I'm new to coroutines and I would appreciate some clarification. I'm following different examples of basic coroutines and I'm seeing these two ways of structuring the code. What are the differences, pros and cons of them? Suspend funs to call:

suspend fun foo1() // function
suspend fun foo2() // function

First approach:

// sequential code

viewModelScope.launch { foo1() }

// more sequential code

viewModelScope.launch { foo2() }

// more sequential code}

Second approach:

viewModelScope.launch {
	// sequential code

	foo1()

	// more sequential code

	foo2()

	// more sequential code
}

I've also saw this second approach with

async

calls inside the main

launch

one. Thanks!

Is this a bug, tried invalidating caches still the same. Using

org.jetbrains.kotlinx:kotlinx-coroutines-core:1.4.2-native-mt

problem started yesterday... (Project builds fine)

Hi folks, now I'm diving into kotlin.coroutines to understand how Coroutines work internally. and I found

suspendCoroutineUninterceptedOrReturn

function. It looks like the lowest API I can reach. (since it says

throw NotImplementedError("Implementation of suspendCoroutineUninterceptedOrReturn is intrinsic")

) Is there any good document to understand how it works in bytecode level or anything else? Especially, I'd like to understand how

suspendCoroutineUninterceptedOrReturn

capture the current running coroutine. thanks you! (If it's not good to post that here, please let me know where to post 🙇 )

Hello 👋 I was just wondering what is the idiomatic way to launch some side effects from main coroutine whose return value doesn’t depend on it (i.e. side effect can take some time to complete and it is fine if that computation fail) so that it actually does not have to wait for its completion, e.g. given something simple like below

suspend fun doWork() = coroutineScope {
  val result = ...
  // do some other processing that calculates/processes result

  publishData(result) // optional stuff that may be slow, e.g. publish analytics, debug info, etc
  result
}

Thanks to structured concurrency if I launch another coroutine from the

doWork

, all the child coroutines (including

publishData

from above) have to complete before returning from the scope. I could obviously just do

GlobalScope.launch { publishData }

but that seems “wrong”. Wondering whether instead I should be using something like buffered

Channel

so publish is async and data is processed in another coroutine. Or maybe there is something simpler?

hi! how can i create a flow where the emitting part runs in a different thread / coroutine context ? assume i have an class which has a method getRemoteService() : Flow<Result>. when calling that method i want the method to immediately start consuming the web service so when i call flow.collect{} it is either already finished or still in the process of fetching. somehow with the flow builders i haven't found a way to do that. should i use in that case a channel directly?

They say that coroutine cancellation is cooperative. That's perfect when it's possible to check for it via

isActive

. But how to make it cooperative when you're dealing with code which doesn't include loops, i.e. when you don't have a right place to put the

isActive

check?

Hi can someone explain me why consuming the channel suspends forever?

private val packetChannel: Channel<ResponsePacket> = Channel(Channel.UNLIMITED)

    suspend fun sendPacketWithBatchResponse(packet: RequestPacket): List<ResponsePacket> {
        sendPacketWithAcknowledgement(packet)
        val packets = mutableListOf<ResponsePacket>()

        for (item in packetChannel) {
            Timber.e("here")
            packets.add(item)
        }

        // Unreached code
        return packets
    }

    // Another non-suspending function which is decoupled from the consuming function
    fun produce() {
        ...
        packetChannel.offer(packet)
    }

The goal is to offer multiple packets (~20) to the channel and when finished jump to the

sendPacketWithBatchResponse

and consume these packets all at once but

sendPacketWithBatchResponse

never returns and I don't know why

Hi folks, I was trying to reproduce a case where coroutines and thread local don’t play well together. I wanted to try out using the threadlocal in a coroutine without using the extension function asContextElement() to see it breaking :) Now, as you can see from the snippet I launch a bunch of coroutines in half the threads I spawn and I put the other half to sleep. I wanted to prove that a suspending coroutine could wake up in a thread that was sleeping, leaking the thread local value. What happens instead is that the coroutines run all in the thread 2 (which was surprising to me ) and all of them prints null when reading the thread local value. Now I guess that happens because somehow threadlocal gets cleared, is that correct? What can I change to make sure the coroutines read the wrong value? (yeah I know, but I’m just experimenting)

What’s the most idiomatic way to transform a flow with multiple threads? Let’s say I have a flow that contains suspend lambdas and I want to transform it to a flow that contains the result of those lambdas, but do it with 16 threads. I know how I would do it with channels, but aber flows more suitable?

Hi there, I need to go to particular destination/dialog in my app when ever the user receives HTTP error 401. Is it possible to observe/post that error from a simple kotlin class to a host activity, doesn't matter whichever fragment I am on? I do not want to abuse live data for this as this condition may only occur sometimes. Any kind help would be appreciated.

Hey, anyone aware of a way to

collect

just once from a

StateFlow

or

SharedFlow

? One can call

cancel()

on the job of the coroutine that is collecting but that seems a bit cumbersome...

how can i run my application with a limited number of cores? i seem to run into a weird issue on google cloud run probably related to having only 2 vCores

I’m doing multiple concurrent API calls. It reads better with awaitAll() and destructuring but the types of the results are upcast to Any. Using await() looks uglier IMO but keeps the type information. Is there a way that reads like the former but still retains the type information like the latter?

I recently created a Timer that is handled through Coroutines, encapsulated as a

TimerFlow

object. I wanted to Unit Test this object but I stumbled across a problem, because I am exposing a

SharedFlow

for every “tick” of the Timer I am having trouble testing it since the

replay

value is set to 0. What are the best practices in regards to setting up a Timer like I am here, should I even use a

SharedFlow

to expose every tick or should I be using something else? If the approach I am using is correct, what would be the proper way for me to test this?

i made a chunked flo operator and it seems to work fine in testing but for some reason only when deployed to google cloud run it shits itself and code executed through it takes 100+ times longer to complete afaik the cloud run VMs are limited to 2 cores, so i am guessing it is running into a parallelism limits on the default dispatcher just have no idea how my code triggers that.. because calling it directly (without the chunker operator and buffering) works fine https://gist.github.com/NikkyAI/025436616e81ff0a1711a6144fba0066 any help in suggestions on how to reproduce this locally or what i might have overlooked are appreciated

so I'm curious if there is an operator that is opposite of

combine

?

I’m getting this compiler crash when compiling this suspend function that I’ve simplified for illustration. Where should I report the bug?

private val events: MutableSharedFlow<Unit> =
    MutableSharedFlow()

fun foo(): Pair<String, String>? = null

private suspend fun bar(): String = events
    .transform {
        emit(
            foo()
                ?.let { it.first }
                ?: return@transform
        )
    }
    .first()

Im investigating the use of corountines in my current Android project. Specifically

kotlinx.coroutines.channels.actor

in my Android `androidx.work.CoroutineWorker`(s). What I am trying to achieve is that my Worker makes multiples calls to a remote API endpoint and each API response is sent to an actor channel to be processed on another thread so that I can immediatley start the API call for subsequent pages of data I have these message types

sealed class Message {

class Responsible(val value: Response<List<MyApiData>>) : Message()

class GetValue(val deferred: CompletableDeferred<Int>) : Message()

}

This Actor

fun CoroutineScope.persistActor() = actor<Message>(context = Dispatchers.Unconfined, capacity = Channel.UNLIMITED) {

for (msg in channel) {

when (msg) {

is Message.Responsible -> managePage(msg.value)

is Message.GetValue -> msg.deferred.complete(1)

}

}

}

private lateinit var persister: SendChannel<Message>

each page returned from my API is processed by this recursively called function:-

private suspend fun managePages(accessToken: String, response: Response<List<MyApiData>>) {

when {

result != null -> return

response.isSuccessful -> persister.send(Message.Responsible(response))

else -> {

manageError(response.errorBody())

result = Result.failure()

return

}

}

response.headers().filter { it.first == HTTP_HEADER_LINK && it.second.contains(REL_NEXT) }.forEach {

val parts = it.second.split(OPEN_ANGLE, CLOSE_ANGLE)

if (parts.size >= 2) managePages(accessToken, service.documents(accessToken, parts[1]))

}

}

Once all the pages have been retrieved from the remote end point I execute the following code to await the actor to complete persisting the returned data

val completed = CompletableDeferred<Int>()

persister.send(Message.GetValue(completed))

println("Counter = ${completed.await()}")

persister.close()

what I am concerned about 1). is this a "good" approach? 2). Does my worker wait for all data to be persisted before completing? 3). What improvements could I make?

Hello 👋 Just wondering whether this difference in behavior is expected, i.e. given flow <-> publisher interop

public fun <T : kotlin.Any> org.reactivestreams.Publisher<T>.asFlow(): kotlinx.coroutines.flow.Flow<T>
public fun <T : kotlin.Any> kotlinx.coroutines.flow.Flow<T>.asPublisher(): org.reactivestreams.Publisher<T>

This is valid

// inferred -> flow: Flow<Any?>?
val flow = when (val publisherOrFlow: Any? = fetchValue()) {
    is Publisher<*> -> publisherOrFlow.asFlow()
    is Flow<*> -> publisherOrFlow
    else -> null
}

Yet this is invalid

// inferred -> publisher: Publisher<*>?
val publisher = when (val publisherOrFlow: Any? = fetchValue()) {
    is Publisher<*> -> publisherOrFlow
    is Flow<*> -> publisherOrFlow.asPublisher() // <-- issue here -> inferred type Any? is not a subtype of Any
    else -> null
}

Is there something like

flatMapLatest

that I can trigger explicitly and only once? I want to switch a Flow to another Flow if a certain condition is met

Is there a better way to do this

scope + SupervisorJob(scope.coroutineContext.job)

? I just want to create a child coroutine scope.

scope.launch { supervisorScope { ... } }

which you can then extract as an extension function.

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines/suspend-cancellable-coroutine.html Are the docs outdated? My

continuation.resume

requires a 2nd argument

Hi, I have the following suspending function:

override suspend fun connect(
        scope: CoroutineScope,
        device: BluetoothDevice,
        uuid: UUID,
        secure: Boolean
    ) {
        dataSource.connectToSocket(device, uuid, secure) { socket ->
            scope.launch(<http://Dispatchers.IO|Dispatchers.IO>) {
                try {
                    dataSource.listenToSocket(socket)
                } catch (e: Throwable) {
                    Timber.tag(TAG).e("Error: ${e.message}.")
                    throw e
                }

            }
            startCommunication()
        }
    }

How can I pass the exception to the caller of connect so that the caller can catch this exception? My approach with a nested coroutine might be wrong: I need another coroutine here because

dataSource.listenToSocket(socket)

is a suspending function with a

while (true) { ... }

statement and therefore won't never return, but I have to call

startCommunication()

directly after calling

dataSource.listenToSocket(socket)

.

Hi guys, I have something like

private val eventFlow = MutableSharedFlow<BotStatusChangeEvent>()

override val events: Flow<Event> = eventFlow

override fun sendEvent(event: Event) {
    coroutineScope.launch {
        eventFlow.emit(event)
    }
}

What is the correct way to test that if I do

sendEvent(Event(1))
sendEvent(Event(2))
sendEvent(Event(3))

I receive all events in

events

flow?

how can i call launch from a suspend method method and not wait for it? if i wrap the launch call with coroutineScope {} it waits for the job to finish

Is there any way / workaround to test

kotlinx-coroutines-reactive

-based coroutines using

runBlockingTest

? The related issue (https://github.com/Kotlin/kotlinx.coroutines/issues/1204) and pull request (https://github.com/Kotlin/kotlinx.coroutines/pull/1206) seem to be abandoned for a long time.

Is there any way to launch onto different coroutinescopes inside a for loop so that every iteration of the for loop still doesn't block each other?