kotlinlang #coroutines

which is generating an incompatibly old version of meta-data vs. say, `4.8`+ which I'm having to use to support Java 12. In the event, I can compile ok, but multi-platform modules which are dependent on coroutines fail when publishing their own metadata.

Shakil Karim

08/11/2019, 10:10 PM

Does this code can have any side effects i want to search parallelly from two source on text change ?

Untitled.kt

giso

08/12/2019, 9:05 AM

Hi there, recently I had need for some kind of a MutableLiveData, but easily integratable with Flows and without the Android dependency. I came up with this solution, but I wonder, if there is a simpler solution. Any ideas?

Unbenannt

Rainer Schlonvoigt

08/12/2019, 1:20 PM

is there a coroutines implementation for general JVM apps, e.g. a desktop application?

540grunkspin

08/12/2019, 1:58 PM

Hi oh lovely community! Is there something similar to a condition variable on coroutines?

Dominaezzz

08/12/2019, 4:34 PM

Flow.conflate()

allows emitter to never suspend but is there an operator to allow the collector to never suspend (too)?

reline

08/12/2019, 7:45 PM

Is there a way to only allow experimental coroutines for tests? My gradle-fu is not exemplary.

tasks.withType(org.jetbrains.kotlin.gradle.tasks.KotlinCompile).all {
    kotlinOptions {
        kotlinOptions.freeCompilerArgs += ["-Xuse-experimental=kotlin.Experimental"]
    }
}

zak.taccardi

08/12/2019, 9:28 PM

Are there any behavior changes between

1.2.1

