Maybe not related to coroutines, but that's where I found it: Can anybody explain to me why the following code works? https://pl.kotl.in/O8wu8VKSE In essence the code sets an instance of type

B

as a value of

StateFlow<A>

. I would not expect that to be possible. The code not only compiles, it actually runs in the Playground without a run-time exception.

This message was deleted.

Hi, I want to define list of operations that will be executed later. What is better solution in this case? •

List<Deffered<T>>

•

List<suspend () -> T>

Is there some way to get job from

CoroutineContext

by key and cancel it?

I'm migrating libraries from Rx to Kotlin Coroutines/Flow --- would be nice if I could start using

shareIn

/

stateIn

operators... I'm currently planning to work around those by transforming to Rx, using

publish().refCount()

or

replay().refCount()

and then turning it back to

Flow

cause i can't wait anymore for the new operator. I need to stabilize the API so the only other option without those operators is to go with Rx. When

shareIn

will be available I should be able to remove the Rx dependency without having to modify the API interface (hopefully). Do you know if there's something particular holding back the release of

ShareFlow

or if we could expect it to be available soon?

Hi guys, I have some libraries that use ThreadLocal in my application and I’m using the

.asContextElement()

extension function to make them coroutine-ready. All works well but i find my code having something like this

launch(object1.asContextElement() + object2.asContextElement()) {}

I was thinking to encapsulate those element into a container such that I can do something like this

launch(myContext) {}

where

mycontext

contains the elements from object1 and object2. I was thinking to do this by implementing the

CoroutineContext

interface but I then I need to implement all its functions, while what I really want is just the default

+

behavior. Is there any nice way to achieve this?

hey guys, there is a flow return a list that every item with a retrofit api service, then i want to get all detail from remote api in one time, not like

flatMapConcat

one by one.

flatMapMerge

can not do it, and start coroutines will break the flow are there any good solutions? thanks

I have a question, it's clearly my misunderstanding, but, I have a JobCancellation on my Workers ($doWork: kotlinx.coroutines.JobCancellationException: Job was cancelled;) Only when my screen is off. If the app is the background or if we are on the app. No problems.

Hmm, why is

Flow.firstOrNull

restricted to non nullable types? Feels like it'd be useful even with flows with nullable types?

hi everyone, why if I convert flow to a rendezvous channel (I need it for custom operator) and collect it, unnecessary dispatch (between collecting 1 and 2.) is performed. Is there a way to fix this?

val flow = flowOf(1, 2)
produce(capacity = 0) { flow.collect { send(it) } }.consumeAsFlow()
   .onEach { println("each $it") }
   .launchIn(this)

I would expect there will be no difference comparing with:

val flow = flowOf(1, 2)
flow.onEach { println("each $it") }
   .launchIn(this)

Link demonstrates unnecessary redispatch: https://pl.kotl.in/LbbXNwWzU

so we’re experiencing some problems with flows and coroutines. general setup: we have a grpc stream that provides messages as flow. inside the collect we launch a new coroutine to do some heavy lifting. this is of course done in a different coroutine context (using the default dispatcher, we used a custom dispatcher before and had the same problem so we went back to the default). what we see now is that after some time, the inner launch is just not scheduled/executed, it’s not blocking, though, the launch actually returns and thus the flow is also consumed properly. we struggle hard reproducing or even just monitoring this issue. here is somewhat example code (our input arrives a lot slower, though). we at some point still see logs like “received” but no “processing” anymore. probably we are doing something really wrong, has anyone seen this before?

runBlocking {
    val coroutineScope = CoroutineScope(EmptyCoroutineContext)
    val flowScope = CoroutineScope(EmptyCoroutineContext)
    flowScope.launch {
        val flow = flowOf(1)
        while (isActive)
            flow.collect { i ->
                println("received $i")
                coroutineScope.launch {
                    // Offload some hard work
                    println("processing $i")
                    Thread.sleep(1000)
                    println("done processing $i")
                }
            }
    }

    // just added so it doesn't terminate right away
    delay(10000)
}

Everything is labeled out here : https://github.com/AnonymousGeekDev/FastLightJP2/blob/coroutine-optimized/app/src/main/java/com/example/jp2ssr/kotlin/MainActivity.kt

I have a wrapper function to limit concurrent calls to a service looking like this:

private val concurrentRequests = Semaphore(50)

private suspend fun <T> requestWithLimit(block: suspend () -> T): T {
    try {
        withTimeout(10) {
            concurrentRequests.acquire()
        }
    } catch (e: Exception) {
        throw RuntimeException("Overloaded", e)
    }
    try {
        return block()
    } finally {
        concurrentRequests.release()
    }
}

I usually call io.ktor.client.HttpClient.request inside block(). If I want to write it as an extension function on io.ktor.client.HttpClient (just to make the call site a bit cleaner), how would I define it? It seems that the extension function cannot access the variable

concurrentRequests

I'm using callbackFlow to adapt a callback from Firestore into a flow, and then store the received data locally, as a "fire and forget" call. I can do this:

