i want to use it like flowOf<Int>().onEmpty{ emit(0) }, sth like this

If I use

asFlow()

on a closed

ConflatedBroadcastChannel

will it receive the latest element sent before it was closed, or will the flow be empty?

In general, even if not closed, it seems like it misses the elements sent before the

asFlow()

was run on it...

Hey guys, I want to send stream of data from one class to the other (adapter/viewholder to activity). Previously I'd use RxJava's Subject for it, what would be recommended way of achieving this with flow? I see how I could do it with channels but it's hard for me to wrap my head around the flow for such use cases Should I use something like this?

📣 📣 📣 kotlinx.coroutines 1.3.5 is here! •

Flow.firstOrNull

operators •

java.time

and

kotlin.time.Duration

adapters • More performant and configurable

DebugProbes

for usages in production environment •

kotlin-coroutines-jdk9

module for integration with

java.util.concurrent.Flow

• Various bug fixes and improvements Full changelog: https://github.com/Kotlin/kotlinx.coroutines/releases/tag/1.3.5

what is the proper way of transforming a flow<List<T>> to a flow<T> ?

This message was deleted.

Hi, i'm making a class that check audio input level in Android, for do that I must call one method each 100ms in a separate thread and notify the value with a lambda

block

, for the moment I have this (old school hava style) and I guess I can do better with coroutines.

mustStopThread

is a boolean that is false when class is created and true when I must stop the check (when I quit the activity). What I can do better ?

init {
    Thread(Runnable {
        while (!mustStopThread) {
            val lastMaxSoundLevel: Int = mediaRecorder.maxAmplitude
            block.invoke(lastMaxSoundLevel)
            Thread.sleep(100)
        }
    }).start()
}

Given the following viewmodel:

class ViewModel {
    val selectedColor: Flow<Color> = ...
    fun onNewColorSelect(newColor: Color) {
        ...
    }
}

How can the

onNewColorSelect

method update the

selectedColor

flow? With Android, you would use

MutableLiveData

. How would you solve this problem with Flow?

Hello! Is there any possibility to run test in supend test function?

class Test{
  @Test
  suspend test() {
    // test logic with suspend calls
  }
}

In my ViewModel, when an event listener is triggered:

onUserSelect(userId: Int)

I want it to offer an event to a broadcastchannel named

userSelectEvents: BroadCastChannel<UserSelectEvent>

. The flow

selectedUsers

should depend on this, so I'm probably looking to some kind of building a reducer. This is my idea, is it good?

Hi folks, Can any of you show me what's the recomended way of scheduling a (async) task to happen in X minutes from current time? If possible in such a way that I can control executors. Thanks in advance.

hi, if I'm given a

scope

called

bigScope

how can I create a scope

smallScope

that will be canceled when

bigScope

is cancelled?

Given the following:

private val _selectedUserIds = ConflatedBroadcastChannel<List<Int>>(emptyList())
private val selectedUserIds: Flow<List<Int>> = _selectedUserIds.asFlow()

A

onSaveButtonClick()

method inside the class should send a network call with the value of selectedUserIds Should I get it's value by using the

.value

of the

ConflatedBroadcastChannel

or should I launch a coroutine and call

.collect()

on the

Flow

? What's the difference?

