📣

kotlinx.coroutines

version

0.23.2

fixes bug in

kotlinx-coroutines-io

that causes content damage

what is content damage?

IDE Fatal Errors

I'm still trying to wrap my head around coroutines, and I think I'm almost there. I'm thinking about it in terms of comparison to Rx. Is it safe to say that the practical differences there are: 1. Natural coding style vs streams/operators 2. Less allocation A. Therefore lower memory usage B. Therefore potential faster execution (less GC)

I need to lookup something by id asynchronously and limit it to one lookup per id. Would something like this be the way to do it?

val cache = ConcurrentHashMap<Int, Deferred<String>>()

suspend fun getById(id: Int) : String {
    return cache.computeIfAbsent(id) {
        async { asyncAction(id) }
    }.await()
}

suspend fun asyncAction(id: Int): String = TODO()

Related question. Does await suspend if the deferred is already resolved?

Will there be a

filterIsInstance

method available for

ReceiveChannel

? If not, should I propose one as a PR? on KEEP or on the main repo?

just out of curiosity.. are there plans to implement suspending semaphores like we currently have suspending mutexes? You can make a very basic one right now by sending/receiving from a channel with a finite capacity - but then there's also stuff like acquiring n semaphores, and optimizing by disabling fairness https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Semaphore.html

I tried to implement

rx.Observable.switchMap

for Channels: https://gist.github.com/hannesstruss/927ec8120d7cb312d80685f230d50c6e It's kind of working ("just" deadlocking occasionally), but I found the way there to be very brittle, so much depends on using the right coroutine contexts in the right places. E.g. when not using

Unconfined

for

produce

in the

transform

function, the wrong channel will be selected in

whileSelect

and the outcome will be wrong. Any general hints on how to approach problems like that?

Does it make sense to use coroutines for a worker queue like this?

Hi all, I was surprised to see, that canceling a job that is suspended in a

consumeEach

will execute code after the

consumeEach

. Is this expected behaviour? In the below example, if

job.cancel()

is called, while the job is waiting for new Items from

receiveChannel

it will print out

Done

job = launch(coroutineContext) {
                receiveChannel.consumeEach { r ->
                    output.send(r)
                }
                println("Done")
                output.close()
            }

If relevant, I do have a self contained sample, which is an alternative implementation of @hannesstruss switchMap example. Please let me know if I should post it.

Sure, would love to see it 🙂

https://gitlab.com/snippets/1725697 Kind of hard to read due to all the log messages.

And yet another question. Inside a

produce

, does `this.close()`cancel the producer? Is this documented?

Well, I ask, because I expected a `ClosedSendChannelException`on next send operation as is documented in the API docs, but I don't get to the next send operation but get a JobCanceledException before https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.experimental.channels/-send-channel/close.html

A channel that was closed without a cause throws ClosedSendChannelException on attempts to send or receive. A channel that was closed with non-null cause is called a failed channel. Attempts to send or receive on a failed channel throw the specified cause exception.

Documentation for

produce

mentions:

The running coroutine is cancelled when its receive channel is cancelled.

But does not talk about the SendChannel being closed

with changing

launch

to

async

and

join

to

await

everything start to work as expect, I see all three catch (as described in https://github.com/Kotlin/kotlinx.coroutines/issues/61). But with

launch

its very strange that

method-catch

doesn’t call

Possible newbie question: does it make sense for regular threads and coroutines to communicate/cooperate and what's the best way to achieve that?

I’m testing against a single thread pool for a given name, and I’ve found that the framework adds some coroutines-specific information to the name

what is the coroutines equivalent of caching the current context to continue there at a later point? something like

Scheduler.trampoline()

what is a “restricted suspension coroutine”

what is a non-empty coroutine context, do you mean literally anything that isn’t Unconfined or EmptyCoroutineContext

I can give a smaller version

I'm using an event queue based on a

Channel

like this:

val eventQueue = Channel<ApplicationInstallRequest>()

	init {
		initEventProcessor()
	}

	private fun initEventProcessor() = launch(Unconfined) {
		for (request in eventQueue) {
			processRequest(request)
		}
	}

	override fun onEvent(event: ApplicationInstallRequest) {
		eventQueue.sendBlocking(event)
	}

from an Android SyncAdapter, but even when the sync is finished, the process keeps on going (some kind of leak...), does Unconfined have it's own thread or something?

Ok, I think I solved this with

whileSelect

, but what's the boolean that needs to be returned for? It doesn't look docuemented in the guide's examples..?

I think that maybe in this context of Process, the only blocking that really needs to be done is when retrieving the result of a command sent to the process, so maybe there could be some kind of logic to block only until idle for a certain amount of time and then just yield until the next command is sent...?

Is there currently a way to call

withContext(CommonPool) { ... }

just like

withContext { ... }

? (similarly as async { } can be called on default pool) However just

withContext { ... }

might seem confusing, so maybe it would be nice to add some shorthand with a different name for this if there isn't anything like that already 🙂

I guess you're right, a

Mutex

will anyways protect that

var syncRequested

... I don't need complex logic, just to squash the next bunch of requests to only one syncRequest... thanks! Right now, we're still on targetSdk 22 🙈 (I know I'll have to change that sooner or later)... But I don't want the sync to suddenly be stopped in the middle if the user goes out of the app, so how can it be foreground? @louiscad

Let's say I have some complex message protocol that requires wrapping objects inside objects inside objects (e.g. IP headers on tcp headers on http headers on data). Let's say I want to construct an IP message. Does anyone have any idea what will have better performance? (I suspect the first): 1. passing a lambda from the IP layer to the TCP layer, then wrapping it in a lambda and passing it to the HTTP layer, then wrapping it in a lambda and passing it to the layer that generates data. I.e. the data generation layer can just directly invoke the method to construct+send an IP message 2. passing a channel (or using producer/actor) between each layer, then spawning a coroutine for each layer to wrap and forward the data. I.e. the data generation layer creates its data and sends it through a channel to the HTTP layer, which then wraps it and sends it to the TCP layer, etc.

Nice! And

LifecycleService

also manages its own thread?