Just to make shure I got that right, I'd like to know if this chain does the following: - Run updateNotifications on each emission - If the function takes too long to execute, buffer 1 emission and ignore all others - When the function finishes, run it again if there's a buffered emission

flow()
        .conflate()
        .buffer(1)
        .onEach { updateNotifications() }
        .launchIn(GlobalScope)

That's what it should do AFAIK, but you can still test it, test sources are a great place for this.

conflate

and

buffer

are contradictory and will be fused, with

conflate

winning:

Note that conflate operator is a shortcut for buffer with capacity of Channel.CONFLATED.

Adjacent applications of conflate/buffer, channelFlow, flowOn, produceIn, and broadcastIn are always fused so that only one properly configured channel is used for execution. Conflation takes precedence over buffer() calls with any other capacity.

https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/conflate.html

That's right!

conflate()

always buffer latest element anyway, so you can get rid of

buffer(1)

and have the desired behavior.

Gosh, I should've read the entire KDoc of the operator 😅, thanks for the clarification, Flows are really awesome

Oh, I thought that was intentional.

Wrote some tests and the

conflate

operator alone did just what I needed 👍