kotlinlang #coroutines

I'm starting to explore flows and I've decided to try and deliver android view clicks with it. Here what I've come up with. It works great in android but the

runBlocking

block in test gets stuck at

collect

call. I think the

awaitClose

call suspends the

runBlocking

scope but doesn't affect the android code somehow. Can you please tell me if I'm doing it right and how I can write unit tests for it?

Unbenannt.txt

Saiedmomen

08/15/2019, 5:57 PM

Also SAM conversion isn't working for FlowCollector interface for whatever reason

coder82

08/15/2019, 6:13 PM

is there a way to count the total elements remaining in an channel without consuming them? can it be even done?

Jacques Smuts

08/15/2019, 6:52 PM

I actually have a question related to the awaitClose block. I’m playing around with the above code and I noticed that if the CoroutineContext that is suspended on

.collect

gets cancelled, then the

awaitClose

block gets called. However if a CoroutineContext is suspended on a Channel’s

.receive()

and gets cancelled, then the

invokeOnClose

on that Channel does not get called. You have to manually close the channel. Is that expected behaviour? There’s no automatic propagation of Coroutine cancellation into Channels?

coder82

08/15/2019, 7:34 PM

I wanted to get a customized Dispatchers.IO with custom number of threads, is it ok and wise to use ExperimentalCoroutineDispatcher().blocking(nThreads) ? @gildor

Yuri

08/15/2019, 11:13 PM

Hey guys, I’m trying to migrate RxJava code dealing with Android SharedPreferences to Kotlin Flow. How correct migration should look like?

private val keyChangesObservable: Observable<String>

  init {
    keyChangesObservable = io.reactivex.Observable.create(ObservableOnSubscribe<String> { emitter ->
      val listener = SharedPreferences.OnSharedPreferenceChangeListener { _, key -> emitter.onNext(key) }
      emitter.setCancellable { prefs.unregisterOnSharedPreferenceChangeListener(listener) }
      prefs.registerOnSharedPreferenceChangeListener(listener)
    }).share()
  }

Dariusz Kuc

08/16/2019, 8:16 PM

hello, was wondering if there is some nice idiomatic way of error handling for awaiting for a list of deferred to complete, e.g.

awaitAll

will throw first encountered exception Is there a better way than

val results: List<Deferred<Any>> = ... // whatever
results.joinAll()
for (result in results) {
  val error = result.getCompletionExceptionOrNull()
  if (error != null) {
    // handle error
  } else {
    result.await()
  }
}

coder82

08/16/2019, 9:55 PM

I want to launch 64 coroutines which will all be listening to a fan-out channel, I want the whole thing to suspend untill all 64 coroutines have been launched and are all listening to the channel, then I would return

👍 2

Saiedmomen

08/17/2019, 5:20 AM

The word coroutine isn't in the dictionary and is marked as typo in android studio 3.4.2. Is it the same in intellij? Can someone pull some strings? 😄

☝️ 2

addamsson

08/18/2019, 10:19 AM

I'm trying to write an API which will be called from Java so I'm using this construct:

runBlocking {
    withContext(someContext) {
        // ...
    }
}

My problem is that these functions will be called from each other which would lead to nested `runBlocking`/`withContext` calls. I tried how this works here:

runBlocking {
    println("In runBlocking 0")
    println("Current thread: ${Thread.currentThread().name}")
    withContext(Dispatchers.Single) {
        println("In withContext 0")
        println("Current thread: ${Thread.currentThread().name}")
        runBlocking {
            println("In runBlocking 1")
            println("Current thread: ${Thread.currentThread().name}")
            withContext(Dispatchers.Single) {
                println("In withContext 1")
                println("Current thread: ${Thread.currentThread().name}")
            }
        }
    }
}

but the problem is that here

In withContext 1

never gets printed. What is the best practice when one needs to nest these calls but has to keep the API accessible from Java?

Amanjeet Singh

08/19/2019, 8:35 AM

I created a zipper for my Coroutines but it is not executing my code parallely. It executes the two Coroutines sequentially can I have them work in parallel?

val a = async { doSomeA( ) }

val b = async { doSomeB( ) }

zipper( a.await( ), b.await( ) )

carbaj0

08/19/2019, 8:49 AM

At some point in time startRetrieveCall will change the value of isAttempting. How can I wait for the condition to be fulfilled without having to make a recursive call?

