Hi, is it possible to cancel the scope without canceling parent scope? I need to cancel all jobs in scope, the scope itself but not parent scope. I am trying to avoid keeping reference to each job by myself and cancel all of them.

coroutineScope {
    val job1 = launch { foo() }
    val job2 = launch { bar() }
    delay(1000)
    job1.cancel()
    job2.cancel()
}

Hi how can you check if a Flow<V> has already been collected?

When working with coroutines should every call that is fired from ViewModel layer run in another dispatcher, e.g.

Dispatchers.Default

. I'm asking because in my vms I have suspending fucntions as well as non-suspending functions (most of the time not expensive operations) and I'm wondering if there is a thumb of rule when to switch from

main

to

default

or should we always switch to default when we working outside the UI scope not matter how expensive the operation is?

maybe it fits better here: What’s the state of Coroutine debugger support in Android Studio? https://kotlinlang.slack.com/archives/C0B8M7BUY/p1631007401262300

Is there a flow operator that does "check condition, if true close flow otherwise keep flowing while the emitted value for this condition is not emitted downstream"? I would like to check if the first item is of specific type, if yes newer values are emitted downstream if no close the flow. I have played around with

withIndex

,

takeWhile

and

takeIf

but it doesn't let me close the flow. I'm not sure but does

takeWhile

close the flow if false is returned? Maybe a combination of multiple operators will do it. Any ideas?

Are there any specific pitfalls when collecting a flow within a flow? I have following code:

