Hey guys, I think there is a problem when calling

join

on an exceptionally completed job. After days digging, I figure out that if the

join

call falls into the fast-path or successfully suspending, a CancellationException will be thrown in this case, but if the

join

call falls into the

joinSuspend

path but not successfully suspending, no CancellationException will happen. I just wonder that is it a bug to get inconsistent result from

join

with the same code? I have created a gist here https://gist.github.com/enbandari/0321cffa792e0f98d90e0e77682aeaed to make it more clear.

Is there an example somewhere how to implement switchMap for Kotlin Flows?

Is

runBlocking

safe for production if we have helper suspended functions, and are using an interface where it’s okay to block the thread? For instance, I’m building a notification widget, which requires updates to be done in

onDataSetChanged

, before the provider will apply the data

Why does the following code crash with a null exception?

val channel = ConflatedBroadcastChannel(Unit)
channel.asFlow().produceIn(GlobalScope).poll()!!

If I understand correctly this shouldn’t be happening because the channel has a default

Unit

value. The following code, however, does not crash:

val channel = ConflatedBroadcastChannel(Unit)
channel.openSubscription().poll()!!

I don’t get why this works but the first example doesn’t. Could someone please explain this behaviour?

I need to adapt a suspending function that returns

Flow

to a non-suspending function that returns

Flow

. Is this the correct way to do that?

suspend fun suspendingFlow(): Flow<String> { ... }

fun nonSuspendingFlow() = flow {
	suspendingFlow().collect {
		emit(it)
	}
}

I'm trying to write a fire-and-forget service which offloads jobs to worker threads am I doing it the right way? I don't want the

Test

service to fail even if the

Job

fails because each

Job

is independent, I just want the error handling.

class Test : CoroutineScope {

    override val coroutineContext = Dispatchers.Default + SupervisorJob()

    fun doSomething(): Job {
        val handler = CoroutineExceptionHandler { _, exception ->
            // log error
        }
        return launch(handler) {
            // do things
        }
    }
}

I’m still not sure about coroutine extensions, does this look ok to you?

suspend fun <T> withMinimumDuration(duration: Long, block: suspend CoroutineScope.() -> T) =
    coroutineScope {
        val delay = async { delay(duration) }
        val task = async { block() }
        delay.join()
        task.await()
    }

After converting some threads code to coroutines code, my benchmarks showed that memory consumption decreases but execution time is much longer with coroutines. I watched some talks and read multiple articles says that coroutines are “very light-weight threads”. So I see that gives me a better results with memory, but is it possible that conversion causes execution time increase? Or its just may be my wrong way of implementing them

Hello. I have a question about combining coroutines and

Result<T>

. I have the code like following:

launch {
    runCatching { doSomething() }
        .onSuccess {
            doSuccessProcess(it)
        }
        .onFailure {
            doFailureProcess()
        }
}

I wanna handle errors if

doSomething()

fails. But this code catches

CancellationException

as well when

launch

was canceled. Are there any problems? Can I just ignore the

CancellationException

?

just(1, 2, 3)

->

flowOf(1, 2, 3)

error(Exception())

->

flow { throw Exception() }

I'm reading the docs about `Channel`s and I was a bit puzzled. In the Fan-out examples it uses extension functions on

CoroutineScope

, but in the Fan-in examples it uses suspending functions. Is there a practical difference between these two approaches? ---->

@JvmStatic
    fun main(args: Array<String>) = runBlocking {

        val channel = Channel<String>()
        launch { sendString(channel, "foo", 200L) }
        launch { sendString(channel, "BAR!", 500L) }
        repeat(6) {
            // receive first six
            println(channel.receive())
        }
        coroutineContext.cancelChildren() // cancel all children to let main finish

    }

    suspend fun sendString(channel: SendChannel<String>, s: String, time: Long) {
        while (true) {
            delay(time)
            channel.send(s)
        }
    }

versus

@JvmStatic
    fun main(args: Array<String>) = runBlocking {

        val channel = Channel<String>()
        sendString(channel, "foo", 200L)
        sendString(channel, "BAR!", 500L)
        repeat(6) {
            // receive first six
            println(channel.receive())
        }
        coroutineContext.cancelChildren() // cancel all children to let main finish

    }

    fun CoroutineScope.sendString(channel: SendChannel<String>, s: String, time: Long) = launch {
        while (true) {
            delay(time)
            channel.send(s)
        }
    }

Both are doing the same if I understand the concepts correctly and also produce the same output. When should I use which?

Does anyone know why the

produce

builder doesn’t take a

CoroutineStart

parameter? (cc @elizarov)

Im having a hard time illustrating cooperative cancellation to users of my library. I tried taking a shot at putting together a diagram but I feel like its not very clear. Does anyone know of any good resources to illustrate this? If not, Im open to any suggestions for improving this diagram.

Im using the following, and It doesn't seem to be removing the warnings from my project. Can anyone suggest an alternative?

tasks.withType(org.jetbrains.kotlin.gradle.tasks.KotlinCompile) {
            kotlinOptions.freeCompilerArgs += ["-Xuse-experimental=kotlinx.coroutines.ExperimentalCoroutinesApi"]
        }

