Hello, can somebody explain me this? 🤨 ( Don’t mind about the typo 😄 )

How does one catch an exception within a

launch

scope? Example code: https://pl.kotl.in/g3Gz4nH43

Hi friends, I'd like to observe a coroutine scope's cancellation. Is there an idiomatic way to do that? My use case is I'm working on something that multiplexes I/O connections (specifically Android binder IPC) within an app. More specifically, if an Activity and a Service both connect to the same service (i.e. Google Play services' location service), I'd like them to share the same binder connection, and if both get onDestroy, I'd like to unbind. The best solution I've come up with is to launch a coroutine from the scope's context, delay forever, and catch the cancellation exception:

scopeToBeObserved.coroutineContext.launch {
  try {
    delay(FOREVER)
  } catch (ignored: CancellationException) {
    // ... unbind the service connection ...
  }
}

How do you suspend a bunch of coroutines, (may be under a supervisor or a context) till given amount of time, i.e. to make sure changes have been saved. I don't want to run any of the job under a given period say 5ms. There is no delay function in coroutine context or a supervisor job.

If I have a fixed thread pool of 8 threads (my CPU has 4 cores, 8 hyper threads) and I look at the queue size and is between 0 and 1k, and the processing is CPU bound. My CPU usage is never maxed out (20% ish). Does that mean that adding more threads should necessarily keep my queue size from spiking to 1k? Don't understanding how my cpu isn't maxed out and I can be getting queuing

I am writing a Proxy handler for

suspend

function. How can I get return

Class<*>

of a suspend function. So far I know the last parameter of

suspend

function is a

Continuation<T>

. How can I get

Class<T>

from

Continuation<T>

Hi! Sorry if this is a silly question, but I sort of need to be sure about this. I’m looking to implement Channels for some script we’re using, but the Experimental annotation sort of makes me a bit anxious. How experimental are

CoroutineScope.produce()

,

ReceiveChannel<E>.consumeEach{}

and

ProducerScope.send()

? Is there any way to know? I’ve got a working implementation now, but knowing that it may break without warning is a bit scary. Also, is this the wrong place to ask this question? Is the discussion boards a better place?

Hi All, can anyone help me how to write unit test for this below method?

fun pollMessages(): Flow<Message> {

        return channelFlow {

            repeat(noOfReceivers) {
                launch {
                    val messages = sqsClient.fetchMessages()
                    messages.forEach {
                        send(it)
                    }

                }
            }
        }
    }

Maybe a vague question for all. What all the benefits I will get if rewrite my Kafka consumer using non blocking I/O and coroutines. Will there be any difference in CPU usage, thread usage compared to my blocking version of the method?

Hi, is it technically possible to resume continuation twice when calling suspendCoroutine { cont -> cont.resumeWith(...) }?

Stupid question: Using GlobalScope in the Application class of an Android app is the same than creating a specific scope in the Application class, isn't it ?

PublisherAsFlow ignores the CoroutineContext, so that it make no sense to call flowOn on it. Is it a bug? I have opened an issue for this: https://github.com/Kotlin/kotlinx.coroutines/issues/1765

Hello, I'm starting to use

Flow

in my production code, but it seems that my understanding of it and coroutines in general isn't sufficient. I reduced my project code to a simple reproducible test example. I'm using

TestCollector

class to collect values, emitted by the

Flow

:

class TestCollector<T>(scope: CoroutineScope, flow: Flow<T>) {

    private val collectedValues = mutableListOf<T>()
    private val job = scope.launch { flow.collect { collectedValues.add(it) } }

    fun assertValues(vararg values: T) = run {
        val valuesList = values.toList()
        if (valuesList != collectedValues) {
            fail("\nExpected values: $valuesList\nCollected values:$collectedValues")
        }
        this
    }

    fun dispose() = job.cancel()
}

fun <T> Flow<T>.test(scope: CoroutineScope) = TestCollector(scope, this)

This is my test itself:

