I would like to turn this into a sequential call like with Rx’s

andThen()

viewModelScope.launch(<http://Dispatchers.IO|Dispatchers.IO>) {
            petsRepository.getPets(this).collect { pets ->
                deviceRepository.getDevices(this).collect { devices ->

is

Job#invokeOnCompletion

always invoked independent on the result of that Job (e.g. it gracefully completed, or it was cancelled while suspended)?

Is there an equivalent to

LiveData builder

in flow?

Do I understand correctly that in this snippet flow2 will begin collection only after flow1 is finished (after approximately 3 sec)?

val flow1 = flowOf { emit(1); delay(3000); emit(2) }
launch {
  flow1.collect()
  flow2.collect()
}

Hi, can anyone show me how to wait for a deferred with a timeout but allow the deferred to continue without a cancellation or timeout exception from the surrounding context?

I want to have a

supervisorScope

in which I want to do mutliple

launch

-es and allow each one to fail individually without cancelling other running ones. At the same time I'd want to install an exception handler on the

supervisorScope

itself to centrally handle everyting. But it seems that this can't be done, according to official documentation, i'll need to do

launch(handler)

every time. Do I understand this right? It's not problematic, but I'd rather avoid "forgetting" to pass "handler" to

launch

and have it automatically provided...

Hi, I am trying to call a suspend function from rxjava but I can’t seem to get it to work. I have imported

rg.jetbrains.kotlinx:kotlinx-coroutines-rx2:1.5.2

and tried to use GlobalScope.rxSingle but that method is not available on GlobalScope.

What's the 'proper' way to bridge coroutines and a 3rd-party library (smallrye-mutiny)? I'd like to be able to do something like:

@GET
suspend fun get(): Uni<String> = uni { getFoo() } // run coroutine and return the result as a "Uni", which is similar to Reactor's Mono.

The existing 3rd-party helpers in kotlinx.coroutines' rely on

AbstractCoroutine

, which warns that it shouldn't be used outside of the core lib. (e.g. mono builder)

I'd like to ask about scope cancelation. Is it 100% synchronous?

someOtherPlace {
   someUserScope.launch {
       val userFlow: Flow<User?> = userDao.user()
       userFlow
        .collect {
             // can 'it' be null here?
         }
   }
}

fun logout() {
	allUserCoroutineScopes.forEach { it.cancel() }
	userDao.deleteUser()
}

lets assume

someUserScope

is in the

allUserCoroutineScopes

list, and I'll only delete user after running the cancelation forEach Is there a way the flow at the top will emit null? Should never happen, right? Is it safe to

!!

there? (I'm looking at sources but its non obvious to me)

Hi, guys. I have AuthtenticationInterceptor to add token into header of request. In overrided intercept function it uses runBlocking builder to get token (token is obtained as flow object. Its repository implements PreferencesDataStore). I should replace runBlocking builder to other proper implementation. Because using runBlocking builder in production code is not good practice. Any sugestions?

@Singleton
class AuthenticationInterceptor @Inject constructor(profileRepository: ProfileRepository, @ApplicationScope coroutineScope: CoroutineScope) :
    Interceptor {

    private val tokenFlow: Flow<String?> = profileRepository.getProfileToken()
        .stateIn(coroutineScope, SharingStarted.Eagerly, null)

    override fun intercept(chain: Interceptor.Chain): Response {
        val requestBuilder = chain.request().newBuilder()
        val token: String? = runBlocking { // this line should be changed
            tokenFlow.firstOrNull()
        }
        token?.let { requestBuilder.addHeader("Authorization", it) }
        return chain.proceed(requestBuilder.build())
    }
}

Hi All, anyone know when I use coroutine on custom view onDraw (just for experiment purpose), it got drawn at weird coordinate?

override fun onDraw(canvas: Canvas) {
        super.onDraw(canvas)
        CoroutineScope(Dispatchers.Default).launch { // <-- add coroutine
            canvas.drawLine(
                0f, 0f,
                width.toFloat(), height.toFloat(),
                strokePaint
            )
        }
    }

More detail (with image) here

Hi ! I'm using

coroutineScope.launch {

in order to start a coroutine in my code. But is there a variant of

launch

which runs the coroutine for a given amount of time, instead of infinitely ? Because otherwise, I'm afraid my code, which is a bit a kind of spaghetti as now, will be worse.

Just quick and broad question — would Virtual Thread change Coroutine implementation? Seems like VT would be GA in 3 years: https://openjdk.java.net/jeps/8277131

Still a flow/coroutine noob. I'm using firebase firestore and it has a listener approach to listen to updates on my data. I want to convert this into a Flowable so I can observe it. Is this an okay approach

suspend fun getThingFlow(myId: String): Flow<Thing?> {
    val thingFlow = MutableStateFlow<Thing?>(null)
    FirebaseFirestore.getInstance()
        .document("things/$myId")
        .addSnapshotListener { value, error ->
            thingFlow.emit(value!!.toObject<Thing>())
        }

    return thingFlow
}

With the new MM, will coroutines

1.6.0

support calling suspend functions (and flow collect) from other iOS threads?

Hi, I still doesn't understand the basics of coroutines, like how to properly create some and launch them in a simple project, what are the dispatchers, how to launch multiple time a suspending function to optimise a big normal function etc Can someone explain me this examples please ?

What is the proper way to listen to multiples flows of any possible type and have a resulting flow whose only purpose is to do something but does not care about the value, just that one of the flow have changed? Like I have 10

flowX.drop(1)

And would like to have a function that collect them all in a scope and call 1 function when any value is received (would be nice it the result was a flow so I can debounce too) So some kind of combine but without types and transform and supporting any number of input flows.

📣 📣 📣 kotlinx.coroutines 1.6.0-RC is here! Major highligths: • Completely reworked and now multiplatform kotlinx-coroutines-test module. • New API to manage dispatchers:

limitedParallelism

along with

<http://Dispatchers.IO|Dispatchers.IO>

being unbound for limited parallelism views. • K/N new memory model is now part of the release and all the coroutine primitives can be shared across multiple threads if the new memory model is enabled. • A lot of improvements and bug fixes. Full changelog: https://github.com/Kotlin/kotlinx.coroutines/releases/tag/1.6.0-RC

Will there be a

native-mt

release for

1.6.0

?

Is there a rxjava

buffer

equivalent in Flow? flow.buffer is not the same behavior from what I see, i.e. I want to collect upstream emits into some buffer, and only emit the buffer when its full/time elapses etc

Hi there,

limitedParallelism

in 1.6.0 looks very promising. One question that jumped immediately to my mind: Is this feature recursive? i.e. let’s say I have 2 endpoints I want to serve and I want to limit thread usage to 64 overall, but only allow 50 threads per endpoint. Would the following code do that?

val overallIODispatcher = Dispatchers.IO.limitedParallelism(64)
val endpoint1Dispatcher = overallIODispatcher.limitedParallelism(50)
val endpoint2Dispatcher = overallIODispatcher.limitedParallelism(50)

I’m struggling to write tests around

callbackFlow

. I want to ensure that the value emitted from the flow matches the value that the callback receives. My problem appears to be that the

ProducerScope

block doesn’t get called until something subscribes to the flow. I thought that using

shareIn

would fix my problem, but it still doesn’t appear to kick off the

ProducerScope

in time

fun someTest() = runBlocking {
  val expected = "someMessage"

  val resultFlow = getCallbackFlow().shareIn(this, SharingStarted.Eagerly, replay = 1)

  sendMessageToTriggerCallback(expected)

  val actual = resultFlow.first()

  assertEquals(expected, actual)
}

After adding some logging, I find that what happens is that first,

sendMessageToTriggerCallback

is invoked, then the callback is registered in the

ProducerScope

, then it suspends indefinitely awaiting the first signal from the

resultFlow

. I’ve tried a lot of variations on this pattern, and I can’t seem to make the

ProducerScope

run without blocking my ability to subsequently trigger sending the message. I’ve tried launching callback flow collection in its own job and sending the message a separate job and yielding between them. I’ve tried leveraging

onSubscription

, but that doesn’t fix the issue either, I get the same execution order as the example above. I’m at a bit of a loss as to how to enforce a deterministic execution order in this test case.

Hi all, can anyone help me understand why this terminal collector does not seem to cancel the flow to which it's attached? https://pl.kotl.in/wB5Pt1jsQ

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import java.util.concurrent.Executors


fun main(args: Array<String>) = runBlocking {
    val requests = flow {
        var i = 0
        val serverDispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()
        val handlerDispatcher = Executors.newFixedThreadPool(4).asCoroutineDispatcher()
        launch(serverDispatcher) {
            while (true) {
                i += 1
                val result = i
                launch(handlerDispatcher) {
                    delay(500)
                    launch(serverDispatcher) {
                        emit(result)
                    }
                }
            }
        }
    }
    
    // I want this to:
    //
    // 1. Suspend while collecting each emitted value.
    // 2. Continue to collect until seeing a value >= 4
    // 3. At that point continue.
    // 4. Cancel the flow (thereby closing the server's socket).
    //
    // But instead it suspends, the while loop keeps running and indicating it's emitting values, and the onEach print
    // statement never executes.
    requests.takeWhile { it < 4 }.onEach { println("got $it") }.collect()
}

Would a

message

posting by

Dipatchers.Main

already in

message queue

but not be picked by

Looper

will be removed from the

message queue

if the coroutine is canceled?

any one here can you explain to me what this code means?

val coroutineContext: CoroutineContext get() = Job() + Dispatchers.Main

👋🏻 Hey guys, question regarding testing the coroutines. Suppose I have a class like this:

class Launcher(dispatcher: CoroutineDispatcher) {
    private val scope = CoroutineScope(SupervisorJob() + dispatcher)
    
    fun launchItem() {
        scope.launch { /* Do my thing here */ }
    }
}

I suppose the correct way to test this is to use

Dispatchers.setMain(myDispatchers)

in the

@Before

function and then use

runBlocking

in my test function, where I call

launcher.launchItem

, right? Or how should I test a method that’s not

suspend

but launches a

coroutine

inside ?

Can someone help me understand about

delay()

s inside a

runTest{}

? In my mind the delays would wait until I call

advanceTimeBy()

but in practice the test just run and ignore all the delays. What’s the point of the

advanceTimeBy()

then?

I have some Java api that looks like this

interface Worker {
    // does the work, throws Exception promptly if cancel() is called
    fun doWork()  
    fun cancel()
}

What is the best way to convert it into

suspend fun Worker.doWorkCancellable()

? Preferably something that works with a single threaded executor.

I'm doing some reading on asynchronous programming in different languages. I'm adding what I learn into a WIP doc - you check it out here Q: Do you know of good blog posts or articles that explain how async is implemented in languages?