val downloadFlow = flow {
            var read = 0
            while (read < total) {
                read += readFromNetwork() // How do I make this run from a background thread ? <http://Dispatcher.IO|Dispatcher.IO> or other ?
                emit(read to total)
            }
        }

        downloadFlow.collect {
            // this should run from the main thread
            displayProgress(it)
        }

What would be the idiomatic way to have a long running background operation emit progress that is read from the main thread ? The

flow

doc explicitely prohibits changing context in the flow block.

Reposting because I never got an answer: What's the recommended way to do a worker pool like structure? eg computing values concurrently (like by using the IO dispatcher) but only allowing a certain number to run at once

anyone got opinion on this style of api? where service has its own scope so doSomething doesnt get canceled then callsite scope does

want to ask one more thing, what would be the preferred way to keep the job instances per each method? so I can cancel granularly?

If I have several coroutines suspended on a Mutex (they all lock from the same thread), will they resume in the same order as they tried to acquire the lock or is this undetermined ?

Where are all the gradle artifacts for coroutines (rx, debug), I can't seem to find how to include them in gradle. The docs only have core...

How can I unregister an Android BroadcastReceiver that I registered in a Flow when the Flow is cancelled?

How can I get the latest emission upon an error and continue - modify that? I have sth like a

Flow<ViewState>

. This

ViewState

also has a boolean property if an error should be shown. Upon error I want to emit the latest view state but call

copy(showError = true)

and complete the flow so that a consumer can use sth like:

.repeatWhen { repeatClickedFlow() }

is anyone aware of a coroutines based rest client for elastic search?

Hello, some of my users are getting this crash in my app:

Fatal Exception: kotlinx.coroutines.flow.internal.ChildCancelledException Child of the scoped flow was cancelled

For some reason this crash does not have a stack trace (on both Firebase and Google Play). This makes it very hard to find the cause. Could someone here help me out? I am using Kotlin

1.3.41

and Coroutines

1.3.0-M2

.

Just saw @elizarov accept an award for Kotlin at OSCON this morning. Congrats!

Hello all I got an external API which delivers data via an callback running on a different thread. How could I encapsulate those data into flows? An simple example, given an external API:

fun externalApi(query: String, callback: (Long,String)->Unit): Unit

I would like to convert this into something like

fun externalApiWrapper( query: String ): Flow<Pair<Long,String >>

Currently I have no idea how to do that.

Hi everyone. I found an issue with broadcast channel, flows and tests:

@Test
    fun `BroadcastChannel keep a coroutine alive`() = runBlockingTest {
        val channel = BroadcastChannel<Int>(Channel.CONFLATED)
        val flow = channel.asFlow()
        launch {
            flow.collect { assert(true) }
            channel.send(1)
        }
    }

this test fails with`kotlinx.coroutines.test.UncompletedCoroutinesError: Test finished with active jobs: ["coroutine#2":StandaloneCoroutine{Active}@2a2d45ba]` it forces me to close the channel for the test to pass. But if i change the

flow

for a

receiveChannel

like this:

@Test
    fun `BroadcastChannel keep a coroutine alive`() = runBlockingTest {
        val channel = BroadcastChannel<Int>(Channel.CONFLATED)
        val receive = channel.openSubscription()
        launch {
            receive.consume { assert(true) }
            channel.send(1)
        }
    }

The test passes without problems

hey folks! I’m trying to merge two flows, is this the simpliest way to do that currently?

fun <T> merge(f1: Flow<T>, f2: Flow<T>): Flow<T> =
    flowOf(f1, f2).flattenMerge()

📣 📣 📣

kotlinx.coroutines

1.3.0-RC is here! Changelog: •

Flow

core API leaves its experimental status and has a lot of new extensions! • Reworked reactive integrations, including simplified lifecycle management and consistent handling of fatal exceptions •

kotlinx-coroutines-bom

for managing transitive dependencies • Various minor bug fixes and improvements • Full changelog: https://github.com/Kotlin/kotlinx.coroutines/releases/tag/1.3.0-rc

Why is it bad in c# and fine in kotlin?

Am I missing a

Mutex.withLock

