Lets say you have websocket(or api), which returns asynchronous response each time a request is made. However if more than one requests is sent the previous one will be canceled and the new one will be analyzed. In other words the most recent request is always the one analyzed. Assuming you have a stream of requests, using coroutines, how can you ensure all requests are processed.(Consider back pressure). Also the only way you know the request is completed or in the process, is via the response callback

Is it normal that coroutine takes almost 50ms to run the code in the

launch

block? No context switching, same thread. Simple

MainScope().launch {}

. Is there a way to run it sequentially or at least with smaller delays?

Hi. Should

suspendCancellableCoroutine

being called in a

runBlocking

or

runBlockingTest

context and using the same dispatcher (Test Dispatcher) block while waiting for a java callback that it wraps? The API that

suspendCancellableCoroutine

wraps spawns a Java thread.

hi, quick question: when unzipping a file, which dispatchers should I use? IO or Default?

Hello. Is there any simple build in extension to receive values from simple Channel like

channel.consumeEach { }

but not close channel on cancel. consumeEach does not close BroadcastChannel nor ConflatedBroadcastChannel on cancelation. Should i just stick to

for(it in channel){
    
}

and create my own helper extension function?

Would this be a correct implementation of interval?

Just lost a couple of minutes for a strange issue with StateFlow. When using .collect {} it was automatically resolved to https://github.com/Kotlin/kotlinx.coroutines/blob/master/kotlinx-coroutines-core/common/src/flow/Flow.kt#L177 and did not propose the proper imports so obviously did not compile. Does anyone know why it resolved to that one without any related imports? (With 1.3.8) Manually adding the proper import fix the issue, but it's hard to figure out by default.

Hey there :) does someone know why currently the coroutines package does not support 32bit Windows? Is there a limitation?

I am trying to test below method and assert that

remoteCloudDataStore.getFoodItems()

and

localDataStore.addFoodItems(it)

methods are called

exactly once

fun getFoodItems(forceNetworkFetch: Boolean) = remoteCloudDataStore.getFoodItems().flatMapConcat {
        localDataStore.addFoodItems(it).flatMapConcat {
            flow { emit(FoodResult.Success(it)) }
        }
    }.flowOn(<http://dispatcher.io|dispatcher.io>)

Below is my test case

repository.getFoodItems(true).collect {
            verify(exactly = 1) { remoteCloudDataStore.getFoodItems() }
            verify(exactly = 5) { localDataStore.addFoodItems(any()) }
        }

Surprisingly this test case pass, even if addFoodItems() method is not being called 5 times, can someone point me in correct direction what I am doing wrong here? I am passing

TestCoroutineDispatcher

during testing.

I have a question about documentation that I think is a little misleading and wondering if someone can confirm my understanding. Specifically looking at https://kotlinlang.org/docs/reference/coroutines/basics.html#extract-function-refactoring which discusses refactoring a

suspend

function out of a block and raises this:

But what if the extracted function contains a coroutine builder which is invoked on the current scope? In this case, the 

suspend

 modifier on the extracted function is not enough.

It offers a few options:

Making 

doWorld

 an extension method on 

CoroutineScope

 is one of the solutions, but it may not always be applicable as it does not make the API clearer.

I agree that it doesn't make the API clearer and usually extensions on

CoroutineScope

are reserved for things that return immediately like the coroutine builders. Roman offered a great convention around this in a talk or blog post I believe.

The idiomatic solution is to have either an explicit 

CoroutineScope

 as a field in a class containing the target function or an implicit one when the outer class implements 

CoroutineScope

.

This makes me 🤔 . Wouldn't a more idiomatic solution be to use

coroutineScope

or

supervisorScope

from the

suspend

function that would 1) respect structured concurrency since the scope that is created would be a child of the scope that the

suspend

function was called from and 2) give you a

CoroutineScope

that would allow you to call your coroutine builder like

launch

, etc.?

Hello, What is the best course/tutorial that covers Coroutines well and that you would recommend for someone who already knows RxJava and Kotlin? Many Thanks

Any idea why I didn't get a compile error there? The

fetchSaleWithPhotosById

returns a

Flow<SaleEntity?>

instead of a List, however it still compiles.

