I thought I understood kotlin coroutines 🤦‍♂️

📣 📣 📣 kotlinx.coroutines 1.7.0-Beta is here! This is a Beta of the major release that contains multiple significant reworks and improvements: • Much more efficient new

Channel

,

Mutex

and

select

implementations • JPMS support • Merge of

kotlinx-coroutines-core

and

kotlinx-coroutines-jdk8

• Partial stabilization and notable improvements of

kotlinx-coroutines-test

• More K/N support: all official targets, better dispatchers,

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

And many more! Full changelog: https://github.com/Kotlin/kotlinx.coroutines/releases/tag/1.7.0-Beta

Hello all, I am new at kotlin, Is this the correct way that the flow is executed?

suspend fun Logout(token: String) {
    val userId = appService.currentUser!!.id       ///// this will execute 1st
    realm.write {                                  ///// this will execute 2nd
        var user = query<UserInfo>("_id = $0", userId).first().find()
        if (user != null) {
            user = findLatest(user)!!.also {
                it.FCMToken.remove(token)
            }
            copyToRealm(user)
        }
    }
    withContext(Dispatchers.Default) {             ///// this will execute 3rd

        realm.syncSession.uploadAllLocalChanges()  ///// this will execute 4rd

        appService.currentUser?.logOut()           ///// this will execute 5th
    }
}

Are they executed in a linear way?

I’m trying to implement concurrent system using actor models. Not able to wrap my head around sending delayed message to actor. Attaching my snippet to implement the same. Is this correct?

This message was deleted.

I’m struggling to redesign this thing:

class Processor {
    private val condition1 = MutableStateFlow(false)
    private val condition2 = MutableStateFlow(false)
    private suspend fun process(): Processed { ... }
    init {
        someScope.launch {
            combine(condition1, condition2) { a, b -> a || b }
                .collectLatest { process() }
        }
    }
    fun setCondition1() { condition1.value = true }
    fun setCondition2() { condition2.value = true }
}

When the user changes one of the conditions, I would like to • know if this change did not trigger a

process()

(for example, throw if not) • return something like a

Deferred<Processed>

to be (optionally) awaited. Does anyone have any idea on how to achieve this?

I have the following code and I want to make sure the callback is unregistered. How can I trigger

awaitClose

from my unit test ? I tried to cancel

testScope

but that triggers a

ClosedReceiveChannelException: Channel was closed

since turbine gets also canceled.

private fun someApiCallbackToFlow(api: API) = callbackFlow {
    val callback = object : ApiCallback {
        override fun next(data: Int) {
            trySend("Success")
        }
    }
    api.register(callback)
    awaitClose {
        println("callback unregistered")
        api.unregister(callback)
    }
}


@OptIn(ExperimentalTime::class)
@Test
fun test() = testScope.runTest {
    val api = SomeApi()
    val flow = someApiCallbackToFlow(api)
    flow.test {
        assertEquals(1, api.callbacks.size)
        api.getData()
        assertEquals("Success", awaitItem())
        // how can I check there isn't any callback here?
    }
}

Hi guys, I am getting strange issue using

runTest

. I have a simple viewModel

class LoadingViewModel(
    private val ioDispatcher: CoroutineDispatcher,
) : ViewModel() {

    var xyz by mutableStateOf(false)
    fun checkXyz() {
        viewModelScope.launch {
            delay(1000L)
            xyz = true
        }
    }
}

more code in 🧵

Wondering what would be the idiomatic way to launch multiple coroutines in a

supervisorScope

while still capturing all exceptions (aggregated) that are thrown in them. Note that I do not want failure of a child to cancel other children (or the scope), hence supervisor scope, but I might want to “sum” (probably

addSuppressed

) all exceptions and throw it later after scope is done Right now I’m using

withContext

with a custom

CoroutineExceptionHandler

that puts all exceptions in a collection and later reduce the collection and throw, but it just feels boilerplate-y and wonder if I’m missing something

Am I wrong in assuming that there haven’t been many books on coroutines yet because some of the APIs have been experimental (e.g. @FlowPreview and @ExperimentalCoroutinesApi)? If correct, when will they be stable?

Hello! When updated to version 1.7.0-Beta got error on mergeDebugJavaResource while building Android apk. (2 files found with path 'META-INF/versions/9/previous-compilation-data.bin') There reason is that in both kotlinx-coroutines-android-1.7.0-Beta.jar and kotlinx-coroutines-core-jvm-1.7.0-Beta.jar there is existence of META-INF/versions/9/previous-compilation-data.bin. While at 1.6.4 version this file exists only in kotlinx-coroutines-core-jvm. kotlinx-coroutines-android depends on kotlinx-coroutines-core-jvm, so when add dependency to kotlinx-coroutines-android also got kotlinx-coroutines-core-jvm as transitive dependency. Is it a bug or there should be changes from library user's side?

Hi all! Pipeline channels is a commun pattern for concurrency... I'm trying to move some code from Channels to Flows... but not sure what's the best way to pipeline Flows without causing callback-hell issue as below:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow

fun main() = runBlocking {
    produceFlowNumbers()
        .collect { // First collect
            squareFlow(it)
                .collect { // Second collect inside First collect
                    println(it)
                }
        }
    println("Done!") // we are done
}

fun produceFlowNumbers() : Flow<Int> = flow {
    var x = 1
    while (x < 5) emit(x++) // infinite stream of integers starting from 1
}

fun squareFlow(number: Int): Flow<Int> = flow {
    emit(number*number)
}

I have the feeling that I'm missing something here. Any suggestion experts?

Hey ! I want to ask people around if

runCatching

