is there any throttle operator for flow ?

# coroutines

[myanmarking]

is there any throttle operator for flow ?

[octylFractal]

do you mean something like https://github.com/Kotlin/kotlinx.coroutines/issues/1216 ? or is existing https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/debounce.html good enough?

[myanmarking]

no. debounce basically filter values based on timeout, but emits the last ones. throthle filter values based on timeout, but emits first ones

[octylFractal]

ah, I'm not super familiar with rx so didn't understand. sounds like https://github.com/Kotlin/kotlinx.coroutines/issues/1446

[myanmarking]

imagine some network call on a search functionality. basically, you filter out quickly emited items, and only get the last one based on some time

[myanmarking]

thats debounce

[myanmarking]

now imagine for instance, a button trigger a network call, and you want to filter out double quick emissions. you cannot use debounce because then the click will be delayed, you need throtle

[streetsofboston]

There is none, as far as i know.
But writing one yourself should not be that hard.

[myanmarking]

yes, im working on that right now

[tseisel]

I need the same operators, especially throttleLatest.
I started writing code to submit a PR in coroutines library

[kevin.cianfarini]

Maybe I'm not following correctly, but will this suffice? https://github.com/Kotlin/kotlinx.coroutines/pull/1798

[myanmarking]

yes, it seems so. is it in latest version yet ?

[kevin.cianfarini]

nope. I'm hoping to finish up the PR this weekend. Has been more complicated than anticipated

[kevin.cianfarini]

however, for the time being, you can emulate 99% of the functionality (except for some edge cases) with this

[kevin.cianfarini]

/**
 * Consume this [Flow] using a channelFlow with no buffer. Elements emitted from [this] flow
 * are offered to the underlying [channelFlow]. If the consumer is not currently suspended and 
 * waiting for the next element, the element is dropped. 
 * 
 * @return a flow that only emits elements when the downstream [Flow.collect] is waiting for the next element
 */
fun <T> Flow<T>.drop(): Flow<T> = channelFlow {
    collect { offer(it) }
}.buffer(capacity = 0)

[myanmarking]

ok. will try. thanks

[dekans]

Did you take a look at conflate()? It is not exactly the same behavior but maybe it could help