Is there a way to compare changes between emissions? Example:

flowOf(1,2,5)
  .operation { last, current -> current - last }
  .onEach { println(it) }
  .collect()

Result: 1,3

Would this work as a system that allows callers to either send a signal ("flush"), or await the next signal that arrives? The calls come from different

scope.launch

blocks that are running in parallell.

@JvmInline
internal value class Queue(
    private val trigger: MutableSharedFlow<Unit> = MutableSharedFlow(extraBufferCapacity = 8),
) {

    suspend fun flush() {
        trigger.emit(Unit)
    }

    suspend fun await() {
        trigger.first()
    }
}

Hey folks, I am currently doing some unit testing and I have been wondering if there is a way to see which coroutines are currently scheduled for execution?

I have

val resultsFlow = MutableSharedFlow<HdmiResult>()
        val results: Flow<List<HdmiResult>> = resultsFlow
            .runningFold(listOf()) { list: List<HdmiResult>, v: HdmiResult -> list + listOf(v) }

But somehow the first ~ 10 items are not present in the

results

flow

Hello, I'm wondering if it's OK to do this kind of code. If I understood correctly, Coroutines use exceptions internaly to "communicate" between child and parent about their state (and propagate cancellation via

CancellationException

). Wouldn't using a big

catch (exception: Exception)

around a suspending function

api.getFooResponse()

defeat this purpose and prevent any coroutine calling

getFoo()

to be cancelled correctly ? Wouldn't the

CancellationException

be "swallowed" ? Should we re-throw it ?

suspend fun getFoo(): Foo? = try {
        val fooResponse = api.getFooResponse()
        Foo(bar = fooResponse.bar)
    } catch (exception: Exception) {
        exception.printStackTrace()
        null
    }

what will happen with Dispatchers.IO now that JDK19 has virtual threads? Will it continue to exist as is or will there be adjustments to make use of virtual threads somehow?

I am trying to get data from a room database via a Flow. I want to make access to this data instant so I am thinking of using a StateFlow to access the state from memory via StateFlow.value. So on app start I want to be able to access the StateFlow.value to get the initial value from the database maybe via a blocking suspend function. What does this look like and any recommendations?

If I was making a networking library how would I best interface with kotlinx coroutines? Right now I'm not putting in any special effort (in fact the library doesn't even depend on kotlinx coroutines). One of the problems is that you have isActive and isOpen and this kind of strange relationship between errors and connection status. You end up having to have a whole bunch of try catch, close cancel. Would it be "more correct" to remove isOpen/close and instead cancel the job if the socket reaches end of stream? (if this is more correct how do you even fully replace close in practice, seems tricky) Largely unrelated but what's going on with the warning when calling a blocking method in a suspending scope? It seems very hard to satisfy that error if you have your own dispatcher or even know the dispatcher is IO. Do most people simply disable the warning?

Emit from a catch block is bad practice ? Android developer guide says it not https://developer.android.com/kotlin/flow

Hello, I'm trying to make a dynamic `sample`ing of a Flow but it feels "ugly". My needs : 1/ A function

updateFoo(foo: String?)

can be called any number of time, from multiple different coroutines 2/ A Flow should be available to emit the "last value that went through `updateFoo`", but should not emit quicker than every 100ms 3/ The flow should not re-emit similar values 4/ This is the bad part, the Flow should emit null values immediately My approach :

companion object {
    private val SAMPLE_DURATION = 100.milliseconds
}

private val fooMutableStateFlow = MutableStateFlow<String?>(null)

val fooFlow: Flow<String?> = fooMutableStateFlow
    .sample(SAMPLE_DURATION)
    .onStart { emit(fooMutableStateFlow.value) }
    .distinctUntilChanged() // Avoid double emission of null value with .onStart...

fun updateFoo(foo: String?) {
    fooMutableStateFlow.value = foo
}

I'd love an operator like debounce so I could do something like that :

val fooFlow: Flow<String?> = fooMutableStateFlow.sample {
    if (it == null) {
        Duration.ZERO
    } else {
        Toto.SAMPLE_DURATION
    }
}

Hey! In my web service I have a requirement to return a response on the request thread pretty fast. So I am using

GlobalScope.launch

to send fire-and-forget HTTP requests to other services. I don't like the

GlobalScope.launch

idea and trying to thinking of other solutions. If I would have been in Clojure I think I would have created a global go-loop to handle the fire-and-forget requests and have a channel to communicate with that go-loop. Could that been done in Kotlin? Is that a good approach for concurrent fire-and-forget stuff or are there better ways of doing it?

Hi, the

combine

operator only emits once every flow has an emission. in my case i want to combine flows and it might that one flow would not emit at all - in that case i want to have the transformation receive a null value for not yet existing values of those streams.

combineAlways(flow1, flow2) { v1, v2 (null as not yet an emission) ->
...
}

is there a way to achieve that?

Hi. We know that

channelFlow

can do what

flow

can't e.g. • Retrieve value from a concurrently run coroutine • Can use a non-suspending way i.e. `trySend` to send data It looks like it is more powerful than

flow

. I wonder • if there's anything where

flow

can do but not in

channelFlow

, or • if there's anything

flow

