Since DataFlow and share operator seems far away what would the current idiomatic way to have "something" that would support multiple senders to it and conflate to keep the last data and multiple receiver that comes and leaves, but when comes instantly get the last pushed value? (Pushed before it register)

Has anyone tested coroutines with graalvm? (@elizarov? In particular the native images. I'm getting errors which I can't resolve and sadly I'm going to have to change my code to use reactor.

There is so much demand for multicasting/sharing support in Flow. Libraries like dropbox/store have their own inline implementations. There are tons of gists of implementations posted on the various coroutine issues. I haven’t seen any libraries published on maven for this though. Has anyone factored any of these out into a library? We’re starting to consider moving from RxJava to Flow in some of our larger codebases and would like to use existing libraries for stuff like this instead of copy/pasting or writing our own implementation of stuff.

I wrote an intro for flow, your feedback is more than welcome https://medium.com/@MohamedISoliman/flow-an-intro-for-an-rxjava-user-1b5b6eb21790

If I am using a library which offers both a synchronous and an asynchronous API to do IO stuff, I am ensure about the pros/cons of using either way: 1. withContext(Dispatchers.IO) { lib.syncAPI() } 2. withContext(Dispatchers.IO) { suspendCancellableCoroutine{ lib*.asyncAPI( object:Listener{ fun onfailure(){} fun onSuccess() ...})}* Option 1 wins on simplicity and I control on my side the thread pool, anything else?

Noticing a performance degradation in a test using coroutines version 1.3.4 vs/ 1.3.3. The test uses Dispatchers.Default and does something like the following: • Launch a coroutine that collects a

channel.consumeAsFlow()

• Send a message to the channel that results in another coroutine that sends another message to the channel that results in decrementing a

CountDownLatch

• In the main thread, await the

CountDownLatch

With 1.3.3, the await took ~180ms. With 1.3.4 its taking ~1.4 seconds. Not sure how this affects the production code yet, but is there a change in 1.3.4 that would explain this?

Hi. I’m not sure if it is a bug or I treat the docs too literally so please advise. The doc for

Flow.launchIn()

says it is “a shorthand for `scope.launch { flow.collect() }`” but when I compare behaviors in unit tests it is a little different. Having a simple flow:

private val integersFlow = flow {
        var i = 1
        while(true) {
            delay(100)
            emit(i++)
        }
    }

and using test scope and dispatcher:

private val observingScopeJob1 = SupervisorJob()
private val observingScope1 = CoroutineScope(observingScopeJob1)
private val testDispatcher = TestCoroutineDispatcher()

When I do the following:

@Test
    fun `should receive 10 integers - collect`() = runBlocking {
        var observedIntegers = 0

        observingScope1.launch(testDispatcher) {
            integersFlow.collect {
                observedIntegers++
            }
        }

        observedIntegers.shouldEqual(0)
        testDispatcher.advanceTimeBy(1000)
        observedIntegers.shouldEqual(10)
        Unit
    }

the test passes ✅ but when I convert it into RxJava-style like this:

@Test
    fun `should receive 10 integers - launchIn`() = runBlocking {
        var observedIntegers = 0

        integersFlow
            .onEach { observedIntegers++ }
            .flowOn(testDispatcher)
            .launchIn(observingScope1)

        observedIntegers.shouldEqual(0)
        testDispatcher.advanceTimeBy(1000)
        observedIntegers.shouldEqual(10)
        Unit
    }

the test fails ❎ unless I add an additional

delay(100)

before

testDispatcher.advanceTimeBy(1000)

which seems very weird to me. Like… why is that needed to delay the main thread in unit test to allow the collecting coroutine to be launched at all?

How to create a

CoroutineScope

that is a child of another ? What I tried:

val parentScope = ...
val childScope = parentScope + SupervisorJob()

parentScope.cancel()
parentScope.isActive // false
childScope.isActive // true (!!)

It seems that cancelling the parent does not cancel its children. I guess I'm missing something here...

What is the proper way make a call back that is called multiple times into a flow? similar to this but that gets called ony once: https://stackoverflow.com/questions/48552925/existing-3-function-callback-to-kotlin-coroutines/48562175#48562175

omg, i thought i knew, I thought knew what concurrency hell was, i thought i could outsmart it. I failed. Now here i am. At least i have kotlinx-coroutines on my classpath?

textField.debouncedAsTyping(): Flow<Change<String, String>>

seemed like such a good idea, but back pressure/cancellation is kicking my butt.

Hello. I am using asCompletableFuture as a bridge between coroutines and futures for code where I want to benefit from ktor client being able to run many active connections. In a small example, when errors occur the completablefuture never completes, not even with exception. How do I let the exceptions that occur bubble up correctly? https://github.com/rrva/ktor-client-connection-leak/blob/master/src/main/kotlin/se/rrva/Client.kt I am expecting ”Done with all” to be printed on each run

Is there any nice class that can be used as a replacement for android LocalBroadcastManager? like a conflated channel but without any “memory” not even the last sent message?

message has been deleted

if I don't await on a

GlobalScope.async

does it get GC'ed allow the stuff being referenced in it get GC'ed?

Hi! I have a class which looks like this

class MyQueueProcessor: CoroutineScope by CouroutineScope(<http://Dispatchers.IO|Dispatchers.IO>) {
    suspend fun start() {
        pollMessages()
        processors.addAll(
            (1..5).map {
                launchMessageProcessor()
            }
        )
    }
    
    suspend fun pollMessages() = launch {
       while(true) {
          val messages = //... get messages from a queue
          internalChannel.send(messages)
       }
    }
    
    suspend fun launchMessageProcessor() = launch {
        for (messages in internalChannel) {
            // process each 'messages'
        }
    }

}

but for both pollMessages and launchMessageProcessor I get "Ambiguous coroutineContext due to CoroutineScope receiver of suspend function". How should I interpret this? Is it because suspend fun start() is public and you inherit the coroutineContext from the caller? Since I extend CoroutineScope, the launch and async stuff I use should run on that scope with that dispatcher (IO) right?

Cross-point from the #ktor channel (https://kotlinlang.slack.com/archives/C0A974TJ9/p1584027549237200)

and no matter what I do, the content variable at the bottom, whether its async, launch, runBlocking, etc. none of them are able to be resolved

https://play.kotlinlang.org/hands-on/Introduction%20to%20Coroutines%20and%20Channels/04_Suspend “Coroutines are computations that run on top of threads and can be suspended. By saying “suspended”, we mean that the corresponding computation can be paused, removed from the thread, and stored in memory. When the computation is ready to be continued, it gets returned to a thread (but not necessarily to the same one).” Who works on the computation while it’s suspended? Another thread? Kotlin std library thread? I shouldn’t care?? 🤔 What am I missing on how coroutines work?

is there any throttle operator for flow ?

I’m trying to create a flow from an EditText change in Android, I’m using

flow {…}

builder, So I’m creating a watcher inside it here is the code, my question is how I clear inner listeners when flow canceled ?

fun EditText.textChangesFlow(): Flow<TextViewTextChangeEvent> {
    flow<TextViewTextChangeEvent> {
        val watcher = object : TextWatcher {
            override fun afterTextChanged(s: Editable?) {
                emit(TextViewTextChangeEvent(s.toString()))
            }

            override fun beforeTextChanged(s: CharSequence?, start: Int, count: Int, after: Int) {
            }

            override fun onTextChanged(s: CharSequence?, start: Int, before: Int, count: Int) {
            }
        }
        addTextChangedListener(watcher)
         //how to call this when flow is canceled or finished
        //removeTextChangedListener(watcher)

    }
}

Is there a way to use synchronized() with coroutines? I have a need for several short running coroutines/threads to synchronize access to another function that gets data. Synchronization is needed since the first call is over http then cached locally, to prevent redundant web requests.

shouldn’t be gone when converting it to Flow and use

launchIn()

?

https://play.kotlinlang.org/hands-on/Introduction%20to%20Coroutines%20and%20Channels/05_Concurrency

Run the code and check the log; we can see that all the coroutines still run on the main UI thread. We haven’t yet employed multithreading in any way, but already we have the benefits of running coroutines concurrently!

It’s very easy for us to change this code to run “contributors” coroutines on different threads from the common thread pool. Specify 

Dispatchers.Default

 as the context argument for the 

async

 function:

async(Dispatchers.Default) { ... }```

I’m sort of confused. If I don’t specify a dispatcher in the argument for

async

function, isn’t the default dispatcher

Dispatchers.Default

? Yet, I did see the logs change from

147 [AWT-EventQueue-0] INFO  Contributors - Clearing result
2251 [AWT-EventQueue-0 @coroutine#1] INFO  Contributors - kotlin: loaded 54 repos
2678 [AWT-EventQueue-0 @coroutine#7] INFO  Contributors - kotlin-benchmarks: loaded 7 contributors
...

to

164 [AWT-EventQueue-0] INFO  Contributors - Clearing result
2172 [AWT-EventQueue-0 @coroutine#1] INFO  Contributors - kotlin: loaded 54 repos
2527 [DefaultDispatcher-worker-5 @coroutine#10] INFO  Contributors - anko-example: loaded 2 contributors
2621 [DefaultDispatcher-worker-5 @coroutine#9] INFO  Contributors - kotlinx.html: loaded 15 contributors
...

But when I read the comments in the source code of

Builders.common.kt

, it was stated like this:

* Coroutine context is inherited from a [CoroutineScope], additional context elements can be specified with [context] argument.
 * If the context does not have any dispatcher nor any other [ContinuationInterceptor], then [Dispatchers.Default] is used.
 * The parent job is inherited from a [CoroutineScope] as well, but it can also be overridden
 * with corresponding [context] element.

and in

Dispatchers.kt

, it was stated:

/**
     * The default [CoroutineDispatcher] that is used by all standard builders like
     * [launch][CoroutineScope.launch], [async][CoroutineScope.async], etc
     * if no dispatcher nor any other [ContinuationInterceptor] is specified in their context.
     *
     * It is backed by a shared pool of threads on JVM. By default, the maximal level of parallelism used
     * by this dispatcher is equal to the number of CPU cores, but is at least two.
     * Level of parallelism X guarantees that no more than X tasks can be executed in this dispatcher in parallel.
     */
    @JvmStatic
    public actual val Default: CoroutineDispatcher = createDefaultDispatcher()

Could someone explain what I’m missing?

This just came across my mind: Is there a tool to visualize the coroutine scope parent and child structure in the project?

ReactiveX has a nice visualizations for describing operators called marble diagrams. I think kotlin flow needs them too. So I made this site, in Kotlin JS. https://flowmarbles.com/

Is it safe to modify local mutable, non thread safe data structures in a coroutine that jumps between threads? e.g.

fun main() = runBlocking<Unit> {
    val a = newSingleThreadContext("a")
    val b = newSingleThreadContext("b")
    launch {
        val set = mutableSetOf<Int>()
        repeat(100000) {
            withContext(a) {
                set.add(it)
            }
            withContext(b) {
                set.remove(it)
            }
        }
        println(set.size)
    }
}

I'd have thought it wouldn't be since set is being modified on multiple threads and isn't thread safe. On the other hand I can't see anything in the docs warning me not to do this and I haven't managed to produce any errors by doing seemingly unsafe operations.

I'm trying to test a suspend method which first tries to reach an API, but if that throws an exception, would fall back to a local method. However, it doesn't appear the catch in the suspend method is being executed

For people interested in what’s doing the Kotlin compiler under the hood 👌 Blog post about this coming along next week

is there any flow operator that filters unique values ?