suspend fun readPacketFlow(): Flow<Packet> = flow {
    api.readByteFlow().collect { bytes ->
        // parse the bytes and build packets from it
        ...
        val packet: Packet = someOperation()
        emit(packet)
    }

I'm observing strange behavior. On the caller site I'm collecting the flow like:

fun requestPacket() {
	coroutineScope {
		launch {
			dataSource.readPacketFlow()
				.onStart { Log.d("PacketFlow", "starting packet flow.") } // logged
				.onCompletion { Log.d("PacketFlow", "completing packet flow.") } // logged
				.collect {

					val packet = it as ResponsePacket

					Log.i("Received in Flow", "bytes: [${packet.bytes.toHexString()}]") // not logged
				}

			Log.i("Request", "after readPacketFlow.") // not logged
		}
    }
	Log.i("Request", "after scope.") // not logged
}

None of the logs (except the first two) are logged. I don't get what's going on here. Any ideas?

return webservice.getAllFacts()
    .also {
    coroutineScope {
        launch {
            catFactDao.insertAllFacts(it)
        }
    }
}

Code compiles, but is it a bad idea to use scope function on the return value (does it block even with the coroutine?)

Hello guys just need some clarification I have this

unit test

I was testing

hot flow / MutableSharedFlow

inside my

runBlockTest

I need to collect the flows then cancel the job,

launch

return

Job

and

coroutineScope

return a

Unit

and both of them can call

suspending function

If I use both

launch

it will work I get the desired output which is 10, but if I used

coroutineScope

to collect the flows and cancel the using

launch

I get

Job was cancelled

or instead of launch using both

coroutineScope

I get the same error. Why? Also if I use move the delay to first

launch

collection, I get the same prompt error of

Job was cancelled

, I thought I was running parallel execution e.g: like

launch1

after completed

launch2

then cancel the entire

TestCoroutineScope

.

In a non-

suspend

function (called by a coroutine) can I somehow access the

coroutineContext

? To be more specific, I'd like to access the

CoroutineName

for debugging reasons if specified.

Did kotlin allows suspend functions to be called from non-suspend functions or without coroutine scope in an older version? The code snippet below shouldn't compile because

getUser

function is not suspend, but it is calling

refreshUser

function. The code is from here https://developer.android.com/jetpack/guide#persist-data

fun getUser(userId: String): Flow<User> {

refreshUser(userId)

// Returns a Flow object directly from the database.

return userDao.load(userId)

}

private suspend fun refreshUser(userId: String) {

// Check if user data was fetched recently.

val userExists = userDao.hasUser(FRESH_TIMEOUT)

if (!userExists) {

// Refreshes the data.

val response = webservice.getUser(userId)

// Check for errors here.

`// Updates the database. Since

userDao.load()

returns an object of` `//

Flow<User>

, a new

User

object is emitted every time there's a` `// change in the

User

table.`

userDao.save(response.body()!!)

}

}

Hi all, coroutines & lifecycle question: If I'm launching a coroutine from an activity / presenter / view model and that coroutine is essentially going to launch a new activity flow and return a result once that flow is complete, what scope should I be launching the coroutine from? I've considered application level scope or creating a scope myself. Ideally, I'd like it to be lifecycle aware, but since I'm actually launching a new activity, I'm thinking making it lifecycle aware to a view model / activity will cancel the coroutine before I get a result back... Use case: start a login flow from different places in the app. Context: I'm trying to hook coroutines into existing legacy code that used to have rxjava + MVP architecture. There are several presenters / other code between me getting the second activity flow launch, otherwise, I would have used the onActivityResult or registerForActivityResult. The legacy code was doing this with RxJava disposables, and actually clearing the disposable with activity lifecycle, but I'm surprised that worked since the onStop() for the original activity should have been called when the new activity flow happened.

[SOLVED] I have a

channelFlow

that stops sending to collector after some time and I can't figure out why:

override fun readByteArrayStream(): Flow<ByteArray> = channelFlow {
        try {
            requireNotNull(bluetoothSocket) { "Bluetooth socket is null. Connection has been closed." }

            val inputStream = bluetoothSocket!!.inputStream

            val buffer = ByteArray(1024)

            while (isActive) {
                val numBytes = inputStream.read(buffer)
                val readBytes = buffer.copyOf(numBytes)
                Log.d("Reading", "$readBytes")
                send(readBytes)
            }
        } catch (e: Exception) {
            Log.e("Error", "error: ${e.message}")
        }
    }.flowOn(<http://Dispatchers.IO|Dispatchers.IO>)

It's still logging so the flow didn't stop. Any ideas?

Hi all, how to safely close resources (http connections e.t.c.) while using coroutines? As far as my understanding goes, Dispatchers such as IO and Default are using demon threads which will be immediately shutdown when VM halts. So we can't be 100% shure if resource was closed in time or not.

I've a very simple Flow, it basically comes down to this

val sourceA = MutableStateFlow<A>(null)
val sourceB = MutableStateFlow<B>(null)

val dataFromA = sourceA.map { it.x }.distinctUntilChanged()
val dataFromB = sourceB.map { it.y }.distinctUntilChanged()

val updateProcedure =
  dataFromA
    .combine(dataFromB) { x, y -> x to y }
    .distinctUntilChanged()
    .onEach { (x, y) ->
        otherMutableStateFlow.update { /*...*/ }
    }
    .launcIn(viewModelScope)

now if I input changing data in sourceA it is completely ignored after the 1st one: my

onEach

run only once. if I remove the combine and just get data from A + hardcode y to something it gets called every time. I've used combine SO MANY times and this is the first time something like this is happening to me. Does anyone have ANY idea of what could be going on here?

If I have a

suspend fun doWork()

and want to also wrap it in "listener-form" like

fun doWork(onWorkDone: () -> Unit) { 
  someScope.launch { doWork(); onWorkDone() }
}

I fear this might have some thread safety issues. What would be the "correct" way to do something like this? Why I want this: having a public API, where coroutines are an implementation detail. So I don't want to expose

suspend

functions or

CoroutineScope

anywhere.

A little follow up question. If some API exposes a suspend function, but the implementation detail is that this function will do a lot of IO-bound work, is it OK to wrap its entire body in

withContext

?

suspend fun callMe() {
  withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
    doAllWorkHere()
  }
}

If

withContext

is not used then I risk that the caller will use a context which is bad for this particular workload. Or is there some other well established pattern or design consideration for such cases?

hi all, how can i convert

Flow<List<T>>

to

Flow<T>

anyone know any networking libs that use coroutines and have support for udp? ktor is the only one I'm aware of, and it doesn't do UDP (I don't think).

Hi everyone! I am facing a strange error while I run the

nativeTest

task in a multi-platform project (using coroutines

1.5.2-native-mt

). I use

runBlocking

around each of my tests. This works correctly when I run each test independently one by one. But when I run the whole test suite, at the start of the second test, I get

Uncaught Kotlin exception: kotlin.IllegalStateException: Cannot execute task because event loop was shut down

(full stack-trace in thread). Did anyone ever faced this?

Is there a way, to have unique singleton coroutines? Background: If someone requests a message, it get's decrypted in a new corountine. But when someone requests the same message again, it should not start a secound coroutine to decrypt it, because there is already a coroutine, which decrypts it.

Hello everyone, I recently bumped into a weird runtime error when my Android app called a suspend function to perform an operation asynchronously. As I could not figure out the cause so far. I post the code snippet below and I am hoping to discuss it further here. Any comments are welcome. Thanks in advance.

typealias AsyncCallback = (Result?, Throwable?) -> Unit
private val listeners = mutableListOf<Command>()

@ExperimentalUnsignedTypes
fun interface CommandInterface {
    fun performAsync(
        arg1: UByte,
        arg2: UByte?,
        arg3: ByteArray,
        callback: AsyncCallback
    )
}

@ExperimentalUnsignedTypes
data class Command(
    val arg1: UByte,
    val arg2: UByte? = null,
    val arg3: AsyncCallback
)


private val executor = CommandInterface { arg1, arg2, arg3, callback ->
        listeners.add(
            Command(
                arg1,
                arg2,
                callback = callback
            )
        )
        // This method returns [Result] and resume [continuation] 
        action(arg3)
    }

suspend fun perform(
        arg1: UByte,
        arg2: UByte?,
        arg3: ByteArray
    ): Result? =
        try {
                suspendCoroutineWithTimeout(10000) { continuation ->

                    try {
                        continuation.invokeOnCancellation {
                            cancelRequest(arg1, arg2)
                        }

                        //Log.d(TAG,"arg1 type : ${arg1 is UByte}")
                        executor.performAsync(
                            arg1,
                            arg2, 
                            arg3 = arg3
                        ) { result, exception ->
                            when {
                                response == null ->
                                    continuation.resumeWithException(Exception("Null point exception"))
                                exception != null ->
                                    continuation.resumeWithException(exception)
                                else -> {
                                    continuation.resume(result)
                                }
                            }
                        }
                    } catch (e: Throwable) {
                        e.message
                        continuation.cancel()
                    }
                }
            } catch (e: TimeoutCancellationException) {
                cancelRequest(arg1,arg2)
                null
            }

When the app called perform() function, it got an error in a coroutine:abstract method “void CommandInterface.performAsync-vmjznIo(byte, kotlin.UByte, byte[], kotlin.jvm.functions.Function2)” It seemed the Kotlin compiler recognised the first argument arg1 as byte, but the interface defined it as UByte. The Kotlin plugin I am using is 1.5.30, AGP version is 7.0.2

Best way to launch a coroutine in a suspend function? See thread.

I just started using Flows, and in migrating my LiveData I often had the use case “generally speaking the LiveData is updated internally, but sometimes I want to manually trigger an update from outside”. An example would be some state that is read from a data source periodically, but on hitting “refresh” I want the newest value immediately. In another case, my data source (some kind of observer provided by Android) is not 100% reliable, so on hitting “refresh”, I would get the same information from another source to overwrite the current value. So in the world of Flow, I might have a

callbackFlow + stateIn

or a flow created with

combine + stateIn

. Now this is a

StateFlow

which has no

emit

that is callable from “outside”. Is there a way to achieve this, basically a

.toMutableStateFlow()

?

can Kotlin actors be orchestrated in some kind of a hierarchy, possibly with restarts, stopping all descendent actors when a parent stops, etc..?

Hi All, sorry if this is a super basic question, I’m starting to learn reactive programming and it is overwhelming! I am calling a library that returns a

Flux

, but I want to have instead a

Flow

. Is this something people do and if so, how do I go about doing it? Or am I thinking of this the wrong way? I have google searched, but surprisingly I can’t find any examples to learn from. Thanks so much!

i notice that it happens when i am simply navigating to another fragment using android navigation sdk/api

Hey, sorry for this (daily) question, but I don't find the root case: I want to use Swift Combine and Kotlin Coroutines in an iOS and Android app. I use `Flow`s in my view models and repos, but create the init data flow with device local daos, on iOS with a Combine publisher. I always use

Dispatcher.Main

when converting the Publisher to Flow and vice versa, (maybe even too much), but I am always getting

Flow invariant is violated:
		Emission from another coroutine is detected.
		Child of null, expected child of DispatchedCoroutine{Completed}@806156c8.
		FlowCollector is not thread-safe and concurrent emissions are prohibited.
		To mitigate this restriction please use 'channelFlow' builder instead of 'flow'

So I have a two mutable shared flows, Flow A and Flow B. When Flow A triggers I want to use the latest value of Flow B, without having to react to when Flow B updates . I am also using the flow at other places so I don't want to cancel it. I tried combine but then I react to Flow B's updates. I then tried

zip

, but it cancels Flow B. I can get it to work with

replayCache()?.latestOrNull()

but that doesn't look good at all...

Is there ever a reason not to use

MutableStateFlow.update { newValue }

vs.

.emit(newValue)

? Basically, just to be safe regarding concurrent updates?