@ExperimentalCoroutinesApi
private fun retrieveFromFirestore(porteroId: String): Flow<Portero> = callbackFlow {
    val porteroCallback = EventListener<QuerySnapshot> { querySnapshot, e ->
        val element = //extract data from snapshot etc.
             launch { //save to local sqllite }
        offer(element)
    }
}

This works well. However, when I tried to refactor the code a bit, and extract the snapshot manipulation to another method:

@ExperimentalCoroutinesApi
private fun retrieveFromFirestore(porteroId: String): Flow<Portero> = callbackFlow {
    val porteroCallback = EventListener<QuerySnapshot> { querySnapshot, e ->
        val element = buildData(querySnapshot, e)
        offer(element)
    }
}

private buildData(querySnapshot: QuerySnapshot?, e: FirebaseFirestoreException?): Element {
    //extract data from snapshot etc.
    launch { //save to local sqllite }  // <--- this doesn't compiles
    return element
}

The launch function is no longer available, with the error Unresolved reference. I can fix it using

CoroutineScope(Dispatchers.Default).launch { ... }

but why is this happening? How come launch is available only in the calling function but not in the called one?

Hello ladies and gentlemen. Who knows when

flowOn, launchIn, etc

functions will be released?

Hello all 👋 Is there a channel that emits its value to all subscribers, while suspending (keeping the value) if none is subscribed and then the value gets consumed?

Is it expected that

kotlin.Result.Failure

throws instead of delivering value when continuation is resumed on a different thread?

suspend fun main() {
    val failure = fail()
    println(failure)
}

suspend fun fail() = suspendCoroutine<Result<String>> {
    Executors.newSingleThreadExecutor().execute {
        it.resume(Result.failure(RuntimeException()))
    }
}

Exception in thread "main" java.lang.RuntimeException
	at MainKt$fail$2$1.run(main.kt:12)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

what is the meaning of "(Coroutine Boundary)" if you see them in a stacktrace?

I'm very new to Kotlin and so, to co-routines. I'm trying the following snippet, but the program always finishes before printing "data". Could'nt make progress out of it

fun main() {
    GlobalScope.launch {
        val deferredUser = getUserByIdAsync(1)
        val data = deferredUser.await()
        print(data)
    }
}


private suspend fun getUserByIdAsync(userId: Int) =
        GlobalScope.async {
            println("Retrieving user from network")
            delay(3000)
            println("Still on coroutine")
            User(userId, "John", "Silva")
        }

With flows it's simple to

debounce

input. I'd like to know if some value is currently being debounced (and I'd like to know the value), so that I don't have to wait for it to be collected downstream. How would I do that?

I have a

suspend

method that gets dynamically overridden by Guice AOP. It appears that if I have a method handle and use

method.kotlinFunction.callSuspend(obj, *args)

to call that method, it fails because the wrapped method's

isSuspend

property is false (probably because it's dynamically generated by something that is not the kotlin compiler and doesn't know about suspend functions). It looks like I can use

suspendCoroutineUninterceptedOrReturn { method.kotlinFunction.call(obj, *args, it)}

to bypass the

isSuspend

check, but is that safe or is there another approach I should take?

Hi everyone, I'm trying to use MutableStateFlow, encapsulating a List of an object, but when I update it with some new value, the collectLatest{} method doesn't work, can you help me please? This method doesn't update the value

@ExperimentalCoroutinesApi
fun deleteSelected() {
    val value = dataPoints.value
    value.removeAll { it.selected }
    dataPoints.value = ArrayList(value)
}

I seem to have an incorrect understanding and I was wondering if someone could help.

Fragment {
  lifecycleScope.launch { vm.attachListeners }
}

VM {
  val locationListener:ReceiveChannel<Location>
  val navListener:ReceiveChannel<NavEvent>

  suspend fun attachListeners(){
     locationListener.consumeEach{//Do stuff}
     navListener.consumeEach{//do stuff}
  }

With the code above, if this was RxJava, it would behave correctly. With Coroutines the first

.consumeEach

is blocking it appears, never executing the second one. I can change it to

suspend fun attachListeners(){
    coroutineScope{
     launch{
       locationListener.consumeEach{//Do stuff}
     }
     launch {
       navListener.consumeEach{//do stuff}
     }
  }
}

And that will work. The goal is to make it automatically cancel with the lifecycleScope, I'm wondering is there a better way of doing this? And why does .consumeEach block?

why hasn't coroutines-test runBlockingTest been fixed yet...

I want to create a library with a custom time-based operator extension on Flow, such as for example

Flow<T>.throttleFirst()

here. @elizarov wrote that to test such an operator we would have to use the

DelayController

if available instead of

System.currentTimeMills()

, which of course makes total sense. My question: How could I publish this library without having

kotlinx-coroutines-test

in the release dependencies? Meaning how would I prevent the user of my library having

kotlinx-coroutines-test

in their release binaries when they use my dependency?

What would be like in coroutines in case I want to do one job first -> wait that to finish -> then 2 other jobs later?

Multiplatform question. I am trying to get

delay {}

and

runBlocking {}

working on iOS/native. Currently, I am trying out

<http://1.3.5-native.mt|1.3.5-native.mt>

and am getting some bizarre/arbitrary freezing errors that would otherwise work on the JVM. What is the recommended way of launching coroutines on iOS? Is there a guide or template anywhere? Is it worth upgrading to 1.4 and using the latest version?

How can i properly pass the value of a async function so i can use "outside" it? I need to do 1 for each block like:

val bar = async { foo() }

Or there's a way to do like:

async { val bar = foo() }

and then use bar value?

The other day I came across

suspend fun main()

for the first time. Where is it documented? How does it work?