when calling a 

suspend

 function within 

runBlocking

 , the following code should not be executed until the function completes, correct? I am seeing some flakiness in a test that doesn't make sense to me.

launch {
    val result = getSomething().apply {
        doSomethingSuspending()
    }
    // code at this point never runs
}

Could anyone explain why this happens?

Hi, I'm using a

MutableSharedFlow

with a collect lambda.

val responseFlow = MutableSharedFlow<Message>(replay = 0)
...
responseFlow.collect {
...
}

When I collect a message, I do a test and for some value , I would like to leave the collect function and return a value. I'm not sure how to do that. Any idea?

Hi, I'm calling a proxy from a suspend function

suspend fun test() {
  o.method()
}

here,

o

is a proxy obtained by

Proxy.newProxyInstance(..., proxyHandler)

. So this will call the

invoke

method of

proxyHandler

. My question is: how can I propagate the suspend up to

proxyHandler

and use a suspend function from its

handle

method (without having to start a runBlocking of course) ?

If I call

delay

or any other blocking operation inside the

collect

block of a Flow, the next value will not be collected until that delay is over right? So I can't get race conditions between successive emissions of the Flow?

Is there a way to do a cartesian product of two flows? Something similar to

join

in RxJava? I have a flow of files and a flow of processors and I need to run all the processors through all the files. I'm looking for something like:

filesFlow.join(processorsFlow) { file, processor -> processor.invoke(file) }

(

proccessor.invoke

returns the result of the processor so I get new flow of type result)

Hi, I do not understand the following behavior:

fun main() {
    val flow = MutableSharedFlow<Int>(replay = 0)

    runBlocking {
        launch {
            println( flow.first { it >= 90 } )
        }

        launch(<http://Dispatchers.IO|Dispatchers.IO>) {
            repeat(1000) {
                flow.emit(Random.nextInt(0, 100))
            }
        }
    }
}

This code does not print anything. It prints when I remove

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

at the second launch or (sometimes) if I add

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

at the first launch... any explanation?

What should I use instead of

MutableStateFlow

as my

flatMapLatest

trigger when I need to allow emitting the same value in succession? Channel?

I'm sorting a large list (10.000+) items with use of the

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

from an Android room database. What pattern is recommended to use?

private val supervisor = SupervisorJob()
private val coroutineScope = CoroutineScope(coroutineContext + supervisor)
...
fun onEvent(event: MyEvent) {
    coroutineScope.launch {
        val data = findHeavyWorkToExecute()?.invoke()
            ?: throw Exception(currentState, event)

        someMoreWork(data)
     }
}

I’m want

someMoreWork()

to be executed only if the lambda returned by

findHeavyWorkToExecute()

is not null and its execution does not fail. But when I test this

someMoreWork()

is called even though the

throw

statement occurs. What am I suppose to change there?

I wasn't sure wether to post here or in #android, but the issue I'm having is with the Kotlin coroutines library. Just need some help getting my head around what I'm doing wrong!

# MainActivity.kt

override fun onStart() {
        super.onStart()
        var applicationID = runBlocking { Application.registerDevice() }

        Log.d("AppID on-start", applicationID)
    }

# Application.kt

object Application {
    suspend fun registerDevice(): String {
        return getTokenAsync().await()
    }

    private suspend fun getTokenAsync(): kotlinx.coroutines.Deferred<String> = coroutineScope {
        async {
            return@async FirebaseInstallations.getInstance().id.result.toString()
        }
    }
}

E/AndroidRuntime: FATAL EXCEPTION: main
    Process: com.example.application, PID: 20890
    java.lang.IllegalStateException: Task is not yet complete

If anyone could see where I'm going wrong and help point me in the right direction on what I can do to resolve this I would really appreciate it 🙂 Thanks in advance!

When I catch a

CancelledException

from a Coroutine, can I find out where the cancel originated from? Some of my coroutines are spuriously cancelled and I'm trying to find out why 🤔

while using flow.first() operator, it throws exception when flow is empty as docs say, but I am fetching list from database using a flow and checking if that flow.first() , and if the list is empty , does it mean flow is empty ?

Given this class

class Client {

    private suspend fun connectInternal(url: String) { /* does connection stuff */ }
    private suspend fun disconnectInternal() { /* does disconnection stuff */ }

    suspend fun connect(url: String) {
        TODO("Should call connectInternal")
    }

    suspend fun disconnect() {
        TODO("Should call disconnectInternal")
    }
}

I'd like to implement

connect

and

disconnect

so that they call the internal APIs in a thread-safe fashion. Client can only be connected to one url at a time so, for example: • disconnect should cancel any in-progress connection • when connect is called from multiple threads, only the last succeeds, the others will be canceled I can implement this in a thousand ways, but I always end up using some Java threading primitive at least in 1 place (like a synchronized block or a lock). Is there a way to achieve the same only with coroutine primitives? One simple option is do everything in like a