I am wondering about handling exceptions (throwables) in this example:

val xHandler = CoroutineExceptionHandler { c, e ->
    println("Handled Crash in coroutine $c")
    e.printStackTrace(System.out)
}

fun launch_handleExceptionInCoroutineExceptionHandler(): Unit = runBlocking(xHandler) {
    launch(<http://Dispatchers.IO|Dispatchers.IO> + xHandler) {
        delay(1000)
        throw Exception("Some Exception 1")
    }
    delay(2000)
}

fun main() {
    Thread.setDefaultUncaughtExceptionHandler { t, e ->
        println("App Crash in thread $t")
        e.printStackTrace(System.out)
    }
    launch_handleExceptionInCoroutineExceptionHandler()
}

The

xHandler

is never called. The output is produced by the default-uncaught-exception-handler instead. I expected the

xHandler

to be called. If I provide another scope to the

launch

, eg

GlobalScope.launch { ... }

it works as expected and the

xHandler

is called. Why is the

xHandler

not called when using the `runBlocking`’s scope? (just to be sure, I provided the

xHandler

in 2 places, when calling

runBlocking

and when calling `launch`…)

If I have a

java.util.concurrent.ThreadPoolExecutor

with a keepAlive of 0, will it respect that

keepAlive

as a

CoroutineContext

?

📣 📣 📣

kotlinx.coroutines

1.2.1 is here! Changelog: • Infrastructure for testing coroutine-specific code in

kotlinx-coroutines-test

(read more here https://github.com/Kotlin/kotlinx.coroutines/tree/master/kotlinx-coroutines-test#runblockingtest). Thanks Sean McQuillan for the contribution! •

Job.asCompletableFuture

extension in jdk8 module • New operators in `Flow`:

switchMap

,

debounce

,

sample

• Various fixes and improvements, full changelog: https://github.com/Kotlin/kotlinx.coroutines/releases/tag/1.2.1

Is there a way to make a transformation of Flow with a statefull collector? I am migrating kmath streaming API to flow and I need to implement a chunking and windowing operation (take

n

subsequent elements and join them into one).

Another question. It is probably somewhere in the documentation, but is the

Flow

map operation parallelizable?

Here is my implementation based on

flowOn

. I am not sure I've done context switching correctly. And I can't use internal functions like

getCancellationException

for proper exception propagation. I will open a ticket for this.

I have a system where I have multiple `Resource`s where only one process (coroutine) should be able to access it at a time. The main function of a

Resource

is

suspend fun take(priority: Int = 0, immediate: Boolean = false)

where a task with a lower priority will be scheduled first and higher later. The

immediate

argument will determine whether the currently executing process should be interrupted if

take(...)

has a lower priority than the currently executing task. A resource is given back with

Resource#giveBack()

. Any recommendations on how I could do this idiomatically?

I'm trying to implement an ideal ratelimiting system with coroutines. Basically, I want to send requests to the server, and if it told me that I'm being ratelimited, requests should be delayed till the ratelimit is over. All this while also returning responses in time (we don't wait for all requests to finish, and provide the response to the caller whenever it comes). I already have an implementation but it's far from ideal. What's the best way to do this?

Is there an equivalent for

Thread.UncaughtExceptionHandler

with `CoroutineExceptionHandler`s that is a globally handled exception handler?

Do I need to keep a strong reference to a continuation given to me in

suspendCancellableCoroutine

? I need to work around a memory leak and I was planning on doing that by keeping a weak ref to the continuation

While experimenting with parallel flow extensions, I found a not so obvious behavior. This code:

val deferred = coroutineScope {
                async(start = CoroutineStart.LAZY) { println("async") }
            }
            deferred.await()

causes a infinite block. I understand that it is caused by interference of lazy deferred with structured concurrency, but it significantly reduces application for lazy deferred. Is it intended to be this way?

I am testing Flow with Observables, having this code:

runBlocking(<http://Dispatchers.IO|Dispatchers.IO>) {
        val myList = mutableListOf(1).asObservable()

        launch {
            myList.asFlow().collect { println("A $it") }
        }

        delay(1000)

        myList += 2

        launch {
            myList.asFlow().collect { println("B $it") }
        }

        delay(1000)
    }

Which produces:

A 1
B 1
B 2

Is there some sort of non-terminal

collect

yet that would listen for further changes, so that I would see

A 2

in the output when

myList

gets updated?

hi

i am confuse with this syntax 
fun task(bloc: suspend () -> Unit) { .... }

what does mean call suspend function inside regular function ?

I want to create a suspending function that calls

startActivityForResult

and it suspends until

onActivityResult

is called. Is using a Channel the right way to do that?

kotlin
class MyActivty : Activity(){

  private val channel = Channel<Uri>()

  override fun onActivityResult(data: Intent){
    val uri = data.data
    channel.offer(uri)
  }


  suspend fun soSthWithStartActivtyForResult() : Uri {
     startActivityForResult(intent, 42)
     return channel.receive()
  }
}