I'm confused by how to properly use CoroutineExceptionHandler. I tried:

import kotlinx.coroutines.CoroutineExceptionHandler
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlin.system.exitProcess

fun main() = runBlocking {
	val handler = CoroutineExceptionHandler { _, t ->
		println("handling exception: $t")
		exitProcess(1)
	}
	coroutineScope {
		launch(handler) {
			println("running")
			throw Exception("OMG")
		}
	}
	Unit
}

I would expect this to print "handling exception" and exit the process. Instead it shows that an exception was unhandled in

main

(I'll paste in the thread)

Hi all, I'm currently trying to replace usages of

Channel

with

Flow

in my Android application. But now I'm stuck in a conceptual thinking problem: Right now I use

Channel

to handle UI events on Android views, for instance clicks inside a

RecyclerView.Adapter

. I create an instance of

Channel

in the view (or presenter, it doesn't matter) and pass it to the

RecyclerView.Adapter

. Then inside the adapter I could just pass click events to the channel. But with

Flow

, the flow needs to be created at the place where the data is emitted. However I don't want to have x flows in my Adapter for every view the adapter manages. Is

Flow

even suited for UI events or are UI events classically hot streams? Am I just trying to apply

Flow

where

Channel

is the right choice after all?

i tried onStart but obviously, it still is debounced

don't know if its the most efficient tho. but it seems to work

is liveData block still a thing? I can't seem to use it even when I implement everything that is needed.

I have a service that uses ktor + websockets + broadcast channel. After some hours it stops responding to http requests and

sayX()
GlobalScore.launch(Dispatchers.Default) { sayY() }

triggers X but not Y... it looks like the whole coroutines engine goes down. Is that possible?

Hi!, all... which is the difference between GlobalScope.async and CoroutineScope(context).launch?? can someone explain me

Hey folks, is there any easy way to auto close the BroadcastChanel as soon as the last subscribed channel is closed?

Good morning, is there an operator for concatenating two `Flow`s, something like:

val flow1 = flowOf(1, 2, 3)
val flow2 = flowOf(4, 5, 6)
val flow3 = flow1.concat(flow2) // = 1, 2, 3, 4, 5, 6

The order of

flow3

does not matter here, so it also might be

1, 4, 2, 5, 3, 6

or any other order, depending on how the values are produced in

flow1

and

flow2

.

On the JVM the Default and IO dispatchers are backed by fixed-size threadpool? right? according to the java docs of

newFixedThreadPool

it is backed by an unbounded job queue. is that true for our dispatchers, too?

Hi, what is the canonical way to create a flow that replays its history whenever it's subscribed to? I have a Flow<String> that contains some log lines and whenever someone calls my API I'd like it to publish all previous log lines and additionally any new ones, by returning the flow as a server-sent event

Hi, guys, is there any API in

Flow

like

Observable.timeout

?

What is the use case for flows?

I want the subject version of a flow (a flow I can force to emit). Does this exist yet? or do I need to back it with an external channel for now?

private val channelA = Channel<Int>()
private val channelB = Channel<Int>()
private val flowA: Flow<Int> = channelA.consumeAsFlow()
private val flowB: Flow<Int> = channelB.consumeAsFlow()
suspend fun whenSourceAEmits(value: Int) {
    channelA.send(value)
}
suspend fun whenSourceBEmits(value: Int) {
    channelB.send(value)
}

Somehow, the

withTimeout

doesn’t work on iOS without

runBlocking

. Is there another way round?

The documentation for

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

says:

This dispatcher shares threads with a [Default][Dispatchers.Default] dispatcher, so using
`withContext(<http://Dispatchers.IO|Dispatchers.IO>) { ... }` does not lead to an actual switching to another thread &mdash;
typically execution continues in the same thread.*

This behavior seems wrong to me. Isn’t the goal of

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

to keep

Dispatchers.Default

threads free. Sharing threads between the dispatchers seems counterintuitive to this goal.

Hi folks. I am confused with

withTimeout

. I have a function which can block the thread for undefined time. The caller should get timeout exception after timeout. The function doesn't support cancellation. There is a similar case, but it doesn't work as described. https://stackoverflow.com/questions/47635367/kotlin-coroutines-with-timeout/47641886#47641886 The snippet behind blocks for 10 seconds.

runBlocking {
   val work = async { Thread.sleep(10000) }
   withTimeout(100) {
        work.await()
   }
}

There are 3 solutions, which helps to make it working: - use

GlobalScope.async

- use

async(NonCancellable + <http://Dispatchers.IO|Dispatchers.IO>)

. Scope will await for async result if i don't specify

NonCancellable

.

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

is needed because

runBlocking

is single threaded. - use custom

scope

Am I missing something? Is there simpler approach?

I'd like to Unit Test the following class :

class MyClass(private val scope: CoroutineScope) {
     // launches new coroutines in the passed scope 
}

Some of my test scenarios requires that the passed

CoroutineScope

be cancelled. How would you write such test with

runBlockingTest

?

I want to get Result<T> class(or something like it) after finished launch block. Because I used invokeOnCompletion with CoroutineExceptionHandler. But It is not enough for me. There is no value of success case. So I made this code. But I want to find a better way. If somebody knows a better way. comment please.

launchCatching(
    block = {
        getRandomNumber()
    },
    result = {
        println("result is $it")
    }
).invokeOnCompletion {
    it is Throwable
}

fun <T> CoroutineScope.launchCatching(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> T,
    result: (result: Result<T>) -> Unit
) = launch(context, start) {
    result(
        runCatching {
            block()
        }
    )
}

fun <T> CoroutineScope.launchCatching(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> T,
    onSuccess: (value: T) -> Unit,
    onFailure: (exception: Throwable) -> Unit
) = launch(context, start) {
    runCatching {
        block()
    }.fold(
        onSuccess,
        onFailure
    )
}

https://android.jlelse.eu/rxbus-kotlin-listen-where-ever-you-want-e6fc0760a4a8 Hello all, how can I create an event bus like this with current coroutine apis?

i have a

fun

that creates a

CoroutineExceptionHandler

. when i do

launch(exceptionHandler()){throw RuntimeException()}

it gets called but with

runBlocking

the exception is not caught. is that expected behavior?

Hello, how can I make forEach continue even if some elements get cancelled e.g.

@Test
    fun example() = runBlockingTest {
        val actual = mutableListOf<Int>()

        launch {
            (0..5).forEach { i ->
                supervisorScope {
                    if (i == 1) cancel()
                    actual.add(i)
                }
            }
        }

        assertEquals(listOf(0, 2, 3, 4, 5), actual)
    }

I have a

kotlinx.coroutines.io.ByteReadChannel

. How do I read a utf8String of certain length?

Seems like this something to do with the Channel buffer (when exists)? Is there any reason the buffered messages will not be delivered?

Is it possible to re-open a cancelled/failed

BroadcastChannel

? I'd like to re-execute the block passed to

broadcast

when receiving from it after being closed :

val channel = broadcast<String>(capacity = Channel.CONFLATED, start = LAZY) {
    val callback = object : Subscription {
        override fun onValue(value: String) {
            // May be called multiple times.
            offer(value)
        }
        override fun onError(error: Exception) {
            close(error)
        }
    }

    subscribe(callback)
    awaitClose { unsubscribe(callback) }
}

// At some point in time, receive latest value from the channel. 
// Because the broadcast is lazy, this registers a callback and wait for a value.
// Lets assume that the callback failed, the exception is rethrown here.
channel.consume { receive() }

// At a later time, I'd like "receive" to re-execute the producer coroutine, re-registering a new callback.
channel.consume { receive() }

We're using JUnit and doing the pattern where we write lots of tests like

@Test fun foo = runBlocking {

. (We haven't tried using

runBlockingTest

since we haven't needed the extra time-related power it brings — but maybe this is the answer to our issue?) A problem we've been running into is that it's quite easy to have a test whose final line doesn't return Unit. Then the runBlocking doesn't return Unit and neither does the test function, and JUnit just skips it. There is an IntelliJ inspection for this at least but it seems like a bit of a pain. What do people do for this?

@Test fun foo: Unit = runBlocking {

?

@Test fun foo() { runBlocking {

? Something else?

Whats the closest equivalent to below code using

Flow

. Or in other words the

create

and

just

operator equivalents.

Observable
		.create(
			{ emitter: Emitter<Boolean> ->
                                 // Do stuff and onNext
                        },
			Emitter.BackpressureMode.BUFFER
                  )

// Also for this 

Observable.just(true)

I am using bellow builder but wanted to see other alternatives

flow { flowCollector -> ...// Do stuff and emit }

is there a way to get the associated

Job

from a coroutine context?

What happens if I use withContext(Dispatchers.IO) and execute a lot of calls, is there a limit? what if all the threads are busy spinning? How to provide backpressure? Don't want to reach out of memory. @gildor

When should one use async-await Vs withContext