Are there any plans for

ReceiveChannel.isClosedForReceive

to graduate from being "experimental"?

Hi I am not sure what I am doing wrong here. In my app I have an authenticationRepo that has two suspend methods. I am aware that using global scope is not the right way to start a coroutine but this is for testing purposes. The problem I am seeing is that after collecting the flow from the first call to authenticationRepo, I don’t see to be able to get the second call to authenticationRepo to fire (as can be seen from the recording). I am not sure if i am using flows incorrectly but I am unable to understand why is the second line of code not hitting

Is there a way to block on a scope to finish it’s jobs while not causing a deadlock, if they’re on the same thread? I have cleanup code in the finally block that needs to run on the same thread before onStop completes.

I have ViewModel, those are driven by Scope of

Dispatcher.Main

ViewModel has state, and therefore State (

MutableStateFlow<State>

) should only be set on main thread -- to avoid synchronization Is there a way I can assert Scope is on main thread without an android reference? (

Looper.myLooper() != Looper.getMainLooper()

) (To keep my viewmodels KMP ready) Or, should I just not care and read-write (reduce) to StateFlow within a synchronization block?

Is there a way to check a

Channel

size (without consuming the contents)? I'm currently counting

send()

and

receive()

calls manually.

I want to cancel all old jobs in a CoroutineScope and start a new job in the same CoroutineScope. What’s the best way to do that?

I have a question about the interpretation of the following sentences. Does this mean it's a good practice to use withContext{} in all functions, even if there is no blocking process in the Repository?

A good practice is to use 

withContext()

 to make sure every function is main-safe, which means that you can call the function from the main thread.

https://developer.android.com/kotlin/coroutines/coroutines-adv#main-safety

I have a stream of events coming into a

ReceiveChanel

every 10ms, representing the state of some buttons on an external controller. When I call a method, I need to wait until I see 10 no button press events, then 10 button press events of a given type in a row, followed by 10 no button press events, then call a supplied completion handler. So far I’ve got as far as this:

fun readButtonPress(press: ButtonPress, completion: (ButtonPress) -> (Unit)) {
  launch {
     eventReceiveChannel
        .consumeAsFlow()
        // wait for:
        // at least 10x no press events
        // 10x or more press events
        // at least 10x no press events
        .consume {
            completion(buttonPress)
        }
     }
}

My questions are therefore: 1. am I on the right track? I want to contain coroutine use to this class, as it needs to be called from a callback-based API 2. what is the correct way to track state in this scenario (i.e. number of events received in a row of a given type)? 3. what are the correct flow methods to use to achieve this goal? I think I’ve just about got my head around coroutines conceptuals and beginning to get my head around Flow, but my brain is blowing a fuse trying to figure this out.

It looks like try catch around a background task will prevent interruption of a coroutine:

val job1 = launch {
    try {
        withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
            SystemClock.sleep(1000)
            throw IOException("Wurst")
        }
    } catch (err: Throwable) {
        Log.d("Test", "code executed")
    }
}
launch {
    delay(500)
    job1.cancel()
}

In this code the Log.d line will be executed. Which is bad when e.g.: i want to show a error message after a network call failed but the fragment coroutine scope is already cancelled. Is there any ellegant solution to avoid executing the catch when corotuine scope is already cancelled?

I would like to write some tests for an operator which filters button press events (see thread from yesterday). To get some inspiration for the right approach, I’ve been looking at some of the existing Kotlin Coroutine tests e.g.

kotlinx.coroutines/kotlinx-coroutines-core/common/test/TestBase.common.kt

. Is it possible to re-use

TestBase()

my own tests? I’ve tried

import kotlin.test.*

however this package is not found, despite having the following in my Android app `build.gradle`:

implementation ‘org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.3’

I’ve also tried adding a

testImplementation

line with the same, but the above import still fails. Is re-using

TestBase()

etc. in my code not possible? Or is their a separate

testImplementation

entry I should add to my app

build.gradle

?

Hi everyone, I have some confusion with two results, the first is Kotlin JVM and the second is Android Application. Anyone can help me to explain why it a difference in results with the same code logic?

how do I cancel a flow after the first time

collect { }