fun main() = runBlocking { // this: CoroutineScope

    coroutineScope { // Creates a coroutine scope
        println(this)
        println(this@runBlocking)
...

and the printed result is:

"coroutine#1":ScopeCoroutine{Active}@64a294a6
"coroutine#1":BlockingCoroutine{Active}@7e0b37bc

Why do they have the same name “coroutine#1”? If they are different coroutines, shouldn’t they have different labels? Am I missing something?

I’d like to ask if I understood this correct: https://kotlinlang.org/docs/reference/coroutines/cancellation-and-timeouts.html#cancellation-is-cooperative

import kotlinx.coroutines.*

fun main() = runBlocking {
    val startTime = System.currentTimeMillis()
    val job = launch(Dispatchers.Default) {
        var nextPrintTime = startTime
        var i = 0
        while (i < 5) { // computation loop, just wastes CPU
            // print a message twice a second
            if (System.currentTimeMillis() >= nextPrintTime) {
                println("job: I'm sleeping ${i++} ...")
                nextPrintTime += 500L
            }
        }
    }
    delay(1300L) // delay a bit
    println("main: I'm tired of waiting!")
    job.cancelAndJoin() // cancels the job and waits for its completion
    println("main: Now I can quit.")
}

The output is:

[DefaultDispatcher-worker-1 @coroutine#2] job: I'm sleeping 0 ...
[DefaultDispatcher-worker-1 @coroutine#2] job: I'm sleeping 1 ...
[DefaultDispatcher-worker-1 @coroutine#2] job: I'm sleeping 2 ...
[main @coroutine#1] main: I'm tired of waiting!
[DefaultDispatcher-worker-1 @coroutine#2] job: I'm sleeping 3 ...
[DefaultDispatcher-worker-1 @coroutine#2] job: I'm sleeping 4 ...
[main @coroutine#1] main: Now I can quit.

From my understanding, the

job

did not get cancelled despite the

job.cancelAndJoin()

call, because there is no suspending functions (e.g.

delay

) which are cancellable (checks for cancellation cooperatively). Therefore the job completes after five iterations, then the

job.join()

returns, and then at last “main: Now I can quit.” is printed. Is this correct? P.S. Does “every” suspending function always check for cancellation?

Simple question... will corotuines launched in a dispatcher that's backed by a thread pool implicitly be parallelized? Eg.

suspend fun foo() = withContext(Dispatchers.Default) {
    for (1..100) {
        async { ... }
    }
}

Will that be bound to a single thread or will is try to spread the workload?

Any ideas why this test fails https://github.com/jshvarts/FlowChannels101/blob/master/app/src/test/java/com/example/coroutines/views/UserDetailViewModelTest.kt#L100 with

Wanted but not invoked observer.onChanged...no interactions with this mock

. Thanks in advance!

Is there any way to throtle the messages to a channel? Adding delay to the messages given a param of the msg. Linke a priority queue or delayed queue? I though to just run a coroutine and then delay but it seems no good for a fan out scenario.

I was reading the documentation https://kotlinlang.org/docs/reference/coroutines/select-expression.html, and when I tried to execute the code

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.selects.*

suspend fun selectAorB(a: ReceiveChannel<String>, b: ReceiveChannel<String>): String =
    select<String> {
        a.onReceiveOrNull { value -> 
            if (value == null) 
                "Channel 'a' is closed" 
            else 
                "a -> '$value'"
        }
        b.onReceiveOrNull { value -> 
            if (value == null) 
                "Channel 'b' is closed"
            else    
                "b -> '$value'"
        }
    }
    
fun main() = runBlocking<Unit> {
    val a = produce<String> {
        repeat(4) { send("Hello $it") }
    }
    val b = produce<String> {
        repeat(4) { send("World $it") }
    }
    repeat(8) { // print first eight results
        println(selectAorB(a, b))
    }
    coroutineContext.cancelChildren()        
}

The compiler gave warning that

onReceiveOrNull

was deprecated:

'onReceiveOrNull: SelectClause1<E?>' is deprecated. Deprecated in favor of onReceiveOrClosed and onReceiveOrNull extension

I guess this is not yet applied in the guide from the documentation. May I ask how I should change the code?

suspend fun selectAorB(a: ReceiveChannel<String>, b: ReceiveChannel<String>): String =
    select<String> {
        a.onReceiveOrClosed {  
            if (it.isClosed)
                "Channel 'a' is closed"
            else
                "a -> '${it.value}'"
        }
        b.onReceiveOrClosed { 
            if (it.isClosed)
                "Channel 'b' is closed"
            else
                "b -> '${it.value}'"
        }
    }

Would this be the preferred way for now?

when i was using threads used to work on concurrentlinkedqueue and others is there any equivalent for it in kotlin when using coroutines?

Can we expect coroutines in K/N in 1.4? https://blog.jetbrains.com/kotlin/2020/03/kotlin-1-4-m1-released/

My team and I found some issues around using coroutines. Just wanted to surface them here. If you are experiencing the same issues, please up-vote. https://youtrack.jetbrains.com/issue/KT-36660 https://youtrack.jetbrains.com/issue/KT-37456

how do I cancel the children a coroutinescope and not the scope itself?

is there a way to wait for not yet started coroutine? I want to achieve something similar to

yield

but with the coroutine that'll be started in the future

Hiya, on my splash screen (android) I want to fetch some remote config in the background, then only move on from the screen once the fetch has completed (either successfully or via exception, I don’t care) I have a private field that should launch immediately upon the class being instantiated

private val remoteConfigFetch = lifecycleScope.launch { FirebaseRemoteConfig.init() }

Then further on before the splash screen finally moves on I’m calling join

private fun continueToApp() = runBlocking {
    remoteConfigFetch.join()
    goToMainActivity()
}

Is this the correct approach? Initially I thought about using async but figured because I don’t care about whether it fails or errors and don’t care about the result it was better to use launch?

Hello, is

Channel

still preferred over

Flow

for use cases that involve multiple emitters/listeners for a data stream? From what I've read online, Flow is meant to be a safer alternative to Channel, but seems best used for one to one communication?

I’m sure this should be obvious, but given an existing synchronous API that expects a blocking

InputStream

or

Reader

, what’s the correct way to wrap that such that I can call it as a suspendable function and map some sort of asynchronous communication such as a

Channel

into the given

InputStream

?

In a multi-module project, should a kotlin module that exposes Flow as its public api, specify its coroutines dependency as

api

instead of

implementation

? Also if so, does this also mean for target dependencies (e.g. commonMain:

api(coroutines-core-common)

and jvmMain:

api(coroutines-core)

)