@Before
    fun setup() {
        Dispatchers.setMain(Dispatchers.Unconfined)
    }

    @Test
    fun testFlowCollector() = runBlockingTest {
        var firstEmit = true
        val channel = ConflatedBroadcastChannel(0)
        val testCollector = channel.asFlow().onEach {
            if (firstEmit) {
                launch {
                    channel.send(1)
                    channel.send(2)
                }
                firstEmit = false
            }
        }.test(this)

        testCollector.assertValues(0, 1, 2)

        testCollector.dispose()
    }

So basically I'm using blocking test and what I'm trying to do, is launch those

send

s only on the first emit. What I'm expecting to collect are all those three numbers

0, 1, 2

sequentially. But the test fails:

java.lang.AssertionError: 
Expected values: [0, 1, 2]
Collected values:[0, 2]

For some reason, the emission of

1

is getting lost. I'm trying to understand what's going on. Most probably I'm just misunderstanding/misusing the flow and the channel. Maybe there are coroutine experts who may explain it to me. 🙂

I'm trying to test some code which uses coroutines in my common project. How can I do so if I want to

.join()

a

Job

? What I have right now looks like this:

@Test
fun someTest {
    myObj.doSomething().join()
}

My problem is that I can't call

join()

if I want to wait for the coroutine to complete because I have no

CoroutineScope

but there is no

runBlocking

in a common project which what I usually use in my jvm tests. What can I do instead to solve this problem?

What'd be the most idiomatic way to have a regular, non-suspend function, launch a coroutine in the scope in which it was called?

Hello, I'm a newcomer to Kotlin Flow (and my understanding is, unsurprisingly, limited) 🙂 ! I'm wondering if there's an elegant way (operators ?) to capture KeyboardInterrupt (Ctrl+C) exceptions and/or SIGKILL signals (and by scope) ? In order to implement treatments such as graceful reload for example. Thanks for your help 🙂

Hi guys, Can someone help me understand what’s going on with this code? I’m trying to create a flow that merge the responses of some requests and emit a new value every time a response is retrieved. For some reason this code crashes after the first request, printing the following log:

Doing first request
retrieving first request
Exception in thread "main" kotlinx.coroutines.JobCancellationException: Parent job is Completed; job=ScopeCoroutine{Completed}@383534aa

Has someone any ideia of what am I doing wrong?

Hey, I'm wondering if there is an operator on

Flow

that achieves the following: If the Flow has not produced at least one value in a given timeframe (x miliseconds), produce a default value followed by any values the Flow might produce afterwards. In contrast to a solution with

debounce

however, if the Flow has produced a value before the timeframe, the default value will just be ignored and the value will be emitted immediatly.

Is it normal that my ~10 lines of Flow code got decompiled(Show bytecode -> decompile) to 20k lines of code with billion nested calls?

I wrote this simple extension function to endlessly subscribe to a channel (the subscriber’s scope is always <= the sender’s scope):

internal suspend fun <E> ReceiveChannel<E>.forEach(action: (E) -> Unit) {
    val iterator = iterator()
    while (iterator.hasNext())
        action.invoke(iterator.next())
}

I was surprised that such an extension didn’t already exist—things like

consumeEach

only consume existing items in the Channel and then cancel. Am I either missing a function that does what I want, or missing a reason that I shouldn’t do this?

How do you deal with multiple "chained" asynchronous operations when using flow? Example:

fun foo(): Flow<ResultA>

and each

ResultA

has a

fun getElements(): Flow<SubResult>

. The

getElements

is tied to the same execution context as the containing flow and has to be cancelled with it.

I recently started seeing a crash in Crashlytics for my Android app regarding a Job getting cancelled, but it has no stacktrace. It happens when backgrounding the app, but only very rarely, so it's very difficult to track down what's actually happening. Here's the exception (it's just one line):

Fatal Exception: JobCancellationException: Job was cancelled

Is there any sort of global debugging I can turn on to try and track this down? If so, are any of them safe for a production environment?

Hey, I need help with

Flow

