Hello,

@Test
    fun example() = runBlocking {
        val a = produce {
            var i = 0
            while (isActive) {
                send(i++)
                delay(100)
            }
        }
        val b = a.broadcast(start = CoroutineStart.DEFAULT)

        launch {
            b.consumeEach {
                println("A: $it")
            }
        }
        
        b.close()

        delay(1_000)
    }

When I close b broadcast channel I get an exception:

Exception in thread "Test worker @coroutine#3" kotlinx.coroutines.channels.ClosedSendChannelException: Channel was closed (Coroutine boundary)

Is this the expected behavior?

📣 📣 📣 

kotlinx.coroutines

 version 1.3.3 is here! •

Dispatchers.Main

doesn’t use

ServiceLoader

anymore, preventing I/O on Main thread during initialization and annoying “Module with the Main dispatcher is missing” exceptions • Reworked

Dispatchers.Default

and

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

, CPU consumption is significantly reduced •

Flow.merge

operator • Significant performance improvements in

Flow

,

Channel

and

Job

classes • Improved stacktrace recovery and debugging experience And a lot more, full changelog: https://github.com/Kotlin/kotlinx.coroutines/releases/tag/1.3.3

If I want to reproduce the behavior of Rx PublishSubject by using a ConflatedChannel. Should I be using the

offer

or

send

function to send a value to the channel and why?

