Is there a difference between 

<http://Dispatchers.IO|Dispatchers.IO>.limitedParalellism(100)

and

Dispatchers.Default.limitedParalellism(100)

?

Trying to come up with good way to suspend until the first emission from the flow which is launched in parallel. app code:

class Feature  {
   private val scope = CoroutineScope()
   private val _flow = MutableSharedFlow<Int>()
   
   fun start() {
     scope.launch { flow.emit(1); flow.emit(2) }
   }

   val flow: Flow<Int> = _flow
}

test code:

should("do something") {
 val awaitFirstEmit = feature.flow.onStart { feature.start() }.first()
 awaitFirstEmit shouldBe 1
}

Is it OK to use

onStart

like this, or will this produce flakiness? the important thing here is I want to ensure that first item is not lost.

New drinking game: When migrating from

runBlockingTest

to

runTest

. For each test you get to pass, take a drink!

I thought I'd get started with the actors, but I notice they are only implemented for the JVM. Shame. Any chance this might change in the future?

Hi everyone! How Flow.sample() should be tested with kotlinx-coroutines-test 1.6.0? Every time I run attached test it runs infinitely 🙈

Is there a guide how to test in memory room database with coroutines especially using new test coroutine api added in kotlin 1.6 ?

Hello everyone! I'm trying to migrate to 1.6.10 coroutines-test and have a ViewModel that contains a flow using the

mapLatest

operator. More details in the thread with code examples.

how to combine two `SharedFlow`s into

Flow

that emits when either

SharedFlow

emits?

Hello there, I was recently bitten by the

MutableSharedFlow

default constructor + tryEmit issue. I was reading through the issue and seems to me that the docs added here are not 100% accurate? It seems to me that they indicate that the

tryEmit

call can succeed, but in reality it will always fail (aka not emit anything) but can return true or false depending on having subscribers or not, isn’t it? cc @elizarov Wondering if it would make sense to have a quick PR there to simplify the texting, could volunteer for that 🙂

Hi, I'm Ini... I recently hopped on the kotlin boat ⛵ I'm using 'kotlin apprentice' as a reference - by R. Wenderlich - currently I'm studying coroutines... My issue: kotlinx dependency isn't building...so I used Retrofit dep (build.gradle) cuz Retro includes coroutines out of the box 🎁 Retro dep builds...autocomplete for kotlinx.coroutines.* works, yet features are red - dispatchers, launch, run blocking etc... I'll search to see if anyone else has this problem here anyways I'm proud to meet you guys and I hope we can do some work - good work together! #coroutines

Hi all, from the doc

*Warning:* Never collect a flow from the UI directly from *launch* or the *launchIn* extension function if the UI needs to be updated. These functions process events even when the view is not visible. This behavior can lead to app crashes. To avoid that, use the *repeatOnLifecycle* API as shown above.

What is the safest way to collect stateFlow in compose since we don’t have access to `*`repeatOnLifecycle`*` since it is not recommended to collect using

launch

?

Is there a nice way to combine 2 state flows into another one without using

stateIn

which requires a

CoroutineScope

?

val s1: StateFlow<Int>
val s2: StateFlow<Long>

val s3: StateFlow<Long> = combine(s1, s2) { v1, v2 ->
  v1.toLong() + v2
}
// ^^^ can't do that, combine results in `Flow<Long>`, 
// must do stateIn...

Am I missing something or there's no version of zip for 3,4,5, iterable flows? there is a top-level

combine

, but no top-level

zip

. I have a

List<Flow<Int>>

and want to zip them, what is my best option?

I'm migrating a test to coroutines 1.6.0 which should check that a suspending function throws an exception. I'm using AssertJ for assertions. This is what the test looked like before:

@Test
fun `when no gateway from scanning present then throw`() {
    every { state.gateway } returns null

    assertThatIllegalStateException().isThrownBy { runBlocking { useCase.execute(Params()) } }
}

This is what it looks like after the migration, according to the official migration guide:

val exceptions = mutableListOf<Throwable>()
    val customCaptor = CoroutineExceptionHandler { _, throwable ->
        exceptions.add(throwable)
    }

    @Test
    fun `when no gateway from scanning present then throw`() = runTest {
        every { state.gateway } returns null

        launch(customCaptor) {
            useCase.execute(Params())
        }
        advanceUntilIdle()

        assertThat(exceptions).hasSize(1)
            .first().isInstanceOf(IllegalStateException::class.java)
    }

However, the exception is for some reason not caught by the

customCaptor

, but thrown in the test and the test fails.

I don’t quite get it with the docs on

Deferred

For

await

, it says:

[…] If the Job of the current coroutine is cancelled or completed while this suspending function is waiting, this function immediately resumes with CancellationException. […]

Why would the await function resume with a CancellationException if the job completes while await is waiting?

Hi, im new to kotlin coroutines and after a look in the docs i have a question, in case i have spring boot app (actually whatever app it doesnt matter) and lets say i have a rest api, something like this:

//@RestController
//@RequestMapping("${v1}/ping")
public class TestController {

    @GetMapping(value = ["", "/"])
    public suspend fun testAsync(): Deferred<ApiResponse> {
        coroutineScope {
            // add here launch, async etc   
        }
    }

}

For every api should i add a coroutineScope to achieve structured concurrency ?

Hello folks, I have a case where I’m collecting a flow to do some operations but on every emission I need to do a side effect and have another different flow to observe until there’s a new value emitted. I was thinking on using flatMapLatest but that would multiply the initial emission every time the internal one emits something + would get delayed, so I though about doing a

onEachLatest

sort of 🤔 Sharing here for feedback or in case I’m missing something 😬

@ExperimentalCoroutinesApi
public fun <T> Flow<T>.onEachLatest(action: suspend (T) -> Unit): Flow<T> = transformLatest { value ->
    action(value)
    return@transformLatest emit(value)
}

the only real difference with

onEach

is the

transform

->

transformLatest

but I guess that should work, right?

Hello guys. I am struggling with

runBlocking

. The documentation says that it should be used in

main

, tests and libraries to bridge two worlds. I found zero information about structured concurrency within plain command line applications. Lets say I have a tool which migrates data from one source to another and I would like to use concurrent executions for some parts of the whole migration process. Now, should I define

runBlocking

somewhere "in the middle", or how to use coroutines in the world where no life cycle managers exists? Thanks

Hi guys, I have a question, in case i use async inside a coroutinescope but I await it outside the scope is it ok to do so or is there something like a deadlock where the actual computation will happen inside the scope and the rest code will complete sequential no matter where I await for it ?

I’m having a weird issue where IntelliJ IDEA always forces this import when attempting to resolve

launch{}

->

import javafx.application.Application.launch

I have a test where a flow should emit with delay, but it doesn't emit:

@Test
fun `test flow`() = runTest {
    val eventUpdateService = mockk { 
        every { subscribeToEvents() } returns flowOf(SomeEvent).onStart { delay(1000) } 
    }
    val presenter = Presenter(
        UnconfinedTestDispatcher(),
        eventUpdateService
    )
    presenter.attachView(view)
    advanceTimeBy(2000)
    verify { view.displayEvent() } // fails because the flow didn't emit
}

class Presenter(
    coroutineContext: CoroutineContext,
    val eventUpdateService: EventUpdateService    
) {

    val coroutineScope = CoroutineScope(coroutineContext)

    fun attachView(view: View) {
        coroutineScope.launch {
            eventUpdateService.subscribeToEvents().collect { event ->
                view.displayEvent()
            }
        }
    }
}

What is actually the difference between

StandardTestDispatcher

and

UnconfinedTestDispatcher

in a case where you always inject a dispatcher in the system under test? Let's say we have a presenter, like in my post above, which always uses one dispatcher for all coroutine calls (e.g.

Dispatchers.Default

). For tests it gets injected a test dispatcher, so during tests every coroutine launched inside the presenter uses this injected dispatcher. Is there a difference which dispatcher is used then?

Im struggling to understand why .map and .combine works differently with .stateIn in my tests The test testing .map works as expected, the one testing the combine does not.

hi, new to kotlin, so bear with me... i'm building a library which relies on coroutine context (via suspend functions) to propagate "sticky" information such as tenant id etc. what is the best way for the caller of my library function to pass the tenant information on the context?

suspend fun myLibraryFunction(query : Query) : QueryResult{
  val requestContext: RequestContext = currentCoroutineContext()[RequestContext]!! //expecting a dedicated context element
  val tenantId = requestContext.tenantId
  do something with tenantId
}

suspend callerFunction () {
  val requestContext = RequestContext(tenantId)
  CoroutineScope(requestContext).async { myLibraryFunction(query) }.await()
}

^^^ This is what i came up with, but it seems redundant to create a new coroutine (i.e async) i just want the user to be able to set the requestContext on the current context and that the called function will inherit that something like

suspend fun callerFunction () : QueryResult {
  val requestContext = RequestContext(tenantId)
  currentCoroutineContext().plus(requestContext) //i know it's immutable so won't work
  myLibraryFunction(query)
}

Hi all, from the doc WorkManager invokes ListenableWorker.onStopped() as soon as your Worker has been stopped. Override this method to close any resources you may be holding onto. How could we override

onStopped

since on

CoroutineWorker

it is defined

public final override fun onStopped() {}

Can someone help me understand.. If I want to create a

StateFlow

, I can call

MutableStateFlow()

. But, if I want to convert a Flow into a

StateFlow

, I call

stateIn()

.

stateIn()

requires a

CoroutineScope

. But

MutableStateFlow()

doesn’t. Why is that?

Is there a reason why combine flow would not emit end result?

val x = combine(y.invoke(), z.invoke()){ y+z }

Both

y

and

z

are flows of same type and they seem to not throw any errors.

is there a function in the coroutines lib that transforms a

List<Flow<T>>

to

Flow<List<T>>

?