kotlinlang #coroutines

Hi! I have a stupid question about coroutines. So we have Channels - hot streams and Flows - cold streams. The question is: if we converting Channel into Flow - we are getting kind of 'hot' Flow???

ansman

07/25/2019, 2:29 PM

It does seem to be a tooling issue so perhaps it should be reported to the Android team

ahulyk

07/25/2019, 2:47 PM

Channel/Flow support is missing for Room right now? Right?

groostav

07/26/2019, 7:52 AM

how bad an idea is mutable state on a custom CoroutineContext element? like all the way bad?

CLOVIS

07/27/2019, 9:38 AM

I watched the amazing talk from KotlinConf 2018 and I'm trying to use the same example with an extension function on CoroutineScope, however I can't seem to be able to import it in Kotlin/Multiplatform, is it not implemented? IntelliJ suggests CoroutineContext, but I don't know if those are equivalent or not

Mohamed Ibrahim

07/27/2019, 11:05 AM

one of the things that I don’t get it, that Kotlin is full of features that help you write code in functional style, but they built Coroutines in a way that doesn’t change your “Imperative style” in code .. wasn’t be a good thing to build Coroutine so it fit nicely with a functional code.

Marko Mitic

07/27/2019, 11:14 AM

Does anyone have an example of OSS project using coroutines? (I run out of Roman's blog posts to read)

CLOVIS

07/27/2019, 7:25 PM

Now my question is: what if I want to have a long term cache? For example a UserRepository. I can query the Repository for the user that belongs to a specific ID. What I want is that this Repository would save the users that are requested and saves them for some time. I know how to do that with synchronisation primitives, with shared mutable states, but that's the opposite of the coroutines mentality. How would you do that with coroutines?

➕ 1

dave08

07/28/2019, 2:47 PM

Is there any better way to wrap this function in Caffeine Cache:

suspend fun <K: Any, V> AsyncCache.getSuspending(key: K): V {
    val outerContext = coroutineContext
    return get(key) { k, executor ->
        val innerContext = outerContext + Job() + executor.asCoroutineDispatcher()
        CoroutineScope(innerContext).async {
            loadValue(k) // loadValue is a suspend function defined elsewhere
        }.asCompletableFuture()
    }.await()
}

dave08

07/28/2019, 2:57 PM

shouldn't it be

coroutineScope { async(executor.asCoroutineDispatcher()) { ... } }

or is there a difference?

CLOVIS

07/28/2019, 5:37 PM

If I cancel a coroutine, it dies and frees all its memory, right? So a valid "clear the cache" mechanism would be to have a coroutine that handles the cache, and when we want to clear it, to kill the scope in which that coroutine runs, and to start another one?

CLOVIS

07/28/2019, 6:56 PM

Is there a ReceiveChannel equivalent where every coroutine waiting to receive does receive, instead of a single random one? The idea is that a single coroutine handles internally whether or not to download some objects from the internet (by delegating to a worker pool, then caching the result). Let's call it cacheManager. The outside program can send an ID to cacheManager over its ReceiveChannel. You can then listen on cacheManager's SendChannel to get the object back, but if multiple people query at the same time, nothing ensures you'll get your request's results...

Adam Powell

07/28/2019, 7:12 PM

anything you use for a return path there will involve some overhead, but you can read the source for Channel, they're fairly small/don't involve a lot of initialization overhead. You might find

CompletableDeferred

to be a better option for just a single value though

Seri

07/28/2019, 10:49 PM

On Android, what CoroutineContext would y’all use for a fire-and-forget network request where the app doesn’t care about the response? Is this a valid case for

GlobalScope.launch { ... }

👍 1

Adam Powell

07/28/2019, 10:53 PM

are there any plans to expose the

ChildCancelledException

mechanism used by

scopedFlow

and `channelFlow`/`callbackFlow` before the latter two lose their

@ExperimentalCoroutinesApi

status, or otherwise make external cancellation of a `launch`ed child in those scopes behave that way by default?

Pablichjenkov

07/28/2019, 11:58 PM

channel.send() or actor { consumeEach {}} may suspend on queue full or empty. I wonder how this suspension is implemented under the hood. How the

suspended send

waits until the Queue has more capacity available. Or in other words how the send coroutine is informed that the Queue has capacity available. It is usual although consider a bad practice in Java, to synchronize on a queue monitor and wait for its size to change. Then whatever consuming thread that acquires the monitor will consume some items release the monitor and notify other waiters. I guess that in coroutines, above mechanics is not used since it blocks/sleeps the

waiter

thread. It probably works by re-scheduling another

suspend send

coroutine in the executor. Maybe not, maybe it spins lock on the Queue capacity. Does anyone can explain what happens internally?

Marko Mitic

07/29/2019, 12:38 AM

Silly question, could coroutines help improve implementation of observer pattern?

Alexander Weickmann

07/29/2019, 8:34 AM

Hi all, is it possible to supervise an object between different coroutines? My service is acting as a gateway between a frontend and other http apis. So I represent each Http Api as an object, and I reuse http client instances like this:

object HttpApi {

    private val httpClient = httpClient {
        it.followRedirects = false
    }

    private suspend fun <T> httpClientWithLogin(work: suspend (HttpClient) -> T): T {
            httpClient.login()
            return work(httpClient)
    }

   suspend fun someApiMethod(): Result = httpClientWithLogin { client ->
      ...
   } 
}

The problem is, that the httpClient might become unusable, e.g. because it is closed for whatever reason. I want to protect against that by recreating the httpClient instance. So I am doing this:

object HttpApi {

    private var httpClient = recreateHttpClient()

    private fun recreateHttpClient() = httpClient {
        it.followRedirects = false
    }

    private suspend fun <T> httpClientWithLogin(work: suspend (HttpClient) -> T): T {
           return try {
            httpClient.login()
            work(htpClient)
        } catch (e: Exception) {
            log.error("Recreating HTTP client", e)
            httpClient = recreateHttpClient()
            throw e
        }
    }

   suspend fun someApiMethod(): Result = httpClientWithLogin { client ->
      ...
   } 
}

I don't particularly like it. For one, I am not sure whether I get multi-threading problems with recreateHttpClient. Also, if a call fails, it is not immediately retried. Can this be handled more elegantly with supervisor coroutine scope? How would you approach this scenario?

Patrick Jackson

07/29/2019, 5:30 PM

Flow question: What is recommended way to get a flow of tap events. In android we give views a callback for these events. This is where many devs will try to get started with flows, yet is unclear how to do so. Is best way to use flowViaChannel with a callback that does channel.offer()? Any thoughts on this?

zak.taccardi

07/29/2019, 6:42 PM

API design question - which would you use to pass around a read-only value that gets initialized once and cached? 1.

suspend () -> T

Deferred<T>

Vote 1️⃣ for

suspend fun () -> T

or 2️⃣ for

Deferred<T>

1️⃣ 7

2️⃣ 4

groostav

07/30/2019, 3:03 AM

is it a bug that

runBlocking(JavaFx) { runBlocking(JavaFx) { }}

causes deadlock but

runBlocking(JavaFx) { runBlocking { }}

does not?

myanmarking

07/30/2019, 1:43 PM

suspend fun requestStepsActivity(): List<Something> = suspendCancellableCoroutine { emitter ->
    val listener = object : SimpleEventListener() {
        override fun onSent() {
            emitter.resume(Something)
        }
        override fun onFailedSending() {
            emitter.resumeWithException(RuntimeException())
        }
    }

    try {
        someApi.addListener(listener)
        someApi.requestData()
    } catch (e: Throwable) {
        emitter.resumeWithException(e)
    } finally {
        someApi.removeListener(listener)
    }
}

Matej Drobnič

07/30/2019, 1:49 PM

I guess it is pretty late for this, but wonder what is the reason to move Flow's functionality behind extension method, while method that should not be used is the regular interface method with the same name? When I type

emptyFlow<Unit>().collect

, method on the top of the list in IDE's completion will be the internal method that should not be used. At first IntelliJ did not suggest any extension methods to me. I was baffled since all examples just mention to call

.collect {}

, but all I got was method marked as internal with documentation saying to not use. Then on subsequent tries, IDE started suggesting extension methods, but it was not transparent process.

➕ 1

JoakimForslund

07/30/2019, 2:38 PM

Question; I tried to add 'org.jetbrains.kotlinx:kotlinx-coroutines-test:1.3.0-RC' to my

androidTest

target in a mpp, however, stuff like

runBlockingTest

and

newSingleThreadContext

is not found Am I missing something?

07/30/2019, 3:34 PM

is it actually invalid to invoke a

Continuation

in the same stack frame? or is that just a best-practice thing to avoid blowing the stack?

Thomas

07/30/2019, 7:04 PM

Sometimes I get reported exceptions like this:

Fatal Exception: java.io.EOFException
       at okio.Buffer.skip + 881(Buffer.java:881)
       at okio.Buffer.clear + 874(Buffer.java:874)
       at okio.RealBufferedSource.close + 462(RealBufferedSource.java:462)
       at okio.RealBufferedSource$inputStream$1.close + 450(RealBufferedSource.java:450)
       at kotlinx.coroutines.io.jvm.javaio.ReadingKt$toByteReadChannel$2.invokeSuspend + 96(ReadingKt.java:96)
       at kotlinx.coroutines.io.jvm.javaio.ReadingKt$toByteReadChannel$2.create(ReadingKt.java:10)
       at kotlinx.coroutines.io.CoroutinesKt$launchChannel$job$1.invokeSuspend + 123(CoroutinesKt.java:123)
       at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith + 33(BaseContinuationImpl.java:33)
       at kotlinx.coroutines.DispatchedTask.run + 241(DispatchedTask.java:241)
       at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely + 594(CoroutineScheduler.java:594)
       at kotlinx.coroutines.scheduling.CoroutineScheduler.access$runSafely + 60(CoroutineScheduler.java:60)
       at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run + 740(CoroutineScheduler.java:740)

It contains no references to my own code. This makes it very difficult to fix these exceptions. I already have debug mode enabled for Coroutines. Does anyone here have any suggestions how I could fix something like this?

Icaro Temponi

07/30/2019, 7:26 PM

Hi everyone. So, I have the following function:

private fun barcodeFlow(): Flow<Barcode> = callbackFlow {
        val onNext: (t: Barcode) -> Unit = { barcode -> offer(barcode) }
        val onError: (t: Throwable) -> Unit = { ex -> throw ex }
        val disposable = binding
                .barcodeView
                .drawOverlay()
                .getObservable()
                .subscribe(onNext, onError)
        awaitClose { disposable.dispose() }
}

I just would like to know if the exception thrown inside the onError lambda will be propagated down the stream of this callbackFlow to be caught on a subsequent catch operator as expected

Title

Icaro Temponi

07/30/2019, 7:26 PM

Hi everyone. So, I have the following function:

private fun barcodeFlow(): Flow<Barcode> = callbackFlow {
        val onNext: (t: Barcode) -> Unit = { barcode -> offer(barcode) }
        val onError: (t: Throwable) -> Unit = { ex -> throw ex }
        val disposable = binding
                .barcodeView
                .drawOverlay()
                .getObservable()
                .subscribe(onNext, onError)
        awaitClose { disposable.dispose() }
}

I just would like to know if the exception thrown inside the onError lambda will be propagated down the stream of this callbackFlow to be caught on a subsequent catch operator as expected

Dominaezzz

07/30/2019, 7:39 PM

No I don't think so.

The exception will be passed back into whatever calls

onError

Icaro Temponi

07/30/2019, 7:42 PM

hm, makes sense 🤔

View count: 2