is there a flow equivalent to http://reactivex.io/RxJava/3.x/javadoc/io/reactivex/rxjava3/core/Observable.html#buffer-long-java.util.concurrent.TimeUnit-int- ?

Hello, What would be the proper pattern to await multiple coroutines with different results? Is there an alternative to awaitAll? I don't like the necessary casting... I was thinking of using joinAll(...) and then

deferred.getCompleted()

, but

getCompleted()

is experimental.

viewModelScope.launch(Dispatchers.MAIN) {
    val deferred1 = async { operation1() }
    val deferred2 = async { operation2() }
    val deferred3 = async { operation3() }
    val deferred4 = async { operation4() }

    val results = awaitAll(deferred1, deferred2, deferred3, deferred4)
    
    val res1 = results[0] as? Type1
    val res2 = results[1] as? Type2
    val res3 = results[2] as? Type3
    val res4 = results[3] as? Type4
}

Hi all! I have a flow, which emits on database change. The query has a date parameter (today) which filters in the query on the data side. I have to detect the midnight and update the query to show correct data. Is there any good solutions to this?

Can I mark my methods as blocking to opt-in to the "inappropriate blocking call" warning somehow?

Are there any benefits LiveData still has over Flow? Or can Flow do anything LiveData can do?

if sharedFlow.tryEmit, returns false those that mean the value will not be emitted ?

I'm trying to track down the source of a

JobCancellationException

. My crash reporting tool isn't offering any useful information, and I don't quite understand this exception..

Hello everybody ✋ Im struggling to find a way how to to deal with order of my coroutines. I have a dashboard with multiple items. After I load the items I load additional info for each item on the dashboard. Meanwhile I can click any item on the dashboard to show item’s detail. Currently the detail starts to load only after all the additional info on the dashboard is loaded, which is something I would like to change. Ideally, I would like to start loading detail info before dashboard is finished. It is an android app and Im using

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

for all the rest api calls. Do you have any suggestions? Thank you :)

Hi All, I had a code looks like below

fun main() {
    viewModelScope.launch {
        doA()
        //call api with Flow as the return type
    }
}

suspend fun doA() {
    viewModelScope.async { doB() }.await()
}

suspend fun doB() {
    //Do logic here, eventually will call line below
    viewModelScope.launch { doC() }
}

suspend fun doC() {
    //call api with Flow as the return type
}

What I want to achieve from the code above is to wait for the

doA()

function to finished and then continue to call the API, but when I tried to run the code, the

doA()

is executed first but it continue to call the API rather than waiting for the

doA()

to finish. Is there's something wrong with the code? Yes, the code is quite bad, I still kind a new with the Coroutine, thanks for helping, really appreciate it

Hey! I have a weird question. I have a list of object and have to make a network call for every object, but they have to be made once a 10 seconds, and not faster. I was thinking, maybe I could somehow create a separate coroutine scope for each object and call a

delay

inside each call inside the

scope.launch

. Any suggestions would be very welcome, thank you!

Hi, I have a problem with switching from ReactorContext to CoroutineContext In coroutine-reactor docs I found an example:

val flow = flow<Unit> {
   println("Reactor context in Flow: " + coroutineContext[ReactorContext])
}

flow.asFlux()
    .subscriberContext { ctx -> ctx.put("answer", 42) }
    .subscribe() // Must print "Reactor context in Flow: Context{'answer'=42}" but print "Reactor context in Flow: null"

But with coroutine 1.4.0, 1.4.1 it prints Null. Am I miss something?

Is the a way to resume a flow on error?

fun main() = runBlocking<Unit> {
    emit()
        .onEach {
            if (it == 5) error("oof! some random error occurred.")
            println(it)
        }.catch { println("some error occurred") }
        .collect()
}


fun emit() = flow {
    for (i in 1..10) {
        emit(i)
    }
}

This will print "1 2 3 4" and stop. I'd like to print every number except number 5. (I'm doing this to simulate a random error, i know that in this specific case i could filter out the number 5).

Are there instructions available on building/publishing the KotlinX Coroutines library for the linuxArm32Hfp and linuxArm64 targets?

