Trying to figure out how to test channels + flow but I keep running into a

IllegalStateException: This job has not completed yet

. I have a channel in my viewmodel which I expose as a flow ViewModel

private val _channel = BroadcastChannel<Boolean>(1)
val myFlow = _channel.asFlow()

fun sendToChannel() {
    viewModelScope.launch {
        _channel.send(true)
    }
}

And in my activity, I collect the flow

lifecycleScope.launch {
     myFlow.collect { println(it) }
}

My Test

@Test
fun `myTest`() = runBlockingTest {
    
    viewModel.myFlow.collect { myVal ->
        assertTrue(myVal)
    }

    viewModel.sendToChannel()
}

How do I ensure the assertion is called. Or what’s a better way of testing this?

Is there much overhead (perf, memory, apk size) to have both rxjava & coroutines in app, given dispatchers are shared from rx schedulers? Is it reasonable?

Question about Workers on Kotlin/Native \ MP on windows: (I figured the coroutines channel was still the best place to ask the following, even tho I can't do this with coroutines (yet?).) I've checked out the workers example here [ https://github.com/JetBrains/kotlin-native/blob/master/samples/workers/src/workersMain/kotlin/Workers.kt ] and got that to work, but I am now wondering if there is a way ( in Kotlin, without resorting to sharing a buffer via C or something like that ) to have a long running process ( e.g. another windows program launched via popen ) running in a worker which periodically sends data ( e.g. a string of the output of the program ) back to the main thread. From what I understand, workers allow you to do work in a separate "environment", and then "transfer back" the (immutable) data once the worker is done. Is there a canonical way to periodically "send data back" to the "main thread"?

How do CoroutineScope() class works? what happens if you create multiple scopes on single context? and what's better

CoroutineScope(context).async{}.await()

or

withContext(context) {}

?

Another quick question, what happens if you join two dispatchers? For example:

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

? (I know its not intended i'm just curious of the functionality)

Can a channel be easily converted to a flow?

Just learning coroutines (on Android in this example). I was expecting this to hang my app (ANR) because I didn't select a scheduler/dispatcher/thread/whateverItsCalledInKotlin

fun buttonClick() {
    viewModelScope.launch {
        delay(10000)
        _myPrivateData.value = "Two"
    }
}

Is there an easy way to tell when a suspend function will run on a different thread? For example, I will be using coroutines with Retrofit and at first it seems like I would have to tell it to run on a different thread, but it seems like I won't have to do that from the tutorials I've read?

Hi guys. What is the best way to implement some long running background work (like in attached code) using coroutines?

Am experiencing some issue with writing coroutine test and I'm in need of help. So here is the situation, I have 2 different coroutine test classes that pass when I run them individually but when I run the entire tests in the package one passes wile the other fails.

Hi! I need to run two `Job`s simultaneously, how would I do this with coroutines? I understand how to do this when I need return values with

Deferred

and

awaitAll

, but I don't know how I should do this with simple `Job`s

how does one simultaneously iterate on multiple flows in the same loop to get the same index? without calling toList() first

Hi, I’m new to kotlin and coroutines, coming from RxFeedback on Swift/iOS and really excited about the language, platform and tooling. But I’m having some problems understanding exception handling. I’d like to have an event bus and state machine for a uniflow architecture and am used to wrapping this in a nice interface for clients. The reducer function throws an exception if the event in invalid and I’d like to give the client the option of handling this. It looks to me that I need a

SupervisorJob

and to

launch

a new coroutine for every transition so as not to have parent coroutines cancelled, but when I’m calling the API, I can’t quite figure out how to get the exception back to the client. Here’s some code, apologies for any conceptual errors on my part:

class CoroutineExceptionTest: BehaviorSpec() {

    val supervisorJob = SupervisorJob()
    val coroutineScope: CoroutineScope = CoroutineScope(newSingleThreadContext("serviceContext") + supervisorJob)
    class InvalidEventException(override val message: String): Exception(message)
    fun tryTransition(state: Int, event: Int): Int {
        return when (event >= state) {
            true -> event + state
            false -> throw InvalidEventException("Argument $event is invalid")
        }
    }
    val stateChannel = ConflatedBroadcastChannel(0)
    val currentState: Int get() = stateChannel.value

    var eventChannel = coroutineScope.actor<Int>() {
        for (event in channel) {
            if (coroutineScope.isActive) {
                launch {
                    val newState = tryTransition(currentState, event)
                    stateChannel.send(newState)
                }
            }
        }
    }

    suspend fun add(number: Int): Int { // HOW DO I GET EXCEPTION THROWN HERE?
        eventChannel.send(number)
        return stateChannel.value
    }

    init {
        Given("we are in initial state") {
            currentState shouldBe 0
            When("adding 2") {
                add(2)
                Then("the state should be 2") {
                    currentState shouldBe 2
                }
            }
            When("trying to add 1") {
                Then("we should get an invalid event exception") {
                    val exception = shouldThrow<InvalidEventException> {
                        add(1) // THIS IS MY PROBLEM: java.lang.AssertionError: Expected exception CoroutineExceptionTest.InvalidEventException but no exception was thrown.
                    }
                }
            }
        }
    }
}

Is there any reason why I can't

await()

a coroutine? I thought I would be able to just define my

val result = myRetrofitSuspendingCall

and call

await()

on it. It seems like I need to wrap it in an

async{}

block before I do, but it seems a little counterintuitive. (I'm still new at coroutines)? Is this async block needed because coroutines themselves still read like synchronous code?

suspendCancellableTimeoutCoroutine() - cancelled by timeout 1. app crashes

runBlocking {

suspendCancellableTimeoutCoroutine()

}

2. app doesn't crash

runBlocking {

launch {

suspendCancellableTimeoutCoroutine()

}

}

3. app doesn't crash, but I don't see stacktrace in logcat

runBlocking {

val handler = CoroutineExceptionHandler { _, e -> e.printStackTrace() }

launch(handler) {

suspendCancellableTimeoutCoroutine()

}

}

4. This is the behavior that I have now, it works as needed

runBlocking {

try {

suspendCancellableTimeoutCoroutine()

} catch(e: CancellableException) {

e.printStackTrace()

}

}

Could anyone explain to me 2 and 3, please?

Hi, is possible to use coroutines as

EventBus

? i found this gist: https://gist.github.com/svenjacobs/57a21405b2dda4b62945c22235889d4a , but

filter

operator of

openSubscription()

is mark as deprecated!, is there any better sample code?

Hi, I have the following suspended function, which I want to expose as Java APIs(blocking) and Kotlin API(non-blocking) with same method name.

suspend fun getStoreMapping(country: String): List<Store> {
    return client.getStoreMapping(country).stores
}

Currently, the only option I have is to use different method name, which calls the suspended function in

runBlocking {}

like below -

fun getStoreMappingInJava(country: String): List<Store> {
    return runBlocking { client.getStoreMapping(country).stores }
}

This does not behave as I'd expect. Can anyone explain why the first example does not print

output

but the second, similar, example does? No permutation of yield, delay or sleep seems to get

output

to print in Example 1. Example 1 (always prints

start stop

):

@Test
    fun testUnexpectedBehavior() = runBlocking {
        print("start")
        captureScope()
        scope.launch {
            print(" output")
        }
        print(" stop")
    }

    lateinit var scope: CoroutineScope
    suspend fun captureScope() = coroutineScope {
        scope = this
    }

Example 2 (always prints `start output stop`:

@Test
    fun testExpectedBehavior() = runBlocking {
        print("start")
        coroutineScope {
            launch {
                print(" output")
            }
        }
        print(" stop")
    }

Hi all. I’ve got a

CoroutineScope

that works off an IO dispatcher (

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

) and now I’d need to use a class from an external library that is not thread-safe and requires a heavy initialization phase (in the order of a few seconds) per thread. How should I approach this problem? Would I simply initialize one instance and add it to the current

CoroutineContext

? How would its access be regulated to make sure there is no concurrent multi-thread access from the various coroutines spread across multiple threads (I assume

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

spans across multiple threads). Any help appreciated.

Is there any way to “connect” to existing running coroutine job and wait for the result? Case: Let’s say I have the same instance of repository injected into multiple different classes (

client A

,

client B

). Now

client A

calls

repository.getUserData()

and this method internally loads data from network. While request is going

cliebt B

also calls

repository.getUserData()

on the same instance of repository. Can we detekt that network call is in progress and somehow wait for it to finish and return data to both clients?

// Client A:
CoroutineScope(<http://coroutineContentProvider.IO|coroutineContentProvider.IO>).launch {
    reository.getUserData()
}

// Client B:
CoroutineScope(<http://coroutineContentProvider.IO|coroutineContentProvider.IO>).launch {
    reository.getUserData()
}

class Repository() {
    private val job: Job? = null

    suspended fun getUserData() {
        if(job != null) {
            // job is already runned by other client 
            // can I some how "connect" (wait fo the result) to this job
            // instead of proceding and making new network request
        }

        job = CoroutineScope(<http://coroutineContentProvider.IO|coroutineContentProvider.IO>).launch {
            // make network call and return client data
            // return network data
        }
    }
}

I recently upgraded to IntelliJ 2019.3 and to Kotlin 1.3.61, and I'm starting to see a ton of "Inappropriate blocking method call" inspections. They mostly seem to be inside

suspend

functions where they call other

suspend

functions I wrote, so they seem like they may be spurious. Does anyone have any ideas about how to learn more about why these inspections are showing up, to either report a bug or to fix my code if it is in fact bad?

I want to suspend from calling a function until some variable is set, what is the best way to do that? It is sort of the reverse of how I normally think of a Mutex. Is the best way just to create a mutex that is locked and just try to lock it again, and when my variable is set just unlock it, which will then allow calling the function

Is there a naming convention for functions returning a

Flow

? (like Async suffix for the ones returning a

Deferred

)

I'm using retrofit to perform two network calls async, and I then await their results. The thing is that I don't know how to do proper exception handling. I thought it would be straight forward, but apparently this isn't working. Is there something with async that I have to do differently?

val work1 = async { getThingsDone(43) }
val work2 = async { getThingsDoneAgain(123) }

if (work1.await().isSuccessful && work2.await().isSuccessful) {
//do my thing
} else {
//failure message
}

Now with that block above I saw that I was sometimes getting some IOException, and so I wrapped it all in a try/catch.

try {
val work1 = async { getThingsDone(43) }
val work2 = async { getThingsDoneAgain(123) }

if (work1.await().isSuccessful && work2.await().isSuccessful) {
//do my thing
} else {
//failure message
}
} catch (e: IOException){
//network failure message
}

But I still get the IOException. It's as if the try/catch doesn't work.

Currently, when using

flow.produceIn(aScope)

, if the source flow terminates with an exception then the

produce

coroutines fails and cancels the scope and its parent. Wouldn't it be better if it only cancelled the resulting

ReceiveChannel

?

What’s the right way to hold a coroutine until a listener is called? I.e. something like:

fun suspend getCar(): Car {
    val returnedCard
    fetcher.getCar({ car ->
       returnedCar = car 
    })
    // await while car is fetched ???
    return returnedCar
}

Hi, I need a CoroutineDispatcher on Android that will make

delay

wait for given real time, not CPU time. I'm assuming that default implementation doesn't take CPU sleep into account, I'm I right? If during 5min delay CPU sleeps for 15 min, will delay effectively take 20min?

Is there a way to get the parent context of a coroutine (the context the coroutine has been launched from)?

What is the error I am making here? The coroutine should run on the main thread as seen by the log output. However I still get the exception that I can’t use a Toast here.

what is the best way to track down something like this? (android app)

ime  D  Shutting down VM
     E  FATAL EXCEPTION: main
     E  Process: com.dhl.lps.consumer.debug, PID: 18730
     E  kotlinx.coroutines.JobCancellationException: Job was cancelled; job=StandaloneCoroutine{Cancelling}@48c8a44

we have scopes like

MainScope() + CxExceptionHandler(..)

with

launch { .. }

consistently. no

async

for now. so this seems to be some weird/interesting edge case.. 🙂 any pointers on how to track this thing down? 🤔