tailrec suspend fun call() : SomeObject =
        if (!isAttempting)
            startRetrievingCall()
        else
            call()

ursus

08/19/2019, 3:08 PM

other than keeping a blocking queue manually, dispatcher abstraction is meant for this I think

Sam Garfinkel

08/19/2019, 8:41 PM

What's the proper data structure to store a synchronized reference to only the most recent value that I can share between a coroutine and any other thread? Should I just use an atomic object or is there something more specialized?

zak.taccardi

08/20/2019, 1:18 AM

I have to use an API that accepts an

Executor

as a parameter. How can I create an

Executor

that is backed by

Dispatchers.Default

? There is no

CoroutineDispatcher.asExecutor()

function unfortunately

myanmarking

08/20/2019, 9:42 AM

what's the idiomatic way of accomplish the same rx clickListener + textChangedListener, where the click triggers a save and the changedListener just validates some edittext. (clickListener.withLatestFrom(textChangedListener) ?

coder82

08/20/2019, 11:59 AM

how do you drain a channel without closing it?

damian

08/20/2019, 3:44 PM

Untitled.kt

napperley

08/20/2019, 10:34 PM

What is the current state of the Kotlin Native version of the KotlinX Coroutines library? Which features are available, and what are the key limitations in the Kotlin Native version?

David Glasser

08/21/2019, 4:05 AM

JUnit 5

Assertions.assertThrows

doesn't work if the block calls a suspend function, because the (Java) function isn't

inline

. What do people do here? Write their own version?

Allan Wang

08/21/2019, 8:01 AM

Are flows cancellable? Can we make them adhere to a lifecycle, or would that be done with channels then converted to flows?

coder82

08/21/2019, 12:26 PM

What's the correct way to call cancel on a job giving another cause so that the coroutine will fail? I tried everything but it looks like that in case of CancellationException with custom cause the coroutine doesn't trigger CoroutineFailureException, so to me it doesn't really fail..., it just cancels... the documentation says it should fail the coroutine. In my case such coroutine is a SupervisorJob child (but that shouldn't affect this behaviour).

ahulyk

08/21/2019, 2:19 PM

coroutines are single-threaded on kotlin-native(iOS) right now - right?

:yes: 2

Sergio Crespo Toubes

08/21/2019, 4:27 PM

Hello, with retrofit what class is the response of a @post call?? Response? only for now the call have finished or its void function? thx

Luis Munoz

08/21/2019, 4:36 PM

Anyone have a simple implementation for a suspending CountDownLatch?

David Glasser

08/21/2019, 5:22 PM

I have a Flow that produces a stream of values. I want to take that stream and launch a coroutine for each emitted function inside Dispatchers.IO (or something like that). I believe I don't want