🙂 I have a

Flow<State>

that represents an application (UI) state. Now what I want to do is convert this into a flow with certain "side effects" on parts of the state. Here's example code:

data class State(
    val string: String? = null,
    val number: Long = 0
)

val flow = flowOf(
    State(
        string = "Hello world",
        number = 0
    ),
    State(
        string = "Hello world",
        number = 1
    ),
    State(
        string = null,
        number = 1
    ),
    State(
        string = "Hello world 2",
        number = 1
    )
)

val flow2 = flowOf(
    flow.mapNotNull { it.string }
        .onEach { println("string $it") },
    flow.map { it.number }
        .onEach { println("number $it") }
).flattenMerge()

val job = flow2.launchIn(GlobalScope)

The thing is, in the isolated example this works however in my real application I have the following behaviour: 1. The first State object triggers both

onEach

, so the output is

string

and

number

. 2. Any subsequent changes to

State

will always only trigger the second

onEach

, which is

number

. (it is always the last Flow

onEach

that is triggered here. So if there are three inner Flows then only the third will be called) 3. If I use

flattenConcat()

instead of

flattenMerge()

then only the first

onEach

will be called for every State change. I'm lost here. Any ideas? Why am I doing this? Imagine that later a

distinctUntilChanged()

is added before each

onEach

so that the side effect is only triggered when the selected value of the state has changed. But first of all this should even work without

distinctUntilChanged()

. Any ideas?

Hi, all the

.catch { }

operator of

Flow

should not get one

NPE

that might occurs inside it?

Does it make sense to add the

flowOn

here:

inline fun <reified T : Any, reified R : Any> Flow<T>.concurrentMap(
	dispatcher: CoroutineDispatcher = <http://Dispatchers.IO|Dispatchers.IO>,
	crossinline mapFun: suspend (T) -> R
)Flow<R> = flatMapMerge {
	flow { emit(mapFun(it)) }.flowOn(dispatcher)
}

?

Hi guys I have a questions, I want to inject using dagger a suspend lambda like this:

suspend () -> String

however when I try to compile it says: cannot provide dependency without provides annotation

Does anyone know the best practice way to do a flatmapMerge on a Flow that runs every map in parallel ? It seems to be sequential by default.

3️⃣ Just a regular coroutine function with async's? ✔️

what is the proper way of sharing a flow ?

Is there a way to configure

suspendCancellableCoroutine

to cause

.join()

after cancellation to suspend until the Continuation is resumed / the background work is finished, similar to how things work when using a

withContext

to switch threads? I'm curious if I can get these to behave the same:

private suspend fun runViaThread(): String {
  return suspendCancellableCoroutine { continuation ->
    thread {
      println("Starting background work")
      Thread.sleep(2000)
      println("Finished background work")
      continuation.resume("done")
    }
  }
}

private suspend fun runViaWithContext(): String {
  return withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
    println("Starting background work")
    Thread.sleep(2000)
    println("Finished background work")
    "done"
  }
}

@Test fun testViaThread() {
  /*
  Prints:
    Starting background work
    Before join "coroutine#2":StandaloneCoroutine{Cancelling}@5c7fa833
    After join "coroutine#2":StandaloneCoroutine{Cancelled}@5c7fa833
    Finished background work
   */
  runBlocking {
    val job = launch(Dispatchers.Default) { runViaThread() }

    delay(1000)
    job.cancel()
    println("Before join $job")
    job.join()
    println("After join $job")

    delay(3000)
  }
}

@Test fun testViaWithContext() {
  /*
  Prints:
    Starting background work
    Before join "coroutine#4":StandaloneCoroutine{Cancelling}@2b98378d
    Finished background work
    After join "coroutine#4":StandaloneCoroutine{Cancelled}@2b98378d
   */
  runBlocking {
    val job = launch(Dispatchers.Default) { runViaWithContext() }

    delay(1000)
    job.cancel()
    println("Before join $job")
    job.join()
    println("After join $job")

    delay(3000)
  }
}