newSingleThreadContext

but that's more of a bad workaround

Is this ok to do?

val pool = Executors.newFixedThreadPool(N)
val dispatcher1 = pool.asCoroutineDispatcher()
val dispatcher2 = pool.asCoroutineDispatcher()

Has anyone experienced this? (Kotlin: 1.4.21, Coroutines: 1.4.2, AS: 4.2 Canary 16) I’ve already wasted hours on this. The interesting thing is that the code runs ok 🤔

I have like

2 flows

and I want to observe their changes with

flatMapLatest

, if either of them emit new value, I want to trigger some action, but how can I observe on

2 flows

, should I combine them ? how can I do so

Do I need to explictly

close

a channel if I only want to

receive

a single value from it? Something like this inside some suspend function

val whatIWant = myChannel.receive()
myChannel.close()
return whatIWant

// VS

return mychannel.receive()

is there an operator to plumb the output of one flow as the input to another (sharedflow)

Why is it when I try to use

MaterialTheme.colors.*

I get an error of "@Composable invocations can only happen from the context of...", but if I just use a string there instead then it works fine, which shows that my @Composable is actually setup correctly. Code snipipets in thread.

Trying to cancel flow emission when a certain scope (not the collector scope) dies. I thought

flowOn

would help but the scope there can't have a Job, which makes sense in the end. So I'm thinking of

flow.onEach { myScope.ensureActive() }

Is this the only way to do it? Are there any concerns other than performance?

This message was deleted.

Is there a specific method I should use on a

Channel

to cast it to a type allowing me only to send/offer? The same way there is

receiveAsFlow

to only be able to collect it, how can I restrict my code to only sending?

fun main(): Unit = runBlocking {
    println(1)
    launch {
        println(2)
        launch {
            println(3)
        }
    }
    launch {
        println(4)
    }
    println(5)
}

why does the above code print 1 5 2 4 3 instead of 1 2 3 4 5 everything is running in main thread?

How to use

withTimeout

inside

suspendCoroutine

? something like:

suspendCoroutine{ cont->
  withTimeout(1000){
    // DO something
    cont.resumeWithException(Exception("Timeout"))
  }
}

How to catch all the exceptions possibly thrown by multiple jobs launched in same scope (SupevisorScope)? Currently I’m doing something like this but no exceptions are getting caught

Hi, is there a way to pass @launch to another function? I have this piece

viewModelScope.launch { 
    validateEmail(user)?.let {
        _errorLiveData.postValue(it)
        stopLoading()
        return@launch
    }
}

And I want it to look something like

evaluate(validateEmail(user), launch)

with

fun evaluate(result: String?, job: (
    context: CoroutineContext,
    start: CoroutineStart,
    block: suspend CoroutineScope.() -> Unit
) -> Job){
    result?.let {
        _errorLiveData.postValue(it)
        stopLoading()
        return@job
    }
}

I get job is unresolved, so what am I doing wrong and how to make it right?

Would this be a good pattern for a function that both fetches one single data element, and also a series of other data elements simultaneously?

fun fetchData (scope : CoroutineScope) : Pair<Deferred<Int>, Channel<Float>> {
    // Start first operation
    val r1 : Deferred<Int> = scope.async {
        delay (200) // Pretend to do some work
        42  // Return a result
    }

    val r2 = Channel<Float> ()

    // Start second operation
    scope.launch {
        repeat (10) {
            delay (it * 100L) // Pretend to do some work
            r2.send(it.toFloat()) // Return one result
        }
        // Second operation done
        r2.close()
    }

    return Pair(r1, r2)
}


fun main () {
    val results = fetchData(GlobalScope)

    runBlocking {
        launch {
            val res1 = results.first.await()
            println("Result 1 : $res1")
        }
        launch {
            for (elem in results.second) {
                println ("Result 2 elem: $elem")
            }
        }
    }
}

Without coroutines, I'd use a callback interface with functions

onResult1(x:Int)

and

onResult2(x : Float)

and call the first one once and the second one multiple times.

Is there any operator for

Flow<Flow<T>>

like

flatten*

which can cancel child flow collection when a new parent value is emitted? Say you have

fun parentFlow(): Flow<String> = ...
fun childFlow(id: String): Flow<Result> = ... // id comes from parentFlow

My goal is to have a single

Flow<Result>

, taking the id from parent flow. One way to do this is:

val merged: Flow<Flow<Result>> = parentFlow().map { childFlow(it) }
val flat: Flow<Result> = merged.flattenConcat()

However in my case the parent id represents a kind of state, so when a new id is emitted, the flow should stop collecting the previous

childFlow

. I'm struggling to find an elegant way to do this. I could collect the child inside the parent body (inside

flow { }

for example), but then not sure how to cancel the previous one