completes? I’d like to do something like this, but the job val is not visible:

val eventJob = launch {
   myChannel
      .consumeAsFlow()
      .filterAllButFinalEvent()
      .collect { event ->
          doSomething()
          eventJob.cancel() // doesn’t compile
      }
   }

Hi. I have something like

val sf1 = MutableStateFlow(listOf(1))
val sf2 = MutableStateFlow(listOf(-1))

sf1.value = listOf(1, 2)
sf2.value = listOf(-1, -2)
sf1.value = listOf(1, 2, 3)

I combine them like

sf1.combine(sf2) { a, b -> a + b }
    .stateIn(scope, SharingStarted.Eagerly, initialValue = listOf())
    .onEach { println(it) }
    .launchIn(scope)

Then I don't have [1, -1] list in combined flow. I changed initialValue to sf1.value + sf2.value and now it works but I think that I could miss something which simplifies this. Is there a better way?

I find that I frequently need a Job+SendChannel that sends to that job, where the Job acts as a consumer on the channel (internally it has the corresponding ReceiveChannel) and the job exits once the channel is closed for receive. I've built some small helper classes&functions to help with this, but I'm wondering if there's something already in kotlinx.coroutines I missed, or if there's some reason I shouldn't be doing things this way.

How would you call a function if and only if a coroutine is suspended for more than X ms?

log.verbose("awaiting for ${request.uri}")
hasTokenStates
    .first { hasToken -> hasToken } // suspending
log.verbose("awaiting complete for ${request.uri}")

I want to only log

"awaiting for ${request.uri}"

if the

first()

suspends for more than `1_000`ms

Hi, Can some explain to me why:

runBlocking {
        launch {
            println("a1")
            runBlocking { delay(100) }
            println("b1")
        }

        launch {
            println("a2")
            runBlocking { delay(300) }
            println("b2")
        }
    }

displays a1 a2 b2 b1 - instead of a1 a2 b1 b2

(I my real code, the runBlocking are processed from a method.invoke to "wait", that why I can' t use suspend functions there) - How can I avoid this unexpected behvior?

The differences between

1.4.3

and

1.4.3-native-mt

are only in the native artifacts, in other words, jvm and android artifacts are the same?

I’m trying to remove channel.consumeEach from bluetooth communication in my project. What is current best replacement for it? Should I use consumeAsFlow() and then just collect it?

When I try to use the Coroutines tab in IntelliJ IDEA's debugger it says "An error occurred on fetching information". How do I get it working?

Assume I have this:

runBlocking {
 ...
 collection.asFlow().flowOn(<http://Dispatchers.IO|Dispatchers.IO>).fold(0) { acc, next ->
  acc + calculate(next)
 }
}

Does it matter if the

calculate

function is suspending or not in terms of performance?

How does one debug deadlocks in coroutines? What I'd like to be able to do is run the code until it deadlocks, and see a snapshot of where each coroutine is "stuck". The coroutines tab in the debugger doesn't work, so I've just been adding logging around every suspend point I think could be responsible. This "works", but is incredibly tedious and slow, and also error-prone because it'll fail if I don't think of the guilty suspend point when adding my logging. Is there a better way?

I am working on a KMM project where I have the following class

@ExperimentalTime
internal class AuthenticationRepositoryImpl(
    private val authenticationApi: AuthenticationApi,
    private val authenticationQueries: AuthenticationQueries
) : AuthenticationRepository {

    override suspend fun authenticateUser() {
        authenticationQueries.clearAuthenticationData()
        authenticationApi.authenticateUser().toAuthentication().apply {
            authenticationQueries.setAuthenticationData(accessToken, expiresBy)
        }
    }

    @ExperimentalTime
    override suspend fun getAuthenticationData(): Flow<Authentication?> {
        val timeNow = Clock.System.now().epochSeconds
        return authenticationQueries.getAuthenticationData(timeNow).asFlow().mapToOneOrNull()
    }
}

and I am calling the functions from my android app like so:

fun checkAuthentication() {
    viewModelScope.launch {
        authenticationRepository.getAuthenticationData().onEach {
            if (it != null) {
                Timber.d("Valid Authentication Found")
                isAuthenticationValid.value = true
            } else {
                Timber.d("Valid Authentication Not Found - Requesting Authentication")
                authenticationRepository.authenticateUser()
            }
        }.collect()
    }
}

In the following code if the authentication data is not present in the SqlDelight DB, I get a null in the onEach of the flow and then I trigger a fetch and save to DB. However, the save to DB does not re-trigger a flow update. Is there something I am doing wrong here?

I just learned a lot about run blocking from this thread: https://kotlinlang.slack.com/archives/C1CFAFJSK/p1615458029368600 Just for reference: 1. two

runBlocking

, launched on the same single threaded dispatcher do not block each other. They share an event queue. 2. former

runBlocking

will only return after later

runBLocking

completes (1) comes as a positive surprise (2) is totaly unexpected and I’d consider it a bug I’ll follow up in a thread with the demo code

What is the best practice to work with

org.reactivestreams.Publisher

in coroutine?

Hello) I have some concern