and 1.3.0-RC2` that we need to be worried about? Especially for `@ExperimentalApi`s?

zak.taccardi

08/12/2019, 9:30 PM

For https://play.kotlinlang.org - can we import libraries like coroutines?

raniejade

08/13/2019, 4:42 AM

Is there any reason why suspend is not supported for delegated properties? (

suspend operator fun getValue/setValue

)

coder82

08/13/2019, 10:31 AM

I need to spawn 10 worker threads which will call blocking legacy services. I also need them to have a bounded queue from which they read from, but I don't want to go through the effort of specifying a fan out channel which they read from and then having to check the message type all the time (10 actors reading and responding pattern). Rather I'd create a custom ThreadPoolExecutor where I can make the queue bounded and submit tasks directly without having to deal with messages flowing through a channel. What do you reckon? The problem of the actor model is that I need to define a lot of data classes and check which message I get so that I know what to do in the worker, while submitting the task directly is more convenient. Unless I miss something or I get some hint here 😄

coder82

08/13/2019, 10:32 AM

Another idea which pops in my mind is to use the actor model, send methods and use reflection to invoke them inside the worker, but all in all, if wonder if my thoughts are sane and if I am taking the right path. I like more the idea of submitting a task rather than a message with a "type" which leads to an implementation called by the worker @gildor @elizarov

Dias

08/13/2019, 1:37 PM

I have an api that provides 2 ways of calling a method. One returns Future (plain java future) and another one accepts callback. Which way is better to convert it to suspendable function? 1. Wrap call into withContext(Distpatcher.IO) { future.get() } 2. Wrap callback with a suspendCoroutine{ } method I prefer second option, because it looks more idiomatic, but withContext looks cleaner for a person less familiar with coroutines

Dias

08/13/2019, 1:38 PM

any other differences between these options in this particular context? apart from using different dispatchers

Jan Skrasek

08/13/2019, 1:42 PM

I'm using these extensions to convert Flow from and to LiveData. After updating to coroutines 1.3 (RC2) this seems doesn't work as intented. Is it bug or am I missing something?

Untitled

ahulyk

08/13/2019, 2:09 PM

what is the future of Coroutines after release of Project Loom?

🔮 4

ursus

08/13/2019, 3:43 PM

naively, if coroutines transform into callbacks, then those do have reference to the outer class, so it should leak, right?

Seri

08/13/2019, 5:33 PM

How would y’all expose subscribing to a BroadcastChannel without allowing external code to submit to it?

molikuner

08/13/2019, 6:36 PM

Hey, I'm just reading the introduction to Channels. Immediately after reading the first example with a function which creates a Chanel, I asked myself why they use an extension function for the CoroutineContext instead of just marking the function a suspend function. Original example:

fun CoroutineScope.produceSquares(): ReceiveChannel<Int> = produce {
    for (x in 1..5) send(x * x)
}

What I would expect:

suspend fun produceSquares(): ReceiveChannel<Int> = produce {
    for (x in 1..5) send(x * x)
}

Later in the document you can find:

All functions that create coroutines are defined as extensions on CoroutineScope, so that we can rely on structured concurrency to make sure that we don't have lingering global coroutines in our application.

But doesn't the

suspend

keyword does exactly that? It makes the current CoroutineContext implicitly available to the called function?

voben

08/13/2019, 8:17 PM

I have a long running operation which I start from my

viewModelScope

. When the viewmodel gets destroyed before the long running operation finishes, the function println(“Hello World”) is still getting called. Wouldn’t the coroutine get cancelled and not print “Hello World” due to structured concurrency?

viewModelScope.launch {
        doLongRunningOperation()
}

suspend fun doLongRunningOperation() = withContext(Dispatchers.Default) {
        longRunningCalculation() // Not a suspending function
        println("Hello World")
    }

zjuhasz

08/14/2019, 12:55 AM

I’m trying to decide between a

Channel

Flow

based API for my library and I’m having a hard time. I’m looking for some info / opinions / suggestions. I posted on the forum with all the info: https://discuss.kotlinlang.org/t/deciding-between-flows-and-channels/13858

Tuan Kiet

08/14/2019, 7:45 AM

Is there anyway/api to convert a Flow to LiveData?

coder82

08/14/2019, 8:10 AM

For calling legacy services which will block on database calls, what's the best dispatcher? is it Dispatchers.IO?

👍 1

:yes: 4

winchester044

08/14/2019, 10:55 AM

Hi folks, I was wondering if there is a way to cancel/stop `runBlocking`(which has an associated context) once the parent job is cancelled and the lambda is yet to start execution on the underlying dispatcher? Consider example:

class Foo {
    val dispatcher// An android handler thread as a Coroutine Dispatcher
    val job // Parent Job
    val coroutineScope = job + dispatcher

    fun <T> run(block: () -> T) {
        // some prechecks
        return runBlocking(coroutinScope.coroutineContext) {
            // this runs on dispatcher
        }
    }

    fun shutdown() {
       // stop underlying runBlocking 
       job.cancel()
    }
}

Consider the handler thread is busy doing some other stuff and

runBlocking

is waiting for it to finish. Meanwhile

shutdown()

gets invoked. Is there a way to 'cancel/stop' this

runBlocking

call? (Without using thread interrupts, although once lambda has started execution cancellation becomes cooperative).

Eric Martori

08/14/2019, 11:55 AM

Would it be posible to include:

suspend fun <T> Flow<T>.first() = take(1).single()

In the flow API?

coder82

08/14/2019, 3:40 PM

where do I find the bext explanation about coroutine contexts? I still struggle to inderstand them, it's really a black hole for me

molikuner

08/14/2019, 6:49 PM

Does this make any sense, or has any benefit?

fun complete(): X {
    TODO("your implementation")
}
suspend fun suspendComplete(): X {
    return suspendCoroutine { continuation  ->
        val result = runCatching { complete() }
        continuation.resumeWith(result)
    }
}

🇳🇴 7

Marko Mitic

08/15/2019, 10:44 AM

Is there an easy way to call suspend functions from Java 7? I saw some useful APIs in kotlinx-coroutines-jdk8 but that's bit too high for Android

tjohnn

08/15/2019, 10:53 AM

Is there a way or an example of using Realm database with coroutines?

nickheitz

08/15/2019, 11:59 AM

Hi all. I had a play around with the ArrayBroadcastChannel today, and I'm not understanding something about cooperative cancellation.

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.CoroutineStart
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.channels.BroadcastChannel
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.broadcast
import kotlinx.coroutines.delay
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import org.junit.Test
import kotlin.coroutines.CoroutineContext

class BroadcastChannelTest : CoroutineScope {
    override val coroutineContext: CoroutineContext
        get() = <http://Dispatchers.IO|Dispatchers.IO> + Job()

    init {
        System.setProperty("kotlinx.coroutines.debug", "")
    }

    @Test
    fun broadcast() {
        val jobs = mutableListOf<Job>()
        val channel = Channel<Int>().broadcast(5, CoroutineStart.DEFAULT)
        runBlocking {
            jobs.add(createConsumer(channel))
            jobs.add(createConsumer(channel, 500, 3))
            delay(250)

            (0..10).forEach {
                channel.send(it)
                delay(200)
            }

            channel.send(-1)

            jobs.forEach {
                it.join()
            }
        }

    }

    private fun createConsumer(channel: BroadcastChannel<Int>, delay: Int = 0, failOn: Int = 11): Job {
        return launch {
            try {
                if (delay > 0) delay(delay.toLong())

                println("creating consumer $delay")
                val sub = channel.openSubscription()
                for (msg in sub) {
                    if (isActive) {
                        println("$msg on ${Thread.currentThread().name}")

                        if (msg == -1) break

                        if (msg == failOn) {
                            sub.cancel()
                            throw Exception("Failing now...")
                        }
                    } else {
                        println("No longer active.")
                    }
                }
            } finally {
                println("Consumer done")
            }
        }
    }
}

in short, I want one of the channels to error out after 3 messages. GIven that all of these coroutines are (I think) running in the CoroutineScope of the class itself, the Job() at the top should lead to cooperative cancellation on an exception. However, the non erroring listener continues to completion. Any ideas?

Title

nickheitz

08/15/2019, 11:59 AM

Hi all. I had a play around with the ArrayBroadcastChannel today, and I'm not understanding something about cooperative cancellation.

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.CoroutineStart
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.channels.BroadcastChannel
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.broadcast
import kotlinx.coroutines.delay
import kotlinx.coroutines.isActive
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import org.junit.Test
import kotlin.coroutines.CoroutineContext

class BroadcastChannelTest : CoroutineScope {
    override val coroutineContext: CoroutineContext
        get() = <http://Dispatchers.IO|Dispatchers.IO> + Job()

    init {
        System.setProperty("kotlinx.coroutines.debug", "")
    }

    @Test
    fun broadcast() {
        val jobs = mutableListOf<Job>()
        val channel = Channel<Int>().broadcast(5, CoroutineStart.DEFAULT)
        runBlocking {
            jobs.add(createConsumer(channel))
            jobs.add(createConsumer(channel, 500, 3))
            delay(250)

            (0..10).forEach {
                channel.send(it)
                delay(200)
            }

            channel.send(-1)

            jobs.forEach {
                it.join()
            }
        }

    }

    private fun createConsumer(channel: BroadcastChannel<Int>, delay: Int = 0, failOn: Int = 11): Job {
        return launch {
            try {
                if (delay > 0) delay(delay.toLong())

                println("creating consumer $delay")
                val sub = channel.openSubscription()
                for (msg in sub) {
                    if (isActive) {
                        println("$msg on ${Thread.currentThread().name}")

                        if (msg == -1) break

                        if (msg == failOn) {
                            sub.cancel()
                            throw Exception("Failing now...")
                        }
                    } else {
                        println("No longer active.")
                    }
                }
            } finally {
                println("Consumer done")
            }
        }
    }
}

kingsley

08/15/2019, 12:08 PM

Quickly glancing over this. It appears that you are setting up the

coroutineContext

wrong, which might be a source of the issue You should have a:

private val job = Job()

// and then
override val coroutineContext: CoroutineContext
  get() = <http://Dispatchers.IO|Dispatchers.IO> + job

// OR
override val coroutineContext = <http://Dispatchers.IO|Dispatchers.IO> + Job()

Basically, as it is now, any coroutine you launch will be started in the context of a new parent job, since a new

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

is being passed down each time

nickheitz

08/15/2019, 7:54 PM

That makes perfect sense. I obviously misunderstood an example somewhere demonstrating how to extend CoroutineScope. Thanks.

👍 1

View count: 29