Trying to test my coroutines with unit tests (JUnit Mockito) and referring kotlin coroutines repo i was using runBlockingTest and it works as expected to advance the execution . But when I run tests concurrently in a suite they are flacky as they are sharing one dispatcher (my assumption) in view model. When looking at the repo there is TestCoroutineDispatcher but it looks like the API to replace actual dispatcher with that one is only .setMain which replaces Dispatchers.Main. I am using Dispatchers.IO so the mentioned api seems of no use . Is there a way to test coroutines with this set up without injecting the dispatcher into view model? @Test fun `test loadMenu calls interactor getMenu`() = runBlockingTest { viewModel.loadMenu() Mockito.verify(interactor).getMenu() } fun loadMenu() { viewModelScope.launch(Dispatchers.IO) { try { interactor.getMenu().menuDetails.menu.run { updateMainCatList(this) } } catch (ex: Throwable) { //todo handle exceptions } } }

if I have a java library that requires me to return a CompletionStage<T> for async calls, and I want to perform the call using a coroutine and then bridge back to java CompletionStage, how do I do that?

Hi, I'm not very familiar with coroutines yet, but I'm doing a code review on code that uses them. I'm trying to figure out something:

someList.map { async { /* stuff */ }.forEach { it.await() }

My immediate reaction is to refactor it to

someList.map { async { /* stuff */ }.awaitAll()

But I see that

awaitAll

allocates an array, so am I actually hurting performance instead of improving things? Is

forEach { it.await() }

a more correct way to do

await

on a collection of

Deferred

?

is there a way to do fan-out but with flow?

I'm probably missing something obvious... :rubber_duck:

Looks like

ReceiveChannel.last()

is deprecated, it links to https://github.com/Kotlin/kotlinx.coroutines/issues/254 but what do I actually have to replace it by?

Similar question, how do I now connect the output of one channel to the input of another? I think

output.consumeEach { input.send(it) }

is correct but that's also deprecated.

in general, flows are the intended way to handle everything now. if you need a hot flow, take your channel and do

chan.asFlow()

first

Okay. I haven't had any luck on adding exception handling to my coroutine. Anyone out there that knows how to prevent this from crashing?

viewLifecycleOwner.lifecycleScope.launch {

        val work1 = async { getThingsDone(43) }
        val work2 = async { getThingsDoneAgain(123) }
        if (work1.await().isSuccessful && work2.await().isSuccessful) {
            //do my thing
        } else {
            //failure message
        }

Basically, the getThingsDone calls are retrofit network calls. If one of them is still going on and I go into airplane mode, I get a crash. What have I tried? 1.

try {
    viewLifecycleOwner.lifecycleScope.launch {
        val work1 = async { getThingsDone(43) }
        val work2 = async { getThingsDoneAgain(123) }
        if (work1.await().isSuccessful && work2.await().isSuccessful) {
            //do my thing
        } else {
            //failure message
        }
} catch (e: Exception) {
//show failure message
}

2.

viewLifecycleOwner.lifecycleScope.launch {
            try {
                val work1 = async { getThingsDone(43) }
                val work2 = async { getThingsDoneAgain(123) }
                if (work1.await().isSuccessful && work2.await().isSuccessful) {
                    //do my thing
                } else {
                    //failure message
                }
            } catch (e: Exception) {
                //show failure message
            }
        }

3.

viewLifecycleOwner.lifecycleScope.launch {
        val work1 : Deferred<Response<MyType>>
        val work2: Deferred<Response<MyType>>
    try {
        work1 = async { getThingsDone(43) }
        work2 = async { getThingsDoneAgain(123) }
    } catch (e: Exception) {
        //show failure message
    }
        if (work1.await().isSuccessful && work2.await().isSuccessful) {
            //do my thing
        } else {
            //failure message
        }
}

4.

viewLifecycleOwner.lifecycleScope.launch {
    val work1 = async {
        try {
            getThingsDone(43)
        } catch (e: Exception) {
            //show failure message
        }
    }
    val work2 = async {
        try {
            getThingsDoneAgain(123)
        } catch (e: Exception) {
            //show failure message
        }
    }
    
    if (work1.await().isSuccessful && work2.await().isSuccessful) {
        //do my thing
    } else {
        //failure message
    }
}

I would really appreciate some help. As of now this works super nice because I essentially have two async network calls, and my code ONLY continues if both of them complete! But in the off chance that I have a network error (or force one by going into airplane mode during the request), my Android app dies.

Am I doing something wrong here? I am trying to use flow with firebase but I am getting

kotlinx.coroutines.channels.ClosedSendChannelException: Channel was closed

@ExperimentalCoroutinesApi
fun <T> firebaseDataAsFlow(reference: DatabaseReference) = callbackFlow {
        reference.addValueEventListener(object : ValueEventListener {

            override fun onDataChange(dataSnapshot: DataSnapshot) {
                val type = object : GenericTypeIndicator<T>() {}
                offer(dataSnapshot.getValue(type))
            }

            override fun onCancelled(error: DatabaseError) {
                cancel(CancellationException(error.details, error.toException()))
            }

        })
    }

The exception gets thrown whenever

offer()

is called

Hello where and when should

withLock{...}

be applicable?

I am having some trouble fixing a race condition where there are multiple executions on the same function , I was thinking of adding a field marking the status of the object that is still in progress, but is there any other elegant way to address this?

Kotlin coroutine actors are marked @ObsoleteCoroutinesApi , does this mean there is plans to replace it with something in the future?

guys, I want to ask for some help, I feel my problem is very simple but I'm struggling for a while now to find the right way to implement it. the pattern I'm trying to build is: I have a list of items. I want to 'process' the items with <n> concurrency and get a list of all the results back after everything is processed. Right now my attempt is • put all items in channel 'todo' • launch n workers that pull out of channel 'todo' and push the result of their processing into channel 'done' • collect all results from channel 'done' In the code snippet there is a simple example. It works, except for the "log progress" part. If I try to get channel.count() , the channel seems to get emptied. Which is by design, if I understand the docs correctly. But how else would I get the channels current number of items? Generally, is this correctly implemented for this pattern? I love kotlin, but for concurrency, there is waaay to many different ways to do things. All the contexts, dispatchers, channels, flows(?), async, jobs, consumer, etc etc ... Compared to e.g. golang, the concurrency is very complex.

Can we expect a native-mt release based on 1.3.3? That'd be helpful to get the latest and greatest for projects sharing coroutines for iOS and Android (and macOS). Context: libraries in dev phase (#splitties) as I'm experimenting before adding an iOS sample.

Hi, I have a method in repository that is just calling some api request through retrofit and I want to solve a situation when this method is called multiple times at the same time and it fires multiple requests. Ideal solution for me would be to somehow reuse the same request while it runs. I’ve come up with this but I dont know if there are any problems or if it does not make sense at all. Thanks

original function

override suspend fun user(): User {
	return api.user()
}

suggested solution

var userDeferred: Deferred<User>? = null

override suspend fun user(): User {
    if (userDeferred == null) {
        userDeferred = async {
            apiInteractor.user()
        }
    }
    return userDeferred.await().also {
        userDeferred = null
    }
}

For anyone looking to do an MVI ViewModel implemented just with Flow, I swapped in Flow for Rx from a pure Rx implementation. Rx version: https://github.com/kaushikgopal/movies-usf-android Flow version: https://github.com/brendanw/graphql-viewmodel/pull/1/files#diff-f7b91eff2ca316133c697563510a0e91 Flow is still missing a multicast operator, so I copied an implementation from some WIP open source lib by Yigit and Mike N. There is another multicast operator implementation by one of the Rx contributors; I tried it and experienced some heisenbugs -- Yigit and Mike's multicast flow operator has more comprehensive test coverage.

I’m using a channel in a viewmodel as a sort of BehaviorSubject. Do I need to explicitly close the channel when the viewmodel is cleared?

I'm trying to understand issues I noticed in the Default Dispatcher. At one point, I know this was backed by a pool similar to CommonPool with only threads = CPUs, coerced to at least 2 threads. What is the pool size today on JVM? I ran into issues not long ago where our app had been deadlocking on dependent work using the Default Dispatcher, but the same issue did not occur if I migrated the misplaced IO-bound work off to the IO dispatcher. The issues would only occur on the 1% of devices with the fewest cores, but the deadlock would extend for the full five second ANR interval on Android before I made the change. It seems like Dispatchers.Default must have a size smaller than the IO Dispatcher. This seems because Dispatchers.Default is still intended for CPU-bound work, like CommonPool. Is that correct or am I off-base?

In the new native-mt model what would be the equivalent of Dispatchers.IO for iOS in a AppCoroutineDispatchers class with main = Main, computation = Default and io? Would that be

newSingleThreadContext("io")

?

I remember this was asked a few times but I thought there was no definite answer: What is the equivalent of a rxjava PublishSubject? I need a non buffering channel that just ignores events when no one is listening

Has anyone tried sqs message consumer using Kotlin flow???

Hello, World! I'm in a Multiplatform library and I'd like to use

runBlocking

... is it possible?

So I managed to use

runBlocking

... but now... it just freezes on iOS...

So I have this weird problem where a flow’s collector is called when the scope is cancelled. I added some logging to confirm this is the case:

val flow = repository
        .observeStuff(id)
        .map { it.map { it.createViewModel() }}
        .combine(otherFlow) { stuff, otherStuff -> stuff to otherStuff }
        // Don't emit until the settings are loaded
        .onEach { settingsLoading.join() }
        .conflate()
        .flowOn(Dispatchers.Default)
        .broadcastIn(viewModelScope, start = CoroutineStart.DEFAULT)
        .asFlow()

scope.launch {
  flow.collect {
    // Rare crash because the scope is cancelled and the fragment is destroyed
    view.update(it)
  }
}

It really feels like it’s a bug in the coroutines framework because I can’t explain it otherwise I can fix it by adding a

yield

or

if (coroutineContext[Job]?.isCancelled == true) throw CancellationException()

in my

collect

but I really want to figure out what is happening. I’ve only seen it in Crashlytics I added some logging to crashlytics to confirm the scope is cancelled:

CrashReporting.log("isStarted: %b, view == null: %b, isCancelled: %b".format(isStarted, view == null, coroutineContext[Job]!!.isCancelled))

And in the crash report is says:

isStarted: false, view == null: true, isCancelled: true

Does

Flow.collect()

block the parent coroutine from executing the code under the flow until the flow is cancelled?

viewModelScope{
    // code here get executed
    someFlow.collect()
    // code here never executes
}