Hi guys. What is the equivalent of “onEachLatest” operator in Flow? I need to cancel previously running suspend fun in onEach block

Not sure what operator you're referring to, I can't find that in reactor or rxjava -- but perhaps

collectLatest

is what you want?

* The crucial difference from [collect] is that when the original flow emits a new value, [action] block for previous
 * value is cancelled.

Or

mapLatest

and simply returning the passed value.

collectLatest is not exactly what I need. It’s a suspend function, and I’ve decided to switch to flow.onEach() {} . launchIn(scope) construction

And

updateUI

can be safely canceled? If you get too many events you could also use

.conflate()

before

.onEach()

@Marc Knaup thx for your implementation but is it really missing in flow library?

A lot of operators could be added to the standard library 🙂

yes, updateUI can be canceled

You could suggest adding it and provide a use case: https://github.com/Kotlin/kotlinx.coroutines/issues/new

no,

debounce

takes the recent value in the time range

Yes, but it waits at least time range for that

e.g. if user types “abcde” I will get value only when he stops typing

Yeah for typing it makes sense

@Marc Knaup and what do you think to use for button clicks? simple flow with

collectLatest{}

or

single()

?

None of the standard operators is suitable 😞

I am wondering what behavior do you want to achieve 😉

In RxJava it would be

throttleLatest

@gabin you can use

mapLatest

as is (according to example you used)

events()
   .mapLatest { event -> updateUI(event) }
   .launchIn(scope)

@bezrukov map has slightly different purpose. It converts 1 type to another. I don’t want to achieve this

in your case you will convert it to unit, and it doesn't matter. You can review collectLatest implementation, and it's done via mapLatest as well.

ok, thanks for proposition