is preferred over

channelFlow

(e.g. in terms of efficiency or performance?)? The reason I ask is, I want to see if we can just use

channelFlow

for everything instead, or what's the scenario we should use

flow

instead? I may have missed out something fundamental?

Is there a

retryWhen

alternative for regular suspend functions? Just like https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/retry-when.html

Could I somehow construct a flow that uses ktor? The API I'm using returns several items per page

I've got incoming data where a reader can connect (via bluetooth), where cards can be put on, where data can be read from. Currently, I have that modeled as nested `Flow`s, but the cancelation is currently a bit... suboptimal, and advancing to the next reader isn't as clean as I hoped it would be. Even using

collectLatest

isn't really working that well, I'm not sure how to declare an element invalid - or is

collectLatest

the correct way here? Or should I be using a different abstraction that

Flow

if a previous element becomes dead/invalid?

I think a simpler question would be - if I have a

Flow<Resource>

, how can I declare a

Resource

to be unavailable? As an additional information, only the newest

Resource

can be available.

Is it possible to cancel a sharedIn cold flow?

If I have this:

runBlocking {
  SMR()
  println("test")
}

it hangs when SMR is like:

suspend fun SMR() = withContext(dispatcher) {
  launch { ... }
}

But not if it's like:

suspend fun CoroutineScope.SMR() {
  launch { ... }
}

From the look of things withContext should just grab the old context, merge with new context and dispatch if the dispatcher changed.

Hi, why

Delay

use millis time(scheduleResumeAfterDelay) instead of more accurate

Duration

? For now there is no way to pass Duration to UI dispatcher which is accurate, i.e don't make errors with rounding 😞

So i have a to implement a Java interface to do some logic. My implementation uses corutines, and my code triggering the library uses corutine. How can i pass the scope past the interface... Now i have runBlocking in the interface but it brakes the context and the tracing

I receive some events from the backend. I want to internally pass the events as a

Flow

or through a

Channel

. After the last event I want to send a "complete" signal. Is there something like this in

Flow

?

Just a though. shouldn't the

SharingStarted.whileSubscribed*()

API be called

SharingStarted.whileCollected*()

instead? since

collect {}

is used on it, not

subscribe {}

?

When would I use

combine

vs

combineTransform

?

TIL

ContinuationInterceptor

and

CoroutineDispatcher

actually have separate “polymorphic” keys. This seems to mean it’s possible to create a coroutine context with a dispatcher A and an interceptor B that wraps A, where if you ask it for the interceptor you will get B but if you ask specifically for the dispatcher you’ll get A. Has anyone written a blog or anything about gotchas with combinations of this, or done it before and hit snags? I’m seeing some weird behavior in this situation when B tries to wrap a continuation and then ask the wrapped dispatcher to intercept.

StateFlow with Room not updating immediately after save. I assume this is me having a misunderstanding of coroutines. 🧵

The Android coroutine testing docs say that if you create your own

TestScope

(e.g. to inject into the class under test), then you must call

runTest

on that scope, because there can only be one

TestScope

instance in a test. But it doesn't say why? Doing this causes my

runTest

block to timeout and fail the test, because the class I injected the

TestScope

into uses it to

shareIn

, which never completes. So my question is what are the repercussions if I disregard this rule and instead call

runTest(testScope.testScheduler)

(which passes), rather than

testScope.runTest

as the docs suggest (which fails). Or, if I should really be using the recommended way, how do I get it to not time out for my use case?

Is there a way I can expose the latest element in a

Flow

to a non-suspend context? Collect to a

var

?

Hello! I am tasked with creating a "polling service" that consumes "polling ids" from a Kafka topic and runs HTTP requests to an API with these ids. Assuming no failures, the HTTP endpoint returns: • HTTP.OK and the applicable resource if it is ready • HTTP.Accepted and

Retry-After

header if it is not ready. (The same http request can be repeated after a delay.) Moreover, the HTTP endpoint does not process the requests that created the polling ids sequentially. When a resource is ready it should be produced to a new Kafka topic. I do realize that Kafka Streams might be a suitable option here, but it sounds quite difficult with the async nature of the http endpoint. I feel like Kotlin coroutines could benefit this implementation and with some research/thought I have made a "Proof of concept". Following is a very psuedo-ish code flow (pun intended):

fun main = runBlocking { <-- to obtain coroutinescope
  // 1. method running 'while loop' consuming Kafka topic inside flow builder. returns Flow<PollingId>

 // 2. method consuming flow from "1." and producing to new SharedFlow<Resource> 
 // flow.onEach { pollingId ->
 //  CoroutineScope(<http://Dispatchers.IO|Dispatchers.IO>).launch { 
 //    val resource = apiService.pollUntilReady(pollingid)
 //    resourceFlow.send(resource)
 //  }.collect()
^-- Launch each polling job in new coroutine job to obtain parallel processing. Should perhaps .map{} instead of .onEach {} to obtain job references for cleanup etc.

// 3. method consuming/collecting SharedFlow<Resource> that method "2." sends/produces to, and sending resources to new Kafka topic
}

Some feedback on the proposed implementation would be appreciated 🙂 Any pitfalls, improvements, is this a bad implementation, etc.