Is it possible to get

MutableStateFlow<S>

from

MutableStateFlow<S?>

?

i was messing around with the

withTimeout

function and found that if I put a loop, it does not trigger the timeout cancellation exception but if I use a delay statement for the same amount of time, then it triggers the catch block. Could anyone explain why?

Mutex vs StateFlow (sounds strange, but please read on) Suppose you need to synchronize few suspending calls (e.g. DB reads and DB writes) so that only one action is running at a time. You cannot use java synchronization or reentrant lock as it does not correctly work with suspension (https://blog.danlew.net/2020/01/28/coroutines-and-java-synchronization-dont-mix/). It seems like the only suitable option is to use

Mutex

, but it is very slow, if you’ll run the examples from https://kotlinlang.org/docs/shared-mutable-state-and-concurrency.html you’ll notice that

Mutex

is like 100 times slower than atomics. I found a way to use

MutableStateFlow<Boolean>

for suspending calls synchronization and it seems to be 5-10 times faster than

Mutex

, I’m not sure if there are any hidden gotchas with this approach though (see more in the thread).

I know Flow<T>.combineLatest is deprecated as it is just called combine but is there a true combineLatest (in the same vain as transformLatest, collectLatest, mapLatest, flatMapLatest) or does combine already behave this way (I don't believe so because the docs don't mention it)

Hello guys, I'm asking if anyone knows a good website to learn about coroutines

hi! is there a way to convert Flow<ByteArray> to single InputStream, but without storing all emitted bytes in memory?

Hey, I have a buffered channel with max capacity let’s say 10 and BufferOverflow.SUSPEND, and I want to let sender to send new values to channel only when there are 5 items left in the buffer, not immediately when buffer changed from 10 to 9. Is there a way to check the number of buffered items on the sender side? Thanks.

Currently doing a tutorial but when I run this code from the tutorial it runs an error can anyone help

Hello if my presenter is written in java and my datasource is in kotlin and I have used coroutines, how do I connect them? Do I have to use the

suspendCancelableCoroutine{..}

? Then do i need the

launch{}

somewhere? Or from the java class I call the method of my presenter and that's it? I can avoid the launch?

I’m building a spring boot web app that receives “events” via HTTP requests, and does some processing then sends them to AWS EventBridge (serverless message bus). These events are “fire and forget” from the producer side, so my app immediately returns a response to the HTTP message. I’ve got it working as such right now, but the IO to call the AWS service seems to slow down. I’d like to batch up multiple Events into a single call to the AWS service. i.e. 5 separate inbound http requests containing Event1-Event5 result in a single outbound request to AWS. Is BufferedChannels the appropriate abstraction for this?

Hi. I'm writing unit tests for multiplatform project and getting inconsistent coroutines results for Kotlin Native. Perhaps this is not the best way to write tests but it is working for Android platform. Here is simplified version of my test: Test function:

@Test
fun testSuccess() = runBlocking {
    val repository = Repository()
    var actual: String? = null

    val job = repository.getData(TEST_URL) {      
        actual = it
    }
    job.join()

    assertEquals(TEST_MESSAGE, actual)
}

Repository getData function:

fun getData(
    url: String,
    callback: (String) -> Unit = {}
): Job {
    return GlobalScope.launch(dispatcher) { // Dispatchers.Main (Android), NsQueueDispatcher (iOS)
        try {
            val message: String = httpClient.request(...)
            callback(message)
        } catch (e: Exception) {
            println("Exception: $e")
        }
    }
}

Test passes running for Android platform but getting infinite loop for iOS platform. Does anybody know a good way to test a callback function which was invoked within coroutine? Any help much appreciated. 🙏🏻

HI all ... basic question, as I'm still fairly new to coroutines ... I've setup a

SimplePoller

class that will run a

block: suspend () -> Unit

on a given interval. it looks like this (code condensed a bit for posting in Slack):

class SimplePoller(private val interval: Long, private val block: suspend () -> Unit) {

    private var job: Job? = null

    fun start(scope: CoroutineScope) {
        if (job == null || job?.isCompleted == true) {
            job = scope.launch {
                while (isActive) {
                    try {
                        block()
                        delay(interval)
                    } catch (ex: Exception) { cancel() } // stop on errors
                }
            }
        }
    }

    fun stop() {
        job?.let { if (it.isActive) it.cancel() }
    }
}

Most of the time, it's used to poll the

block

while some Android

Lifecycle

is in a certain state. For example:

val myPoller = SimplePoller(60_000) {
    //do some work that needs to be repeated like call a network api, etc
}

lifecycleScope.launch { 
    whenResumed { 
        myPoller.start(this)    
    }
}

Inside the poller, I launch a new job in the passed in scope so that I can "manually" (via a call to

stop()

) stop the polling if necessary. Does this seem like a good approach? It seems to work pretty well. In testing the poller runs while the lifecycle is resumed, and suspends (because the passed in scope uses a magical Android "PausingDispatcher") when lifecycle < resumed. It also seems flexible enough that I could really use it in other, non-android-lifecycle situations:

someScope.launch {
   myPoller.start(this)
}

// And then sometime later:
myPoller.stop()

I don't see anything immediately wrong, but as I mentioned, I'm still new to coroutines, so... 🤷

if I call

lauch

inside a

channelFlow

, what coroutine scope is used for that

channelFlow

? The one that collects from the

channelFlow

?

Are there any coroutine bindings for

Service

in Android? We’re currently using

override fun onHandleWork(intent: Intent)

in a

JobIntentService

and I was wondering if there was a more idiomatic way to make that function Coroutine friendly, as opposed to using

runBlocking { }

in there

Are there any issues in using Flow (MutableStateFlow) with databinding?

I’m new to coroutines and struggling to get my head around the conceptuals while trying to implement a debounce:

override fun onOptionsItemSelected(item: MenuItem): Boolean {
    return when(item.itemId) {
        R.id.refresh -> {
            flow {
                callbacks?.refreshAllTheThings()
            }.debounce(5000)
            return true
        }
        else -> {
            return super.onOptionsItemSelected(item)
        }
    }
}

...however I get the following compilation error:

Not enough information to infer type variable T

I’ve gone and looked at the implementation of

debounce()

but the examples in the comments suggest what I have above should be sufficient. Why does flow need type information if I just want to execute one line of code, at most, every 5 seconds?

Is StateFlow acceptable to use for short running operations where I need a flow of data? State flows don’t complete, so the collection of them doesn’t stop right?

With

StateFlow

, is it safe to access the current state inside a

catch

block?

Hi, I’m experimenting with state flow to see if i can get some sort of reactive stack implementation. My idea is that i have a  class that holds the state, which only one subscriber observes, but multiple classes can update the state. This is what i have so far (in 🧵), but i have some strange behaviour. Calling 

replaceState

 works as expected with the subscriber getting the new value, however, calling 

updateState

 doesn’t. Does any one have a clue why this happens?

I'm writing a suspend function but it's not safe to call concurrently. Usually one would call this method "Not thread safe" but in the coroutine world what do you call this? "Not coroutine safe"? Sounds weird.

Hey, guys,. There is a problem about kotlin coroutine. I used

kotlinx-coroutines-test

library to test my coroutines code. But after I used

TestCoroutineDispatcher

to mock the dispatcher, the coroutine resumed continually until throw an exception with the message:

kotlinx.coroutines.CoroutinesInternalError: Fatal exception in coroutines machinery for DispatchedContinuation[TestCoroutineDispatcher[currentTime=0ms, queued=0], Continuation at xxx.viewmodel.CarListViewModel$loadCars$1.invokeSuspend(CarListViewModel.kt)@1da6ee17]. Please read KDoc to 'handleFatalException' method and report this incident to maintainers

You can find the specific code at https://youtrack.jetbrains.com/issue/KT-45088.

How can I create a single threaded Dispatcher on Kotlin Native for iOS (that isn't the Main Dispatcher)? I want to keep business logic off the Main thread but also not worry about state being frozen between threads, which is a pitfall when using the Default Dispatcher (on a multi core device). A two-threaded delineation between UI/business logic seems like a more manageable, pragmatic middle ground for KMM projects until the new Native Memory model changes the landscape.

I was watching this talk by Elizarov about coroutines (Kotlin Conf 2018), where a downloading example was given. In this post, he explained some problem that existed with a deadlock, and a solution in a 2019 post here. However in this last post, there is a note about how Channels should be considered a "low level primitive" suggesting that flows(?) may be a better alternative. Now my question is, is this example code still considered the optimal way that one would solve this downloader problem, or should the entire channel process optimally be replaced with a different implementation without raw Channels? If so, is there an update on this problem, perhaps by Elizarov himself on how this would be solved with the state of coroutines as of right now? It would be really interesting to see the differences in order to also understand better what problems these new constructs can solve easier than raw channels.

Would it be correct to say that this code:

withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
  foo()
}

is conceptually equivalent as this one:

launch(<http://Dispatchers.IO|Dispatchers.IO>) {
  foo()
}.join()

Since the implementation is very different, I expect that for performance reasons it does something else, but as a developer should I use

withContext

as syntax sugar for

launch().join()

, or is there something else I need to know about?

Hi folks. I need to retrive information from coroutine context (ktor logged user) in an implementation of a function from an abstract java class provided by an external library. I can't implement this function as a suspend function and so i can't access to the coroutine context. Do you have any suggestion?

Hello 👋 I have a question regarding Unit Test and Flow. I'm following the Testing flows Android documentation to test my work. But I'm stuck with the following error

This job has not completed yet

Here is my implementation:

class MyUseCaseImpl(
    myRepository: MyRepository,
    myID: String,
    ioDispatcher: CoroutineDispatcher,
) : MyUseCase {

    override val state = MutableStateFlow<MyState>(MyState.Loading)

    init {
        CoroutineScope(ioDispatcher).launch {
            myRepository
                .requestStuff(myID)
                .map { MyState.Success(it) }
                .catch { state.emit(MyState.Error(it)) }
                .collect { state.emit(it) }
        }
    }
}

interface MyUseCase {

    sealed class MyState {
        object Loading: MyState()
        data class Error(val throwable: Throwable) : MyState()
        data class Success(val hikeTrack: HikeTrackResponse) : MyState()
    }

    val state: Flow<MyState>
}

And here is the UT failing:

...
private val coroutineDispatcher = TestCoroutineDispatcher()
...
@Test
    fun `Given ... `() = coroutineDispatcher.runBlockingTest {

    When calling myRepository.requestStuff(myID) `it returns` flowOf(stuffResponse)

    val myUseCase = MyUseCase(
        myRepository,
        myID,
        coroutineDispatcher,
    )

    myUseCase.state.take(2).toList() shouldBeEqualTo listOf(
        MyState.Loading,
        MyState.Success(stuffResponse),
    )
  }
}

From the documentation it says

If the test needs to check multiple values, calling toList() causes the flow to wait for the source to emit all its values and then returns those values as a list. Note that this works only for finite data streams

. For what I understand, the UT fails because the flow I'm returning to my repository is an infinite streams. But shouldn't

flowOf(value)

create a finite stream ? Thanks !

Can multiple different threads 'join' the same Job? I have a library which expects blocking calls to some 'authenticate' method. This method will then be called from multiple threads and invoke a 'refreshToken' suspend function (which has to be invoked in runBlocking because 'authenticate' is expected to be blocking) but if some of these calls are called simultaneously I only want the 'refreshToken' to be called once for all of of them. My idea is to create a Job when the first one hits and them all the other could join this Job if it still exists. Is it a good approach or is there a better way?

Is

Mutex

suitable to be used at business logic level for example in

sendChatMessage()

to keep them sequential?

Hy folks, I'm using

receiveAsFlow

to transform a Channel in Flow. After cancel the CoroutineScope where I'm collecting the Channel values it seems not have closed the Channel. Can be codes like below create leaks without call

close()

? Should I add a call to

invokeOnCompletion

to make sure that close my Channel as:

invokeOnCompletion {
            channel.close()
        }