this is a hypothetical scenario: I want a

Job

that sends a batch of analytics every certain time. Sending analytics should not return anything and thus is a perfect case for a

Job

. Every so often I may want to check that the Job has completed without exception

Use async instead with

Unit

value, it’s completely valid

because your code now behaves differently depending on environment

in some environments your exception just crash application And don’t give you a chance to handle this

while that's a completely valid suggestion and in reality I may do something like that, I find it hard to believe that no one ever checks if their Jobs completed without any exception

yes, exactly!

this is not the case for laucnh, at least usually

if you didn’t catch exception in launch your programm can crash or not, depends on environment and Thread.UncaughtExceptionHandler implementation

honestly every scenario that I can think of can be replaced with a

Deferred<Unit>

, but I think that it may be good to have such a property so that whenever someone wants to validate that their Job ended without exception they are not forced to change their code to use

Deferred<Unit>

or to validate the message of the

CancellationException

I think I will create the issue just in case

thanks for sharing your views @gildor

My opinion that if you have needs to check launch result you should use async instead. Otherwise one can misunderstand how Job works and believe that you can omit error handling for launch. Deferred behaves identically for different environments/UncaughtExceptionHandlers, provide strict semantics and client knows what to expect (get result or catch an exception)

every scenario that I can think of can be replaced with a

Deferred<Unit>

Not every. There are different reasons for uncaught exception and you want just crash an app. But Deferred user can do not handle result and ignore error, it’s completely client responsibility.

I have created https://github.com/Kotlin/kotlinx.coroutines/issues/249 laying out my thoughts on this

So i've been using Java 8 Completable Futures for running CPU heavy and blocking operations instead of coroutines to avoid starving the CPU time of coroutines scheduled on the same thread:

suspend fun <T> asyncResult(execute: () -> T): T {
    data class AsyncResponse<T>(val result: T?, val exception: Exception?)

    val responseChannel = Channel<AsyncResponse<T>>()
    val asyncResponse: AsyncResponse<T>?

    CompletableFuture.runAsync {
        try{
            val executionResult = execute()
            launch(NoopContinuation.context) {
                responseChannel.send(AsyncResponse(
                        result = executionResult,
                        exception = null
                ))
            }
        } catch(exception: Exception) {
            launch(NoopContinuation.context) {
                responseChannel.send(AsyncResponse(
                        result = null,
                        exception = exception
                ))
            }
        }
    }

    asyncResponse = responseChannel.receive()

    if (asyncResponse.exception != null) {
        throw asyncResponse.exception
    }

    return asyncResponse.result!!
}

I guess I should just use

newSingleThreadContext()

instead? https://github.com/Kotlin/kotlinx.coroutines/blob/master/coroutines-guide.md#dispatchers-and-threads

@Michael Kotlikov what this code does is essentially

async

coroutine builder, in other words in your code instead of calling

asyncResult {lambda}

you can do simply

async {lambda}

and not even build this function. You can specify where the coroutine will be executed by passing

CoroutineContext

to

async

like this:

async(routineContext) { lambda }

, where I have

routineContext

defined as follows:

internal val routineContext: CoroutineContext by lazy {
        log.debug("Initializing pool with configuration $configuration")
        when (configuration) {
            is SingleThreadConfiguration -> newSingleThreadContext(configuration.name)
            is FixedThreadPoolConfiguration -> newFixedThreadPoolContext(configuration.threads, configuration.name)
            is CachedThreadPoolConfiguration -> CachedCoroutineDispatcher(configuration.name)
            is ForkJoinPoolConfiguration -> ForkJoinCoroutineDispatcher(configuration.name, configuration.parallelism)
            else -> throw ConfigurationException("Unknown thread pool requested")
        }
    }

configuration is an instance of data class defining which executor should be used for a

CoroutineContext

There are

newSingleThreadContext

and

newFixedThreadPoolContext

provided by coroutines library out of box, and you can construct your own

CoroutineContext

based on any

ExecutorService

like this:

open class BaseCoroutineDispatcher(private val executor: ExecutorService) : CoroutineDispatcher() {
    override fun dispatch(context: CoroutineContext, block: Runnable) {
        executor.execute(block)
    }
}

@Sergey Chikin is the

CachedCoroutineDispatcher

the same as Java's native

newCachedThreadPool

under-the-hood or will I need to roll my own

CoroutineContext

?

like this? I feel like im missing something because every call will create a new thread pool. I got some learning left to do.

fun newCachedThreadPoolContext() = CachedThreadPoolDispatcher(Executors.newCachedThreadPool())

class CachedThreadPoolDispatcher(private val executor: Executor) : ExecutorCoroutineDispatcherBase() {
    override fun dispatch(context: CoroutineContext, block: Runnable) = executor.execute(block)
}

What about this with a static cachedThreadPool?

fun newCachedThreadPoolContext() = CachedThreadPoolDispatcher()
class CachedThreadPoolDispatcher : ExecutorCoroutineDispatcherBase() {
    companion object {
        private val executor = Executors.newCachedThreadPool()!!
    }
    override fun dispatch(context: CoroutineContext, block: Runnable) = executor.execute(block)
}

is it a premature optimization to check and not start a coroutine on a CorroutineContext if it’s going to be on the same context as the current one, and use EmptyCoroutineContext instead

asking question from yesterday

they operate at different levels of abstraction

so it's not exactly 1:1

or, put another way, the entirety of RxJava API could be backed by coroutines as an implementation

What I have realized for myself is that if one expects lot of data-flow from zillion of different directions, zippings, mergings, unzippings, re-zippings etc. transformations on those data flows, then Rx will do what it's good for. Buf if Rx is just used for simple data flow transformations and mostly used as multithreading substitute for callback hell, then Coroutines will be better. Especially it's interesting to look into Coroutines channels/actors and that fancy stuff, which might turn out to be better choice... at the end of the day there is only one good advice "Use your brain"

To me Rx is more about monadic composition, push based approach and more important it actually provides IMO an abstraction on top of sync vs async behaviour. Being parametrised over schedulers, allows to write your business logic in a way that is often independent on whether the code is executed synchronously or async (i'm simplifying things a bit here), in some cases allows the consumer to later decide the concurrency model by injecting the proper scheduler, and the built-in testscheduler allows to test the async implementation in unit tests just by plugin a different scheduler (you can time travel etc, something difficult to achieve without a proper abstraction on top of the async mechanism).. not to mention, from a purely pragmatic point of view, the number of built-in Rx operators (deboucing, combineLatest, merge, zip, etc). It's hard to make a comparison with coroutines, to some extend it's kind of asking a comparison between thread executors (and other java concurrency primitives) and rx (even though suspended functions indeed can really provide a way to reason about cooperative tasks)

Here are few benefits of channels over rx (IMO):. * Can handle null emissions (and in a safe way, thanks to Kotlin type system) * Threading is explicit. With Rx, it's often hard to know on what thread pool each piece of code is executed * ReceiveChannel and SendChannel use [on-site variance](https://kotlinlang.org/docs/reference/generics.html#declaration-site-variance) which can make thinks easier when generic variance is involved. * It is (a lot) easier to create custom operators with channels. * When it comes to parallelism, Coroutines are (IMO) way easier to use and produce clearer code. * pipeline pattern out of the box. So far, I haven't run in a use case that Rx solved and I couldn't handle in a nice and easy way with Channels. If you know one, i would be interested to look at it.

I think I have a crash I'm hunting for a long time down to a simple example

@Paul Woitaschek uploaded a file: RxCoroutineCrash.kt