is in use in any of your Kotlin (or Android) projects ? I feel my team is abusing it and I'm trying to understand if there's any "pros" about using it in any coroutine-based codebase. My team just adopted this convention and I'm at loss of words. (Examples in thread for anyone interested)

I'm attempting to create a small wrapper for SharedFlows to add some priority functionality and some better dependency injection usability. While doing this I'm observing some extremely odd behavior that I can't seem to figure out. Using this test class which mirrors my unit tests as much as possible you can see the behavior. Basically, I'm acquiring a lock to wait on the flow to be called in order to test the outcome, but the lock is never released because the subscriber is never called despite having clearly subscribed. The extreme weirdness comes in when I add certain specific

println

calls that reference variables in which case the problem disappears while that statement is present. The pasted test class will print the following when ran (and never complete):

Registering
Registered
Emitting
Emitting here
Subscribers
0
Done
Suspending

Note that is says there are zero subscribers and finished emitting even though it was registered. However, when uncommenting lines 59 or 63, it still says there are zero subscribers, but the object is emitted to my registered collector and the program finishes. I am relatively new to Flows and have avoided them mostly because of these kinds of things, but I would love to make them work since they are exactly what I need. I feel like I must be doing something wrong. I appreciate any help!

Hi ! I'm having an issue with coroutine in a Jetpack Compose for Desktop project. More in the thread 🧵 .

In a suspend function, is there a way to check if the calling coroutine is cancelled?

delay

is accessible and does check the cancellation status so there must be a way right?

Hello, can someone help me with this please? https://stackoverflow.com/questions/75779125/kotlin-compose-how-to-know-when-suspend-function-is-finished?noredirect=1#comment133681725_75779125

I would pay to someone that helps me refactor my login/logout methods

I'm trying to upgrade to Coroutines

1.7.0-Beta

, and it appears that

kotlinx.coroutines.future.future

has been moved - does anyone know what it's replacement is?

What is the recommended way to use MDCCOntext? For example, in a Ktor route like:

routing {
    get("/some-route") {
        MDC.put("foo", "bar") // Will it leak in successive requests for the same route?
        println(MDC.getCopyOfContextMap())
        delay(1)
        println(MDC.getCopyOfContextMap()) // Should I use `withContext(MDCContext()) {}` here?
    }
}

Should I do something like this?

routing {
    get("/some-route") {
        // Use MDCContext before modifying the MDC to ensure the data is restored
        withContext(MDCContext()) {
            MDC.put("foo", "bar")
            println(MDC.getCopyOfContextMap())

            // Use MDCContext before suspending the coroutine to ensure
            // the data is still available when resuming
            withContext(MDCContext()) {
                delay(1)
                println(MDC.getCopyOfContextMap())
            }
        }
    }
}

As you can see, I’m a bit lost and I’m not sure when I need to use

MDCContext

😬

Hi all, I'm trying to replicate some

CompletableFuture

code in Java to

Coroutines

in Kotlin, the interface that my Kotlin class implements has functions that are returning

CompletableFuture

Here is example of functions that I want to rewrite

private final Map<String, FooWrapper> fooRequests = new ConcurrentHashMap<>();

public CompletableFuture<FooResponse> retrieveFoo(FooRequest request) {
    return completedFuture(...)
        .thenAccept(client::callFoo)
        .thenCompose(ignore -> storeFooRequest(request))
        .whenComplete((response, error) -> log(request, error));
}

private CompletableFuture<FooResponse> storeFooRequest(FooRequest request) {
    var fooWrapper = new FooWrapper(request, new CompletableFuture<>());
    fooRequests.put(request.getId(), fooWrapper);
    return fooWrapper.getResponseFuture();
}

eventually there will be call to other functions that will call

complete

on a

CompletableFuture

stored in

fooRequests

and allowing completion of

retrieveFoo

here is my Kotlin code

private val fooRequests: ConcurrentHashMap<String, CompletableDeferred<FooResponse>> = ConcurrentHashMap()

override fun retrieveFoo(request: FooRequest): CompletableFuture<FooResponse> {
    return GlobalScope.future {
        retrieveQuoteAsync(request)
    }
}

private suspend fun retrieveFooAsync(request: FooRequest): Deferred<FooResponse> {
    return coroutineScope {
        storeFooRequest(request)//.await()
    }
}

private suspend fun storeFooRequest(request: FooRequest): Deferred<FooResponse> {
    val deferred: CompletableDeferred<FooResponse> = CompletableDeferred()
    fooRequests[request.id] = deferred
    return deferred
}

I'm trying to replace

CompletableFuture

with

CompletableDeferred

but I do not know how to get the value without calling

await()

, I do not want it to block Any suggestion how to make it work in similar way to Java?

I think I am misunderstanding the semantics of

transform

. I have a flow along the lines of`sharedFlow -> transformWhile -> map` , and I see the

map

run even when the downstream collector has completed. Is this expected? Example in the thread.

Hey guys, I have a question that I just posted on Stackoverflow It’s about the coroutine scopes and structured concurrency Could someone please answer it

Here is simpler one 😄

Am I correct that ‘coroutineScope’ function is the suspending equivalent of runBlocking? If so, wouldn’t it make more sense to have called it runSuspending?

Hey guys I’m trying to understand how

suspendCoroutine

works, I tried to implement an example, here is the GitHub Gist link Does this code make sense ? it works but I was wondering if it can be improved somehow Thanks in advance

I have the following test, which fails. It seems like the update done by one flow does not affect the other one. Is this a case where I should use a

Channel

to be able to communicate data between coroutines?

Is it possible to create a test that terminates when all the fire & forgot coroutines in the test have completed? 🧵