private fun loadData() = with(presenterScope) {
    listOf(
        async {
             metadata = transferOtherBank.getMetadata()
        },
        async {
            sources = transferOtherBank.getSourceList()
        },
        async {
            favorites = transferOtherBank.getFavorites()
        },
        async {
            fee = transferOtherBank.getTransactionFee()
        },
        async {
            banks = transferOtherBank.getBanks()
        }
    )
}

where presenterScope is Coroutine scope provided by moxy2 library (I can attach implementation if it’s necessary)

presenterScope.launch {
    loadData().awaitAll()
    setName(favorites, banks)

    Timber.d("ASSSAS after await")

}

Once some of async fails (getFavorites for example) setName(…) function is never called. How can I ensure calling of setName(…) despite any error in any async? Thanks!

Is setting the value on a MutableStateFlow thread safe?

Is this an okay approach to download images from an URL while saving them to local storage. It should happen in parallel and I need to await for all three images to return their URI before proceeding. I kinda dislike the

if (edgesUri.getCompleted() == null

null checks and the

labelUri.getCompleted()!!.path!!

. Any way to improve this?

suspend fun saveImageToDB(networkImageModel: CBImageNetworkModel): Result<Long> {
        return withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
            try {       
                val edgesUri = this.async {
                    val urlEdges = URL(networkImageModel.edgesImageUrl)
                    return@async saveBitmapToAppStorage(urlEdges.openStream(), ImageType.EDGES)
                }
                val finalUri = this.async {
                    val urlFinal = URL(networkImageModel.finalImageUrl) 
                    return@async saveBitmapToAppStorage(urlFinal.openStream(), ImageType.FINAL)

                }
                val labelUri = this.async {
                    val urlLabels = URL(networkImageModel.labelsImageUrl)
                    return@async saveBitmapToAppStorage(urlLabels.openStream(), ImageType.LABELS)
                }

                awaitAll(edgesUri, finalUri, labelUri)

                if (edgesUri.getCompleted() == null || finalUri.getCompleted() == null || labelUri.getCompleted() == null) {
                     Result.failure<Long>(Exception("An image couldn't be saved"))
                }

                val image = CBImageDataModel(
                    labelsImageUri = labelUri.getCompleted()!!.path!!,
                    finalImageUri = finalUri.getCompleted()!!.path!!,
                    edgesImageUri = edgesUri.getCompleted()!!.path!!,
                )
                Result.success(db.imageDao().insertImage(image))
            } catch (e: Exception) {
                Timber.e(e)
                Result.failure(e)
            }
        }

when I wrap an object that holds resources and requires a callback through a

suspendCancellableCoroutine

, do I need to cleanup the resources inside the error callback?

suspend fun isSessionValid(): Boolean = suspendCancellableCoroutine { continuation ->
    lateinit var auth: Auth

    val callback = object : AuthHandler {
        override fun onSuccess(session: AuthUserSession) {
            continuation.resume(session.isValid)
            auth.release()
        }

        override fun onFailure(e: Exception) {
            continuation.resumeWithException(e)
        }
    }

    auth = builder.setAuthHandler(callback).build()
    auth.getSession()
    continuation.invokeOnCancellation { auth.release() }
}