So I'm trying to wrap my head around interop between traditional threads and coroutines. I have a pair of reader/writer threads performing blocking operations on a serial port (using a Java library that naturally, has no concept of coroutines). I would like to execute a suspendable callback for each received message on the serial port, with some ordering guarantees (e.g, callback for message N+1 must not execute before the callback for message N has finished). Below is a toy example that demonstrates roughly what I'm doing. I have a channel that serializes the messages to be processed and a coroutine that consumes the messages in order. To actually put messages into the channel though, I launch a new coroutine just to send the message. That seems kind of weird and has me wondering if I'm overlooking something obvious.

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.consumeEach
import kotlinx.coroutines.delay
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlin.concurrent.thread

var readCount = 0
fun blockingSerialRead(): String {
    val message = "MSG$readCount"
    readCount += 1
    // Pretend to take some time, blocking.
    Thread.sleep(500L)
    return message
}

fun main(args: Array<String>) {
    runBlocking {
        val messages = Channel<String>()
        thread(start = true) {
            while (true) {
                val message = blockingSerialRead()
                println("Read message $message")
                launch {
                    messages.send(message)
                }
            }
        }

        // Start a coroutine that does work on each received message
        launch {
            var takeExtraTime = false
            messages.consumeEach {
                // Do some suspendable work with the message
                // For example, maybe we'll send a message back
                // and wait for the response.
                if (takeExtraTime) {
                    // Pretend to do some expensive suspendable work
                    delay(1000L)
                    takeExtraTime = false
                } else {
                    takeExtraTime = true
                }
                println("Processed message $it")
            }
        }
    }
}

Trying to understand thread safety in coroutines… Will there be threading issues with this piece of code:

coroutineScope {
    var count = 0
    val mutableList = mutableListOf<Int>()
    launch {
        while(isActive) {
            mutableList.add(count++)
            delay(1000)
        }
    }
    launch {
        while(isActive) {
            delay(10000)
            mutableList.take(10).also(mutableList::removeAll)
        }
    }
}

I'm trying to synchronize some calls in an okhttp authenticator and it works fine, but I'm having troubles testing it. To simplify things I reduced it to such test (in 🧵)- the important part is that

anImportantBlockingCall

resembles

okhttp.Authenticator.uthenticate

- it's a blocking call, which in my case will run a runBlocking coroutine which will synchronise a few things and wait for them. So it really is important that in this test case this remains blocking. 🧵

when running code with

withTimeout

is it possible to get a stacktrace of the thread where the timeout occurred when it occurred?

Hello, i have a question regarding the use of runBlocking in production code and what is causing my intermittent failure in the following code snippet

Hi! I'm trying to migrate from 1.5 to 1.6 and I have a question regarding

UnconfinedTestDispatcher

. Can I use

UnconfinedTestDispatcher

as a

StandardTestDispatcher

but that it does the

advanceUntilIdle

automatically for me? Am I missing something here? I know that I'll lose some flexibility but in a lot of my tests I don't need that flexibility.

How could you create

sharedFlow

builder with the same signature like the cold

flow

builder:

fun<T> sharedFlow(block: suspend FlowCollector<T>): SharedFlow<T>

? It always requires a

CoroutineScope

...

HI! I am fairly fresh to Kotlin and new to coroutines and flow. I am coming from Java backend so am familiar with threads and completable futures. I am wondering about one scenario, thought I ask here before doing dozens of tests myself: Given external Java API that returns

CompletableFuture

of different entities, for example:

getFoo: CompletableFuture<List<FooDto>> 
getBar: CompletableFuture<List<BarDto>>
getWhatever: CompletableFuture<List<WhateverDto>>

Would it be possible to use coroutines to trigger each of the call in parallel, and combine final object to be:

Object {
    foo: List<FooDto>
    bar: List<BarDto>
    whatever: List<WhateverDto>
}

It is not so easy and obvious to solve with `CompletableFuture`s in Java, maybe it would be easier with

coroutines

? 🤔 The point is I need to fetch multiple different entities before can calculate the final object. I can easily fetch them in sequence, but they are not dependant on each other, so should really be fetched in parallel

Hi all, I was trying to use

callbackFlow { }

but the code inside the block never gets called, am I doing something wrong?

hey got a question about the suspension point of channels. Given the following code snipet

class SimpleChannelTest {
    val simpleChannel = Channel<Int>()