pattern that supports a suspending body? Aside from just writing it myself, of course. Is there some anti-pattern or gotcha here that prevents it from being provided out-of-the-box?

Is there a way to get coroutines stats via prometheus or influx? Like how many are currently active/running/idle etc

I need bit of guidance, I'm trying to use coroutines to offload DB writes to another thread. I have this DB table that is read on app start and is all writes after that thanks to caching. I wanted to avoid blocking the calls for clients so I'm offloading writes using a channel (writes should be executed in-order so I used a channel). Is this ok? (GlobalScope will be replaced later)

internal class BlockedAppsStoreDatabaseImpl(private val blockedAppDao: BlockedAppDao) : BlockedAppsStore {

    private val channel = Channel<DbOp>(Channel.UNLIMITED)

    init {
        GlobalScope.launch {
            select {
                channel.onReceive { op ->
                    when (op) {
                        is Get -> op.cb.complete(blockedAppDao.getAll().toSet())
                        is Put -> blockedAppDao.putOrReplace(BlockedAppEntity(op.uid))
                        is Remove -> blockedAppDao.remove(op.uid)
                        is PutSet -> blockedAppDao.putOrReplace(op.uids)
                        is Fill -> blockedAppDao.fill(op.uids)
                        is RemoveAll -> blockedAppDao.removeAll()
                    }
                }
            }
        }
    }

    override fun getAll(): Set<Uid> {
        val deferred = CompletableDeferred<Set<Uid>>()
        channel.sendBlocking(Get(deferred))
        return runBlocking { deferred.await() }
    }

    override fun put(uid: Uid) {
        channel.sendBlocking(Put(uid))
    }
    ...

Any particular reason why there's no

Flow<T>.firstOrNull(): T?

Or did I miss it ?

So I’m working through the Coroutine Basics https://kotlinlang.org/docs/reference/coroutines/basics.html#scope-builder and this section says: “_runBlocking and coroutineScope is that the latter does not block the current thread while waiting for all children to complete._” but if i add a coroutineScope block to the example code, it doesn’t change the output?!? In order to reproduce replace the code provided with:

fun main() = runBlocking { // this: CoroutineScope
    coroutineScope {
        launch {
            delay(200L)
            println("2Task from runBlocking")
        }

        coroutineScope { // Creates a coroutine scope
            launch {
                delay(500L)
                println("3Task from nested launch")
            }

            delay(100L)
            println("1Task from coroutine scope") // This line will be printed before the nested launch
        }

        println("4Coroutine scope is over") // This line is not printed until the nested launch completes
    }
}

Did some basic performance evaluation of file loading on Android with different scheduling methods. I found that

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

produced significant garbage overhead. Log snippet:

Background concurrent copying GC freed 1692(904KB) AllocSpace objects, 127(2MB) LOS objects, 49% free, 3MB/7MB, paused 390us total 103.292ms
Background concurrent copying GC freed 1965(1129KB) AllocSpace objects, 163(3MB) LOS objects, 50% free, 3MB/7MB, paused 552us total 118.408ms
Long wait of 8.836ms for Thread[20,tid=16153,Native,Thread*=0xe0c63c00,peer=0x12c40078,"DefaultDispatcher-worker-6"] suspension!
Background concurrent copying GC freed 2006(1273KB) AllocSpace objects, 197(4MB) LOS objects, 50% free, 3MB/7MB, paused 613us total 120.900ms
Background concurrent copying GC freed 2420(1437KB) AllocSpace objects, 216(4MB) LOS objects, 37% free, 10MB/16MB, paused 1.202ms total 256.912ms
Waiting for a blocking GC ProfileSaver
WaitForGcToComplete blocked ProfileSaver on HeapTrim for 29.549ms
Background concurrent copying GC freed 2242(1268KB) AllocSpace objects, 189(4MB) LOS objects, 49% free, 3MB/6MB, paused 520us total 102.964ms
Background concurrent copying GC freed 1782(1071KB) AllocSpace objects, 154(3MB) LOS objects, 49% free, 5MB/10MB, paused 386us total 112.534ms

Any clue why this might be?

Is there a better syntax for collecting a

Flow

to a

Channel

than the following ?

flow {
    coroutineScope {
        val channel = produce {
            aFlow.collect { send(it) }
        }
    }
}

I want to write a custom logic for combining 2 `Flow`s,

zip

and

combineLatest

don't fit my use case.

what's the closest thing in java to

suspend fun getAThingThatMayTakeTime() : T

?

fun getAThing(): Future<T>

?

Hey everyone 👋 I have a question regarding `Kotlin.Flow`s I’m moving a data stream through the layers of an app, and each layer does its thing transforming data whatnot. My data source(s) return a flow, my problem here is: - If I do the mapping before declaring something asFlow() (currently converting a Rx Flowable to a Flow) everything works, but when it reaches the second layer I get an exception :

kotlin.coroutines.intrinsics.CoroutineSingletons cannot be cast to MY_MODEL

- The way I apply these transformations is with the .map() operator To make it more clear Layer 1 - DB -> Flowable -> .map() -> asFlow() - ✅ Layer 2 - flow (from layer 1) -> .map() -> flow() - 💥 I’ve been searching google and GitHub but it seems like this hasn’t been solved. logs

Caused by: kotlinx.coroutines.JobCancellationException: Parent job is Cancelling; job=DispatchedCoroutine{Cancelled}@173d7e3
     Caused by: java.lang.ClassCastException: kotlin.coroutines.intrinsics.CoroutineSingletons cannot be cast to MODEL_CLASS_HERE_
        at CONTACT_REPOSITORY_CLASSl$fetchFlowContacts$1$1$invokeSuspend$$inlined$collect$1.emit(Collect.kt:73)
        at kotlinx.coroutines.flow.FlowKt__TransformKt$map$$inlined$unsafeTransform$1$2.emit(Collect.kt:98)
        at kotlinx.coroutines.reactive.flow.PublisherAsFlow.collect(PublisherAsFlow.kt:78)
        at kotlinx.coroutines.reactive.flow.PublisherAsFlow$collect$1.invokeSuspend(Unknown Source:12)
        at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
        at kotlinx.coroutines.DispatchedTask.run(Dispatched.kt:241)
        at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:594)
        at kotlinx.coroutines.scheduling.CoroutineScheduler.access$runSafely(CoroutineScheduler.kt:60)
        at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:740)

