I want to share a

Flow

between a fixed number of collectors. Currently I just do

data.shareIn(this, SharingStarted.Lazily)

. Then I let all the collectors colllect that shared flow. However there is a race condition here. In theory the following can happen: • first collector starts collecting, this starts the sharing • the emitter emits things and they are immediately consumed by the already subscribed first collector. • the rest of the collectors start collecting, but they missed out on the first values. I could set

replay

to some number, but that's just a stopgap. Is there a way to say "okay, sharing coroutine, you can start dropping things now, there won't be any further subscribers" after all the subscribers are there?

I have a

Flow

of requests that need to be processed by (many) different workers, and I was wondering which option is better from the following: 1. Have each worker apply

filter {  }

on the main

Flow

to check if this request belongs to it. 2. Have a separate

Flow

for each worker, and a general worker that checks all the main `Flow`'s requests and re-emit each request to the corresponding worker's

Flow

. 3. (possibly something I haven't thought of?)

I can share ("fan-out") a Flow between multiple collectors while only collecting the original flow once using

shareIn

. However this does not support a finite flow, as

shareIn

makes the flow endless ("Shared flow never completes"). How can the collectors of the shared flow ever complete? Even if the original flow completes, they will continue to "hang", expecting more elements, because the shared flow never completes.

if I have a background/worker coroutine that is processing tasks like so

1 suspend fun processTasks(chan: ReceiveChannel<MyTask>) {
2    for(task in chan) {
3        doWork() // non-suspend function
4    }
5 }

is it correct to say that on Line 3 once the coroutine is running on a thread it will not suspend again and potentially switch threads until the next call to

chan.receive()

on line 2? More generically the only time a coroutine has a chance to suspend is at a suspension point which is defined by a

suspend

function call? (i.e. no instruction level suspension possible)

Can anyone help me understand the use case difference for a

callbackFlow

and a

Channel

? both seem to support "hot" streams

Documentation of SharedFlow says:

All methods of shared flow are *thread-safe* and can be safely invoked from concurrent coroutines without external synchronization.

But IDE on SharedFlow.emit says:

Collects the value emitted by the upstream. This method is not thread-safe and should not be invoked concurrently.

Which one is right?

I have a ArrayBroadcastChannel, which I subscribe to like channel.asFlow().launchIn(scope) between launchIn <-> flow.onStart is about 100-150ms so all items I emit in those 150ms are lost can I subscribe to this channel in some blocking way, so every emit after launchIn would be collected? I’m currently tied to coroutines 1.3.9 because of other dependencies, also Conflated channel is redundant in my case

Hello, is there an easy way to convert a

Flow

to a

StateFlow

on the fly? ( i.e.

flow.asStateFlow()

)

Hello, I was reading about coroutines and I came across this in an SO answer. "when you execute 1,000,000 concurrent routines, it doesn't have to create 1,000,000 threads" Let's say, these many routines run on just 4 threads. In that case it's same as that of a fixed thread pool of size 4 ?

This message was deleted.

I have a few subcomponents, each which have a state. From the 'parent' component, I want to monitor the state changes of the subcomponents and, if any of them go to an error state, take some action. I was hoping to be able to do a

select

expression in the 'parent' on a set of

StateFlows

from the subcomponents, but I saw

onReceive

isn't directly supported by

StateFlow

. I tried using

produceIn

on the

StateFlow

so I could use

onReceive

, but then I get the most recent state over and over. I looked at using

distinctUntilChanged

on the

StateFlow

, but then saw that has no effect...What should I go with here? I also can't 'miss' a value here, so maybe

StateFlow

is the wrong tool?

Given a block -

var x: Int = 0

suspend fun modifyX() {
    if (x > 10) {
        x -= 1
    } else {
        x += 1
    }
}

fun main() {
    runBlocking {
        withContext(<http://Dispatchers.IO|Dispatchers.IO>){
            repeat(10) {
                modifyX()
            }
        }
    }
}

Does

x

need to be

@Volatile

?

hey there. I have a list of a few million urls and i want to fetch the head response for them. What would be the best approach if I want to limit the amount of urls that can be fetched concurrently? Of course I could have a sequence that produces the requests, chunk them and process one chunk at a time but this feels like my default coming-from-a-java-world solution instead of something tailored to kotlin / around co-routines.

Didn’t find much in Google about this. I am trying to test this function which uses

SharedFlow

but I am getting an

IllegalStateException: This job has not completed yet

. It’s basically pretty easy because I commented out the rest of the code for now:

fun registerNewUser(email: String, password: String, hint: String) {
        viewModelScope.launch { _state.emit(State.Loading) }
        // commented out for the moment just to make sure it's not the rest of the code that is failing the test
    }

And the test:

@Test
    fun `starting registration process will set the state immediately to loading`() =
        runBlockingTest(testCoroutineDispatcher) {
            registerViewModel.registerNewUser("anyEmail", "anyPassword", "anyHint")

 ~           assertEquals(registerViewModel.state.first(), RegisterViewModel.State.Loading)
        }

Versions:

coroutines: "org.jetbrains.kotlinx:kotlinx-coroutines-android:1.4.1",
            coroutines_android : "org.jetbrains.kotlinx:kotlinx-coroutines-android:1.4.1",
            coroutines_test: "org.jetbrains.kotlinx:kotlinx-coroutines-test:1.4.1"

Probably doing something wrong, but not sure what. Any ideas/input?

I just deleted a driveby PR comment that should either be a tweet into the ether or a slack question, so here 'tis. Is CancellableContinuation.resume's handling of a cancellation lambda an example where it's less readable? I love Kotlin's last param lambda logic generally. I think in this API, this callsite is an example of where it hides the intent.

cont.resume(Unit) { unlock(owner) }

vs

cont.resume(Unit, onCancellation = { unlock(owner) })

📣

kotlinx-coroutines

version

1.4.1-native-mt

released (with multithreading in Kotlin/Native)

why does

Publisher<T>.asFlow()

not support nullable values? https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.reactive/org.reactivestreams.-publisher/as-flow.html

Hello there, if

repositoryContract.retrieveTopStories()

is a suspend function which returns

List<Story>

why I can’t convert it to Flow by calling

repositoryContract.retrieveTopStories().asFlow()

Hi.. I want to use mutex.withLock{...} but it requires ‘suspend’ keyword which is a breaking change, how can I achieve the same effect (like synchronized in java) without leaking/changing my fun def? (KMP btw)

how can i add throttling to a code like this:

(0 until entries).map {
                    async {
                        repo.create(connection, user)
                    }
                }.forEach { it.await() }

Can someone explain how to convert live data to state flow in this situation: - livedata2 is map of another livedata1 in vm and there are 2 observers in ui layer, one observes original livedata1, and other observes mapped livedata2 What I discovered is that simply converting live data to state flow is not working as expected. Map is converting state flow to flow and not state flow. Second I cannot get both flows to trigger collect when collect are written as part of the same scope in ui layer. Any resource on the subject?

I'm not sure if this is the right place to ask but is it on the roadmap to support a concurrent hash map in Kotlin common? It's very important for our multiplatform project and we didn't like 3rd party libraries.

Am using

StateFlow

in view model (along with use of

stateIn

)....is there something equivalent to

MediatorLiveData

or

switchMap

when using that?

I am using

Flow

with

suspend functions

to read a stream of data from my API. To get all the registered users,

flow

seems like a cool option. But while creating a user which is a one-shot function, would flow really come in handy? I really like the idea of emitting streams by passing a Result wrapper around the data and this gives me more clarity and readability in code. Just want to know if it is considered as best practice to use Flow for one shot functions like creating a user or does it become unnecessary?

Why do all methods in the docs for producing functions extend CoroutineScope? Is that just a convention or does it serve a special purpose?

Are there any articles available on Shared Flow and error handling? I have a common use case that I want to share a network call but want new consumers no trigger the upstream flow when subscribed after an error.

i am curious if there is a better way to run a “fireAndForget” action on the items in a flow as an intermediary step e.g in a repository

suspend inline fun <T>Flow<T>.fireAndForget(crossinline block: (T) -> Unit): Flow<T> {
    coroutineScope {
        launch {
            runCatching {
                collect {
                    block(it)
                }
            }
        }
    }
    return this
}

We're trying to perform some tasks sequentially and i was wondering if an actor or a Mutex is better suited for this case. There'd be many queues and each of them will be processing few tasks.

Is there a way to poll a single element from Flow, like Channel.receive ()? I want to poll the data inside a recursive parser function instead of always receiving the data in the same lambda.