Hi, is there a way to add an event listener to coroutine exceptions that we don't have access to ? So like if at any moment there is an exception in any coroutine the listener is run ?

In the

kotlinx.coroutines.test

migration guide for 1.6+ when talking about replacing

runBlockingTest

with

runTest

there is a little line:

It works properly with other dispatchers and asynchronous completions.
No action on your part is required, other than replacing runBlocking with runTest as well.

What is meant by this? Why does

runBlocking

need to be replaced with

runTest

if there is no dispatcher/virtual time manipulation?

If i have three flows that return boolean, what is the best operator equivalent to this: (if(aaa || bbb || ccc) ?

class MyObject private constructor() {
    companion object {
        @Volatile
        private var INSTANCE: MyObject? = null
        
        fun isReadyInstance(): Boolean {
            return (INSTANCE != null)
        }

        fun getInstance(): MyObject {
            return INSTANCE  ?: throw Exception("Error Instance is not initialized.")
        }

        fun createInstance() {
            synchronized(this) {
                if (INSTANCE == null) {
                    INSTANCE = MyObject()
                }
            }
        }

        fun deleteInstance() {
            synchronized(this) {
                INSTANCE = null
            }
        }
}

there is condition in my code to check for instance

if (MyObject.isReadyInstance()) {
    MyObject.getInstance() //BUT HERE I GET -> "Error Instance is not initialized."
}

I think when

isReadyInstance

returned true meanwhile at some other place

deleteInstance

is called and therefore

getInstance

returned null, but how can i make sure that it won’t happen ?

Hi everyone, in my team we’re having a hard time understanding if launching a coroutine in a certain point of the application is useful or not. We’re developing a gRPC service, and inside a global interceptor we need to publish a domain event, for tracking purposes. Since this action should be the “fire & forget” kind, we thought it would be a good idea to launch a coroutine for publishing the event and then carry on with the execution. Here’s a simplified version of our code with changed names:

class PublishingServerCall<ReqT, RespT>(
        next: ServerCall<ReqT, RespT>,
        val publisher: EventPublisher
    ) : ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT>(next) {
        override fun close(status: Status, trailers: Metadata) {
            if (!status.isOk) {
                runBlocking {
                    supervisorScope {
                        launch(Dispatchers.Unconfined) {
                            publisher.publishError(status)
                        }
                    }
                }
            }
            super.close(status, trailers)
        }
    }

I know there’s lot of grpc-specific stuff but bear with me, because it’s important details. Here is an implementation of a

io.grpc.ServerCall

, an abstraction needed by gRPC infrastructure to implement a custom interceptor. As you see we override the

close()

function because we’re interested in firing an event when the call is returning from a service and we have a response

Status

. We thought that this setup would allow us to publish the event asynchronously and move on in the interceptors chain without waiting for the publisher to complete. It seems to be working so far, but we only did some manual tests on our machines, it hasn’t been deployed yet. But the more I look at this code, the more I’m convinced it’s not doing what we thought: I have a feeling that the whole

runBlocking {}

part is useless and it’s blocking the thread anyway waiting for the return, and it would be the same to just invoke the publisher directly. Of course it’s needed because the enclosing function is not suspendable, but I’m full of doubts and don’t know if we wrote the right thing. Is this function blocking the request thread or is it returning right away, without waiting for the publication? Although skilled in service programming and Kotlin, we’re pretty much beginners on coroutines, so please if anyone has the expertise an advice will be much appreciated, thank you!

Is there a way to collect each item in a separate coroutine in Kotlin Flow? For example, if I have this source:

val source = flow {
        emit(1)
        delay(200)
        emit(2)
        delay(200)
        emit(3)
    }

source
    .onEach { delay(500) }
    .collect {
        ...
    }

I want my items to be collected after:

1 - 500ms
2 - 700ms
3 - 900ms

This message was deleted.

Hi guys, i have a question. In case i have a shared flow and i want to apply filter, which is the simplest way to retrieve back the sharedFlow, since i cant just cast it back to SharedFlow as filer returns a Flow. Thanks in advance for any help

Is there something like

Flow<T>.distinctUntilChanged(areEquivalent: (old: T, new: T) -> Boolean)

but emits the newest of the equivalent values instead of the oldest?

I would like to return a

flow of T

if cache is empty, otherwise T (as flow). I'm wondering which way is more appropriate:

fun fetchStuff(): Flow<Stuff> = 
        if (cache.isNotEmpty()) {
            cache.asFlow()
        } else {
            api.fetchStuff()
        }.map { model -> model.toStuffModel() }

or

fun fetchStuff(): Flow<Stuff> = flow {
        if (cache.isNotEmpty()) {
            emit(cache.first())
        } else {
            emitAll(api.fetchStuff())
        }
    }.map { model -> model.toStuffModel() }

This message was deleted.

I'm trying to mimic CoroutineScope in Swift concurrency. My issue is that after I cancel the scope, it no longer accepts new `launch`es Which is good It does how ever still return a

Job

. What is this job? Is it already cancelled or what is it? Wouldn't be cleaner to throw?

Are there any Flow debugging tips or tools? For example, I think I have gotten myself into a bit of a mess with a few Flows that I'm using, and other flows that are collected as a result of one main flow (think Flow<UserSession> being collected and in collect{ someLongLastingFlowable() }) Is there like anyway to see how many flowsables I currently have going on? This would also be great when I put my android application into the background and hopefully I'd be able to inspect how many flowables haven't properly "shut down", or at this point should I just put log statement everywhere?

To the Android Developers: Flow VS LiveData What is the advantage of exposing Flows instead of LiveData from your ViewModel to your Views? One argument I keep hearing is that you can then use flow operators in your View. However, since Views should actually only render received states to the screen and not have actual logic in them, I don't see any use case where I would need a flow operator there... I asked this on twitter as well: https://twitter.com/LukasLechnerDev/status/1551584399485026307

It there a way to limit the rate at which a flow emits elements? My use case it to avoid running costly operations too often. For example, given the following source flow:

val source = flow {
  emit("A")
  delay(100)
  emit("B")
  delay(50)
  emit("C")
  emit("D")
}

when applied a constraint of "max 1 element each 100ms", it should emit: • "A" at t = 0ms • "B" at t = 100ms • "D" at t = 200ms

What's the pattern to dispose of a resource when a suspend function needs to allocated a disposable resource. How do I handle the case where the cancellation happens after the resource is allocated but right before it's returned?

That is what if the cancellation happens when return is invoked

DataSources in Android Apps I have a datasource that exposes a

SharedFlow

and performs network request every minute while there are active subscribers and emits the result. I don't want to perform network requests when the app is in the background. Should it be the responsibility of the datasource to stop performing network requests when the app goes into background, or should it be the responsibility of the clients of the data source to cancel their coroutines that collect from the datasource? If it's the responsibility of the clients, then one faulty client implementation can make the datasource to stay active ....

What does

resumes

mean for continuations? Where and when the code after suspension point run? Like if the continuation runs in an infinite loop, does it mean that

resumes

will never end?

I dont understand this. The warning says suspending fun has a CoroutineScope receiver, but the second class fails to find the "launch" method. Is this a bug or am i really misunderstanding this?

hey guys is it possible to use flows with the

select

builder? something like this

flow {
 emit(select {
  someFlow1.onCollect()
  someFlow2.onCollect()
 })
}

Is it OK to share single instance of

CoroutineExceptionHandler

between several `CoroutineScope`s?

Is there a best practice for performing for handling a situation where you want to cancel/invalidate a flow if some input changes? The line of thought goes: 1. Get a flow representing the values for the term “a”. 2. Trigger a network call that results in updating the search with new results for “a”. 3. The collected flow now receives the updated data for the search “a”. 4. Start a new search for the term “abc”. 5. The previously collected flow is canceled and only the new flow with results for “abc” is collected. It works to hang onto the old Job where the initial flow is collected and then cancel that before collecting the new flow but that feels hacky. Playground Link (Code in the thread)

I'm stress testing some callbackFlow code I have written. Here it is currently

fun getBooks(): Flow<List<Book>> {
  return callbackFlow {
    val listener =
      FirebaseFirestore.getInstance()
        .collection("books")
        .addSnapshotListener { value, error ->
          if (error != null || value == null) { /*dosomething*/ }
          var books: List<Book>? = null
          runBlocking(<http://Dispatchers.IO|Dispatchers.IO>) {
              books = (value!!.toObjects())
          }
          trySend(books!!)
        }
    awaitClose { listener.remove() }
  }
}

this seems to work well... but to stress test it I wrapped

books = (value!!.toObjects())

with

repeat (1000)

and now my UI hangs. Shouldn't the dispatchers.io take care of this?

runBlocking(<http://Dispatchers.IO|Dispatchers.IO>) {
  repeat(1000) {
    books = (value!!.toObjects())
  }
}

What could cause this exception? (thread)

I have some coroutine code that is getting cancelled with a

JobCancellationException

. How can I pinpoint the exact line of code that actually caused the coroutine to cancel? I've tried setting an exception breakpoint for

JobCancellationException

, but it doesn't get triggered for some reason.

This is potentially something that should just go into #getting-started but I'll just ask here. I have a list that I search through and while I search through it, sometimes the list is updated. No issues, except for that the list is large and so it's "slow" and lags the UI. I wrapped it in a

withContext(Dispatchers.Default){

but now I get a

ConcurrentModificationException

. I could understand if I was adding/removing to the list on Dispatchers.Default while main thread was swapping the list entirely... but I'm just searching via

indexOfFirst

. thoughts?

So I know coroutine cancellation is cooperative, but I have a fairly simple case here that I'm not sure how to dig myself out of. I essentially have a button that calls

fetchMoreData()

fun fetchMoreData(i) {
  viewModelScope.launch { ...

but now if I call fetchMoreData() again, I actually want to cancel the old data fetching. How would I do that?

Hey folks I am trying to parallelize the processing of a flow. I am doing something like:

private suspend fun process() = coroutineScope {    
    getFlow()
        .map { async(<http://Dispatchers.IO|Dispatchers.IO>) { ioBlockingFunction(it) } }
        .buffer(20)
        .map { it.await() }
        .collect()
}

my expectations is that 20 blocking operations will be trigger at time, then they will be awaited and then next 20 blocking operations will be trigger. What I am noticing is that somehow flow stop producing values and my app got block at flow element 25 of 30, or 100 (at a random point), even when flow function has more records to produce