------------- A workaround is using the

flatMapConcat()

instead of

map()

and return a

flowOf(someTransformation())

which feels a little dirty creating a new flow every-time data is emitted. --------------- // Kotlin versions.kotlin = “1.3.41” versions.kotlin_coroutines = “1.2.1" versions.kotlin_coroutines_reactive = “1.3.0-RC” versions.kotlin_rx = “2.1.0" Ran on Android

What's the compiler arg to enable experimental features of coroutines? (like

-XXLanguage:+InlineClasses

for inline classes)

My understanding of RendezvousChannel is the if the receiver isn't waiting to receive, the sender will suspend until there is a receiver. In this case it doesn't appear to be happening, because it seems like the values sent are getting dropped if there is no receiver waiting.

hello, what is the recommended way to do this - I have an api layer that has suspend functions .. my viewmodel implements

CoroutineScope

so I can correctly cancel jobs when my view is destroyed .. and I use repository pattern to fetch/store data. In one case though I use RxJava to observe database changes and If there are no data in database, i will fetch data from Api.. the method looks something like this

override suspend fun observeData(): Flowable<Data> {
    return dao.observeData()
        .doOnNext {
            if (it.isEmpty()) {
                dao.insertData(api.fetchData())
            }
        }
}

however as you may notice, this does not work because the

doOnNext

method is not suspend and i cannot call my api. I’ve been trying some approaches, the current one which is working but I dont know if its correct is this one

override suspend fun observeData(scope: CoroutineScope): Flowable<Data> {
    return dao.observeData()
        .doOnNext {
            if (it.isEmpty()) {
                scope.launch {
                    dao.insertData(api.fetchData())
                }
            }
        }
}

any better approaches? The easiest would be to implement

CoroutineScope

in my repository but it does not have clear strict lifecycle so I could cancel jobs when its “ending”

is

GlobalScope

considered bad practice, what is the specific use case of it?

Hi there 🙂 What is the best way to integrate Channels/Flow with LiveData ?