    fun simpleFunction() {
        runBlocking {
            launch {
                simpleChannel.consumeEach { println(it) }
            }

            var x = 0

            repeat(10) {
                simpleChannel.send(x++)
                println("sent Message")
            }
            simpleChannel.close()

        }
    }

I would expect each time send is called , the main coroutine will suspend until the data is received, however I end up with this as the result

0
sent Message
sent Message
1
2
sent Message
sent Message
3
4
sent Message
sent Message
5
6
sent Message
sent Message
7

just wondering why this is?

I want to become a Coroutines expert. I would like a resource (strongly prefer a book) that starts from the basics, and thoroughly goes until the very end. I found this book but it was published in July 2018, and I feel like a lot has changed, and am worried about learning out of date information. I took the tutorial on the Kotlin website, but I felt it went too fast on certain ideas, and didn't expound enough. I also learn better from many worked examples, as is common in books. I also want something more comprehensive than those tutorials. Any tips?

Hi, there. I am writing a network library for Android in Kotlin in which I used a lot of coroutines functionalities inside. So it’s possible to use the library from Java code, suspend functions are wrapped and hidden by callback base APIs. I found a really critical issue there. The problem happens when a user used the library with Future. If the client wrap the callback function with calling the get() method of Future in the UI thread,

delay

function call which is used inside the Library implementation never complete. As a result, Future.get() blocks forever. I know it’s not recommended to call Future.get() is not recommended practice. But, I would like to avoid this issue because it’s too disastrous given library clients has no idea if the

delay

suspend function is used under the hood. The following is the complete example code that reproduces the issue. https://gist.github.com/bananaumai/da23494ed2ba1d22c39af23a7c07ba2f#file-mainactivity-kt Any advise and/or recommendation would be appreciated. Thanks in advance.

I’m having a hard time with runTest because it makes things pass that should not pass. For example it just runs everything in sequence:

@Test
  fun test() = runTest {
    repeat(5) {
      launch {
        println("enter")
        Thread.sleep(Random.nextLong(10))
        println("exit")
      }
    }
    println("start yield")
    yield()
  }

Will print:

start yield
enter
exit
enter
exit
...

Is there any way to use runTest and having parallelism?

I’m collecting from a flow like so, as recommended:

lifecycleScope.launch{
    lifecycle.repeatOnLifecycle(Lifecycle.State.STARTED){
        flow.collect{
            ...
        }
    }
}

I would like the collection, and the entire coroutine to stop when the flow completes. But

repeatOnLifecycle

never returns, even after

collect

returns, and if the activity is stopped and restarted, collection is launched again. It seems I need

runOnceOnLifecycle

.

A little bit of a discussion based around method naming. I essentially have an interface with methods like this. Not all methods have a "oneshot" and "flowable" version, but a lot of them. Some of them will be exclusively oneshot or flowable. But idk how I feel about the names. I could just have a suffix of

Flowable

but that smells like hungarian notiation... but I do remember a book/blog post (effective java?) that said it should be clear when a method is doing stuff with threads/async. Thoughts?

interface MyRepository {
    suspend fun getLatestBooks(): List<Book>
    fun getLatestBooksWithLiveUpdates(): Flow<List<Book>>
    suspend fun getBookDetail(): BookDetail
    fun getBookDetailWithLivesUpdates(): Flow<BookDetail>
}

Has anyone here been able to use

limitedParallelism

on Kotlin/Native? When I try to use it I just get a

ClassCastException

. More details here: https://github.com/Kotlin/kotlinx.coroutines/issues/3223

what is the idiomatic way to implement a resources pool with Flows? let's say it's a list of numbers 1..10 that can grow and shrink resources.borrow { resource -> ... } then once a resource is done being borrowed, it is returned to the pool, if no resources are available, it waits until another borrow is completed a secondary thread can add resources or remove resources from the pool if they are no longer valid I think this is possible with Channels, but would like to use a cold flow.

What's the proper way to test if a flow collector received a value emitted by a sharedflow? using

runTest

doesn't seem to wait for a collector running in another scope to fully process the emitted value before returning. Keeping everything in the runTest scope and context just hangs the test because Flow.collect on a sharedflow never returns.

how would you pause a

timer

type flow? it’s not really idiomatic, is it? or maybe akin to backpressure, i.e. downstream telling upstream to stop?

suspend fun debug() {
    println("${coroutineContext[STreeEvent]}")
}
fun main() = runBlocking {
    val test = SSubject(0)
    SComponent().apply {
        println("Component: $this")
        println("Test: $test")
        test(this, BEFORE) {
            println("${coroutineContext[STreeEvent]}")
            debug()
        }
...

Can someone help me understand why :

println("${coroutineContext[STreeEvent]}")
            debug()

prints:

null
1212899836

Why on earth would the context change between these two places?

One question guys! in my

ViewModel

I have injected a

CoroutineDispatcher

something like

class FooViewModel @Inject constructor(private val dispatcher: CoroutineDispatcher){}

Then when I'm trying to do the test I'm seeing that the test and the code in the

ViewModel

are working on a different thread. I know it because I've added some logs

println("[${Thread.currentThread().name}] viewModel")
println("[${Thread.currentThread().name}] test")

And in the console I see :

[Test worker] test

[Test worker @coroutine#3] viewModel

In the test I'm doing it as :

private val testDispatcher = UnconfinedTestDispatcher()
private val testCoroutineScope = TestScope(testDispatcher)

@Test
fun test() = testCoroutineScop.runTest {
 val viewModel = FooViewModel(testDispatcher)
 ...
}

What I'm missing?

Hi guys! I am trying to learn coroutines, coming from Java background and CompletableFutures: Does this snippet make sense?

private suspend fun fetchExternalData(futures: List<CompletableFuture<Reply>>): List<Reply> = coroutineScope {
        futures.map {
            async {
                it.join()
            }
        }.awaitAll()
    }

I have a list of CompletableFuture, which I would run async and do action when all of them complete. Is this the right place where I call

join

(or

get()

) to get reply? The idea is to get a bunch of request, send them in parallel (fork) and wait for all of them (join), then combine the result

Hi, i recently updated to to newest coroutine version (1.6.0) and i am now experiencing some strange behaviour when it comes to unit testing using turbine. While the following minimal example is working fine with the now deprecated

TestCoroutineDispatcher

, it fails (

TimeoutCancellationException

) with the new coroutines version and the

UnconfinedTestDispatcher

.. any ideas how this issue can be resolved?

// ViewModel
    val loading = MutableStateFlow(false)

    fun onReloadClicked() {
        viewModelScope.launch {
            loading.value = true
            // load data
            loading.value = false
        }
    }

// Test Code
    private val testDispatcher = UnconfinedTestDispatcher()    

    @Before
    fun setUp() {
        Dispatchers.setMain(testDispatcher)
    }    
    
    @After
    fun tearDown() {
        Dispatchers.resetMain()
    }

    @Test
    fun testLoading() = runTest(testDispatcher) {
        viewModel.loading.test {
            assertFalse(awaitItem())

            viewModel.onReloadClicked()

            assertTrue(awaitItem()) // fails with TimeoutCancellationException
            assertFalse(awaitItem())
        }
    }

Hi guys, I have a

List<Flow<T>>

what is the best way to keep collecting from each

Flow<T>

?

Hello everyone. I wanted to know if someone had some concrete data about the performance of Coroutines vs RxJava? I have used coroutines myself for several years and I prefer it over RxJava but I am working to migrate some codebases from Rx into coroutines and I need to convince the right stakeholders about this change. My plan was to put together a small sample program and run multiple scenarios. But if someone has some data already that they would be willing to share, that would be awesome 🙂 .

Hey guys, I’m making 4 separate DB calls sequentially, getting the results from each and returning an AggregatedResult(call1, call2, call3, call4) object with the results to the client. I’m trying to figure out how I can parallelize the DB calls and return the AggregatedResult() object once all DB 4 calls complete. Any hints as to which coroutine function to use?

hi there guys I`m working on code that should call from lock context suspend functions

kotlin.concurrent

@kotlin.internal.InlineOnly
public inline fun <T> Lock.withLock(action: () -> T): T {
    contract { callsInPlace(action, InvocationKind.EXACTLY_ONCE) }
    lock()
    try {
        return action()
    } finally {
        unlock()
    }
}

but it says

The 'checkRegistry' suspension point is inside a critical section

Newbee with corutines\suspend so - can anyone share document reference that will help me to understand what I’ve miss

I'm trying to model a 'lazy suspend' task. For the moment, it's encoded as:

val task = scope.async {

foo()

}

suspend fun bar() {

// All calls require the task to be over,

// however only the first one(s) will really suspend,

// since it will be ready for the others

val foo = task.await()

doSomething(foo)

}

The problem is that this requires a

scope

to be available when the task is instantiated. Since I do not care whether the task starts when it is instantiated or when the first call to

bar

is made, is there a way to adapt this pattern without requiring the

scope

?

Hello everyone. Some people in my team asked me for some performance numbers between coroutines and RxJava so I wrote this little app to measure performance between the two. https://github.com/CRamsan/coroutine-rx-performance I hope this is useful for people here. But the gist of what I found is that Coroutines can be a bit slower compared to Rx. But RxJava memory consumption is much higher. For me, this is another strong argument for coroutines.