.flowOn(<http://Dispatchers.IO|Dispatchers.IO>)

for this — that's the opposite of what I want as it affects where the emission happens. Do I just want to do launch with context inside collect?

someFlow().map { ... }.filter{ ... }.collect { x ->
  launch(<http://Dispatchers.IO|Dispatchers.IO>) { doSomethingBlockingWith(x) }
}

Like this?

David Glasser

08/21/2019, 5:56 PM

Also I'm a bit confused about the implementation of

supervisorScope

. (Lots of questions this morning!)

JobSupport.childCancelled

claims that it has this invariant:

* Invariant: never returns `false` for instances of [CancellationException], otherwise such exception
     * may leak to the [CoroutineExceptionHandler].

but SupervisorCoroutine's override always returns false, even for CancellationException? Mostly I'm trying to understand if with CoroutineExceptionHandler/supervisorScope I need my handler to ignore CancellationException. (I think the answer appears to be no from my tests — the CEH never gets the CancellationException, right?)

myanmarking

08/22/2019, 1:07 PM

someAsyncJobHere not to cancel the outer coroutine. how would i define such function ?

coder82

08/22/2019, 2:39 PM

Whet's the difference between these ways of launching a coroutine?

Untitled.txt

Robert Menke

08/22/2019, 2:42 PM

Basic question: Am I able to create a

Channel

, launch a coroutine in

CoroutineContext

A to send some data through the channel, and then launch a coroutine in

CoroutineContext

B in order to receive that data? I threw together a basic example and it looks like you must send/receive to/from a channel in the same

CoroutineContext

. If so, why is this the case?

Title

Robert Menke

08/22/2019, 2:42 PM

Basic question: Am I able to create a

Channel

, launch a coroutine in

CoroutineContext

A to send some data through the channel, and then launch a coroutine in

CoroutineContext

B in order to receive that data? I threw together a basic example and it looks like you must send/receive to/from a channel in the same

CoroutineContext

. If so, why is this the case?

kingsley

08/22/2019, 2:49 PM

Am I able to

Yes

it looks like you must send/receive to/from a channel in the same
CoroutineContext

No. It doesn’t have to be the same coroutine context. You can go through the documentation on channels. It shows several examples of this https://kotlinlang.org/docs/reference/coroutines/channels.html

Robert Menke

08/22/2019, 3:16 PM

Thanks for clarifying. So I’ve put together a really dumb example:

fun main() {
    runBlocking {
        produce()
        consume()
    }
}

val channel = Channel<String>()

suspend fun produce() {
    coroutineScope {
        launch {
            println("Producing")
            channel.send("Hello world")
            println("Done producing")
        }
    }
}

suspend fun consume() {
    coroutineScope {
        launch {
            println("Consuming")
            val value = channel.receive()
            println(value)
        }
    }
}

My understanding is that the

coroutineScope

builder creates a new scope that inherits the context of its parent, which in this case is blocking. I don’t understand why

channel.send

is blocks the thread in this case though.

Done producing

is never printed.

If I reorganize my example to the following it works as expected

fun main() {
    runBlocking {
        produce()
        consume()
    }
}

val channel = Channel<String>()

fun CoroutineScope.produce() {
    launch {
        println("Producing")
        channel.send("Hello world")
        println("Done producing")
    }
}

fun CoroutineScope.consume() {
    launch {
        println("Consuming")
        val value = channel.receive()
        println(value)
    }
}

So, simply put. Why does this not work:

fun main() {
    runBlocking {
        produce()
        consume()
    }
}

val channel = Channel<String>()

suspend fun produce() {
    coroutineScope {
        launch {
            println("Producing")
            channel.send("Hello world")
            println("Done producing")
        }
    }
}

suspend fun consume() {
    coroutineScope {
        launch {
            println("Consuming")
            val value = channel.receive()
            println(value)
        }
    }
}

but this does

fun main() {
    runBlocking {
        produce()
        consume()
    }
}

val channel = Channel<String>()

fun CoroutineScope.produce() {
    launch {
        println("Producing")
        channel.send("Hello world")
        println("Done producing")
    }
}

fun CoroutineScope.consume() {
    launch {
        println("Consuming")
        val value = channel.receive()
        println(value)
    }
}

streetsofboston

08/22/2019, 3:44 PM

Because

coroutineScope { ... }

only resumes/returns after all its child coroutines have finished. In your example, the

courtineScope

calls `launch`creating a child-coroutine that never finishes. It never finishes, because this child-coroutine calls

channeld.send

on a RENDEZVOUS channel. the

send

call will never resume/return.

The

channel.send

never resumes because the

channel.receive

is never called.

The

channel.receive

is never called because the

consume

function is never called. This is because

produce

never resumes/returns in your first code-sample.

kingsley

08/22/2019, 3:45 PM

In your first example.

produce

and

consume

are 2 suspending methods, and they both follow structured concurrency which is good. But this means they will execute sequencially.

consume

will only get called after

produce

has completed. But this will never happen because produce will get suspended waiting for a consumer that will never come The second example is indeed the right way to go about this, since each method is executed concurrently, so produce and consume are able to communicate with each other If you really want 1 to work though, you could set a buffer value for the channel, but that doesn’t really scale if you want implement a proper channel based setup

Robert Menke

08/22/2019, 3:47 PM

Gotcha, thank you @streetsofboston and @kingsley I’m really more interested in understanding the underlying scoping mechanic here. I’ve typically gone with the approach in the 2nd example, but I couldn’t explain to my coworkers why example 1 didn’t work.

Because
coroutineScope { ... }
only resumes/returns after all its child coroutines have finished.

being the key reason 👍

bdawg.io

08/23/2019, 3:46 AM

I would highly recommend watching Roman's talk at HydraConf about Structured Concurrency which goes into the design paradigm and rationale for

coroutineScope

to best understand how you would pick one over the other

https://youtu.be/Mj5P47F6nJg▾

👍 1

View count: 11