override fun read(key: SalesRequestKey): Flow<List<Sale>> = flow {
        when (key) {
            SalesRequestKey.AllSales -> emitAll(
                salesLocalDataSource.fetchAllSales()
                    .map { saleEntities -> saleEntities.map(saleEntityToDomainSaleEntityMapper) }
            )
            is SalesRequestKey.SingleSaleRequest -> emitAll(
                salesLocalDataSource.fetchSaleWithPhotosById(
                    key.saleId
                ).filterNotNull()
                    .map { saleEntity -> saleEntityToDomainSaleEntityMapper.invoke(saleEntity) }
            )
    }

Is there a way to share emitted value between consumers of the flow that's backed by the channel? Can't figure out a way for both consumers to get the same value. In attached test producer is playing ping pong between them. I hope SharedFlow will be released soon, but until it is, is there any other way? 😅

I have a view emitting a flow of Actions which is collected by a handler which emits a flow of Mutations (that will update the state) and they have different lifecycle scopes (the handler can outlive the views). How can I do it so that I can cancel the view’s scope (no more Actions come through) yet the handling of the actions still being processed is not cancelled? Say the handler takes 5 seconds processing each action. When I cancel the view scope, I know that I should expect more mutations to come through but they get cancelled too. I’ve gone through the docs and tried many different alternatives but none worked.

Hi guys, I’m playing with Kotlin coroutines and there is something I don’t understand. Here’s my code

val myScope = CoroutineScope(Executors.newFixedThreadPool(1).asCoroutineDispatcher())

@Test
fun testCoroutines() = runBlocking<Unit>(myScope.coroutineContext) {
    myScope.launch() {
        println("first 1")
        delay(100)
        println("first 2")
    }
    myScope.launch() { println("second") }
}

Can someone explain to me why I will never see “first 2" printed? if I dont use myscope.launch all works as expected and I see the 3 lines printed. That makes sense since the launch coroutines inherit the scope from runBlocking. But if I set it explicitly something changes not sure why (edited)

Here’s another question from some experiment I’m doing. I’m trying to write an async EventBus. It simply registers EventListeners associated to a class, and when he receives a notify it finds the right listeners and invokes them asynchronously. Now, I was thinking that it’s a good idea to make this component being initialized with a CoroutineScope, such that someone from outside can actually decide which scope it runs into (ie I want to decide which threadpool will power the eventbus). I came out with something like below and wanted to ask your opinion if that’s good practice or am I missing something.

class AsyncEventBus(
    private val eventBusScope: CoroutineScope
) : EventBus {
    private val eventListeners: MutableMap<Class<*>, MutableList<EventListener<*>>> = mutableMapOf()

    override fun <E : Event> registerListener(aClass: Class<out E>, eventListener: EventListener<E>) {
        val listeners = retrieveListeners(aClass)
        listeners.add(eventListener)
    }

    override fun <E : Event> notify(event: E) {
        eventListeners[event::class.java]?.asSequence()
            ?.filterIsInstance<EventListener<E>>()
            ?.forEach {
                eventBusScope.launch {
                    it.handle(event)
                }
            }
    }

    @Suppress("UNCHECKED_CAST")
    private fun <E : Event> retrieveListeners(aClass: Class<out E>): MutableList<EventListener<E>> =
        eventListeners.getOrPut(aClass) { mutableListOf() } as MutableList<EventListener<E>>
}

My questions on my code (apart from whether I’m not following best practices for some reasons) are: • is there anything that can happen that I’m not considering? • I’m happy that the EventListener handle is not a suspending function since I wanted to make it transparent to who implements handle(). Of course this has the drawback that if someone start to use async blocks into the handle() function (with a different scope) it may find that code to run even though the listener has been killed. Is this something I should be scared about? or is it fair to say that if the handle() goes async, you will take care of what can happen?

can deferred value be shared between different scopes? Create builder in one, then start and await in another.

Hello I have a simple question

fun main() {
    val observable = Observable.create<Int> { emitter ->
        for (i in 1..5) {
            emitter.onNext(i)
        }
        emitter.onComplete()
    }
    observable.asFlow().onEach {
        println("Flow: $it")
    }
}

Why this code doesn’t print anything?

Anyone else encounter fatal exceptions in the Coroutines internals on Android 7 on specific lower end devices? I've filed the following issue but the stack traces don't really lead to the source. https://github.com/Kotlin/kotlinx.coroutines/issues/2161

I'm refactoring an app for a unidirectional data flow between the View and the ViewModel, I'd like to get some opinions on this, I think there's probably a better way to declare this Flow.

data class State(action: Action, list: ListState)

fun getState(events: Flow<ViewEvent>): Flow<State> {
    return combine(
        getActionState(events).onStart { emit(Action.None) },
        getListState().onStart { emit(ListState.Loading) }
    ) { action, lovedOnesState ->
        State(action, lovedOnesState)
    }
}

If a ListState change causes a new State to be emitted, the next ViewEvents need to be reduced against this new State, and vice-versa, that's why I combine

Should I avoid doing this

launch {
            launch {}
            launch {}
            launch {}
        }.join()

over this?

listOf(
            launch {},
            launch {},
            launch {}
        ).joinAll()

Either way, why? Thanks!

Hi! When I introduce a

scan

in my Flow chain, it starts to act strange — items emitted by channels stop arriving into it. I tried to come up with minimal example to reproduce, posted on SO, would be grateful if anyone could take a look and help: https://stackoverflow.com/q/63092171/258848

is there any difference in terms of efficiency between doing this

urls.map {

GlobalScope.async { fetchLink(it) }

}.awaitAll().flatMap { it }

versus doing this?

urls.map {

GlobalScope.async { fetchLink(it) }

}.flatMap { it.await() }

I am little bit confused in coroutine. i just tried this code and i want this code to check the network if offline Ops! offline or online You are online it works but when i touch the R*etry(*in snackbar action) button the snack bar dismiss

This message was deleted.

Hi guys, I have a question. I know that when you launch coroutines, Kotlin is nice enough that wants to keep structured concurrency such that if a coroutine in a context fails, than all the running coroutines gets cancelled.

runBlocking {
  launch { dosomething }
  launch { throw Exception }
}

This is nice because it make your thinking easier, and you are pretty much saying: here’s all I need to do, do that in parallel and in case something happens fail everything. Now my question is this: what if I want to launch many coroutines that are independent from each other? I take the example of threadpool when on many threads you launch some operation but one thread throwing exception does not affect the others. Imagine that on each coroutine I’m handling a separate http request for example. From the official doc, apart from handling explicitly the exception handling, I haven’t seen any of such examples and I wonder: is the price of launching coroutines that easy paid off when you really want to go in parallel with no noisy neighbours?

Given a

val isLoggedIn: Flow<Boolean>

that can emit

true

or

false

in any order and multiple times each - what would be the best way to observe when it switches from

true

to

false

? I would like to run a function when the user logs out - meaning the

Flow<Boolean>

emits

true

then

false

?

val isLoggedIn: Flow<Boolean>

isLoggedIn
  // <-- what operators do I need here?
  .onEach { 
    onLogout()
  }
  .launchIn(scope)

simple one for y'all.. I have a bounded blocking queue. So

offer

will be blocking. I the right way to offer to this queue is with

runInterruptibe

?