Was wondering if there some more idiomatic ways to accomplish :

fun scheduleRefresh() {
	job?.cancel()
	job = launch {
		while (isActive) {
			val newRefreshTime = getARefreshTime()
			if (newRefreshTime == 0) {
				break
			}
			delay(newRefreshTime)
			doSomething()
		}
	}
}

how can i retrieve MDC from a CoroutineContext in a exception handler ?  the code looks like so currently

private val exceptionHandler = CoroutineExceptionHandler { context, e -> GlobalScope.launch(MDCContext() + context) {

can i somehow do 

context.getCopyOfContextMap()

  ?

hey in the documentation SharedFlow.flowOn(...) has no affect if I understand it correctly . So how do you switch dispatchers when collecting from the shared Flow?

Hi guys, I'm trying to add unit tests to the StateFlow and it's awfully unpredictable (to me). Can you explain to me based on this example (the test passes like this):

@Test
    fun `stateIn self-contained example`() = runBlockingTest {

        suspend fun makeHeavyRequest(): String {
            return "heavy result"
        }

        val flow = flow<Unit> {} //in production this is a channel meant to refresh the Flow
            .onStart { emit(Unit) }
            .map { makeHeavyRequest() } //using mapLatest breaks 
            .flowOn(testDispatcher)
//            .onEach {  } //uncommenting this line also breaks
            .stateIn(GlobalScope, SharingStarted.WhileSubscribed(), "init state")

        val results = mutableListOf<String>()
        val job = launch {
            flow.collect { results.add(it) }
        }

        assertEquals("heavy result", results[0])

        job.cancel()

    }

1. Why the test breaks (I get

init state

instead of

heavy result

) when I use

mapLatest

instead of

map

? 2. Why the test breaks (I get

init state

instead of

heavy result

) when I uncomment the

onEach

below the

flowOn

? 3. Bonus question: why don't I get both, the

init state

AND the

heavy result

?

Hello folks, is there any way to transform flow into state flow like an extension function or something else?

another questions about MDC and coroutineContext.. can i update a ThreadContextElement from inside the coroutine ? issue is this code seems natural

suspend fun doStuff() = withContext(MDCContext() + exceptionHandler()) {
   MDC.put("flow", "do-stuff")
   <http://logger.info|logger.info> { "doing stuff" } // mdc is fine here

   someSuspendFun()

   <http://logger.info|logger.info> { "done doing stuff" } // mdc was reset to the state of MDCContext ie. is empty
}

ideally what i would like to be able to do is update the MDCContext or update the

val contextMap: MDCContextMap

after i modify the MDC ( i tried, it fails in different ways ) or use a mechanism where the MDC is stored before doing a suspend call, so any changes are actually picked up TL;DR currently it leads to rather unnatural looking code and additional nesting or errors that are really hard to find

Hello there! Is there a way to check if a

Flow<T>

with

debounce(...)

operator applied to it has currently a pending value that will be emitted after debounce delay? I have a specific problem with the following use case on Android. Inside a

ViewModel

, I am using a

Channel<String>

to organize pipeline for sending text changes to some remote backend: this channel is consumed as

Flow<String>

, where on each emission the value is sent (with delay resulting from

debounce()

operator) to the backend. When this

ViewModel

is to be destroyed and

onCleared()

is called, I cancel that same

Channel<String>

. But there is an expected but still unpleasant side-effect, which I need to eliminate: the last value is not guaranteed to be sent if there is a pending emission due to debounce delay, which gets cancelled when

channel.cancel()

is called. Any suggestions will be very appreciated!

is kotlin.coroutines diprycted yet, why i am asking this is kotlinx.coroutines are used rather than kotlin.coroutines(from standard-lib)

Hi Everyone, Please help. I was trying to do:

2sec     2sec     2sec
------[A]------[B]------[C]------...----------------> InitailFlow
       \        |        | 
        \      drop      drop
         \
     5sec \    5sec        5sec
----------[1]---------[2]---------[3]-----|> AnotherFlow

result: [A1, A2, A3]

So I have

InitailFlow

which emits a short amount of time (2 seconds) which is then transformed to

AnotherFlow

which takes longer to finish (15 seconds in total)... I would like to drop the other incoming items emitted by the

InitialFlow

while

AnotherFlow

isn't finished... I've tried:

flow{
    delay(2000)
    emit("A")
    delay(2000)
    emit("B")
    delay(2000)
    emit("C")
}.buffer(0, BufferOverflow.DROP_LATEST)
    .onEach {
       println("Event for $it")
    }
    .flatMapConcat {
       flow {
           delay(5000)
           emit("${it}1")
           delay(5000)
           emit("${it}2")
           delay(5000)
           emit("${it}3")
        }
     }
     .onEach {
         println(it)
     }
     .launchIn(scope)

But for some reason this is the result:

Event for A
A1
A2
A3
Event for B
B1
B2
B3

It still process Event B for some reason even when I have a

.buffer(0, BufferOverflow.DROP_LATEST)

. Why does it still process Event B? Is there a way to do this? Thanks in advance.

if I use

runBlockingTest

, will it execute coroutines launched within it sequentially? (as if I called

join

on them)

What is the alternative using flow or channel for SingleLiveEvent in this situation:

val onSuccess = SingleLiveEvent<Unit>()

fun load() {
    // do something
    onSuccess.call()
}

How do I properly clean up resources when using

flow

? For example if I open an IO Stream in the

flow {}

builder, how do I clean it up? I can't use

onCompletion

, because I don't have the stream available there obviously. If I understand the docs correctly I am not supposed to use

try-finally

in the

flow {}

builder, correct?

Hey Kotlin Community! I am trying to write end to end testing where entry point is Android

ViewModel

and last point is

MockWebServer

from OkHTTP using

Coroutines

. In ViewModel I make a call using

viewModelScope.launch

. However when I try to run this code

@get:Rule
    var instantExecutorRule = InstantTaskExecutorRule()

    @get:Rule
    var mainCoroutineRule = MainCoroutineRule()

    @Test
    fun reloadExpense() = runBlocking {
        viewModel.reloadExpense()
        assertThat(viewModel.state.value, CoreMatchers.instanceOf(LCE.Content::class.java))
    }

It does not wait for API request and just finishes the test. The MainCoroutine Rule is

@ExperimentalCoroutinesApi
class MainCoroutineRule : TestWatcher(), TestCoroutineScope by TestCoroutineScope() {

    override fun starting(description: Description?) {
        super.starting(description)
        Dispatchers.setMain(this.coroutineContext[ContinuationInterceptor] as CoroutineDispatcher)
    }

    override fun finished(description: Description?) {
        super.finished(description)
        Dispatchers.resetMain()
    }
}

I have tried using

runBlockingTest

,

runBlocking

and different types of rules. Nothing seems to work. Thank you in advance!

When is it hoped that there'll be

1.4.1-native-mt

version of Kotlinx Coroutines available?

Hi folks, I’m working with coroutines and I’ve a question about how to implement a coroutine factory. My use case: I’m implementing an interface 

MetricEmitter

 where consumers can simply call 

emitMetric(observation)

 . Under the hood this 

MetricEmitter

 will spin up an actor responsible for ingesting those metric observations. From the external consumer of this 

MetricEmitter

 I just want to expose methods such as: • `metricEmitter.initialize()`: which will spin up the actor and ticker • `metricEmitter.emitMetric()`: which will ingest the observation • `metricEmitter.close()`: which will flush remaining metrics and tear down the actor My problem is that in the 

initialize()

 function, when I try creating the actor, it requires me to provide a 

CoroutineScope

. If I use 

coroutineScope { actor<Msg>{... }

 this code block will freeze the whole execution since it will complete when the actor terminates. That’s not what I want. I’ve managed to solve the issue by passing a coroutineScope as argument of 

metricEmitter.initialize(scope: CoroutineScope)

 but not sure if this is a smell/antipattern. Any idea how to solve/remodel the solution?

Hi. Does anybody know why mutex is much slowly compared to synchonization? In thread I will add small example: