/**
 * Implements function from
 * [RxJava](<http://reactivex.io/RxJava/3.x/javadoc/io/reactivex/rxjava3/core/Observable.html#throttleLatest-long-java.util.concurrent.TimeUnit->)
 */
internal fun <T> Flow<T>.throttleLatest(timeout: Duration) =
    conflate().transform {
        emit(it);
        delay(timeout);
    }

Hey, am I missing any flaws in this implementation? Thx 👋

Rx doesn't emit that final #FFFE4B item, but Flow.conflate() will

Are you sure? I think it will eventually be emitted if nothing else is emitted. It’s just that the collection ends before the throttle window, so it doesn’t have a chance to be emitted - compare the the green one for example. Thx for your answer!

if the upstream observable/flow ends within the window, that's what will happen

🤔 I think I’m missing this subtle difference. I’m gonna write a test for this case, maybe this will help me understand why this yellow one would be emitted in my implementation. Thanks!

a concrete example:

Observable.fromIterable(1..10).throttleLatest(1, TimeUnit.SECONDS).blockingForEach(::println) // (Rx) prints "1\n"
runBlocking { (1..10).asFlow().throttleLatest(Duration.seconds(1)).collect(::println) } // (yours) prints "1\n10\n"

Thanks, I get it now. So my implementation behaves like

Observable.throttleLatest

with

emitLast

set to

true

. That’s actually what I wanted 🙂

still not entirely the same as `emitLast = true`: Rx will terminate as soon as the last item is emitted, while your Flow operator will continue waiting

The remaining differences don’t matter in my use case, it’s close enough 🙂 Thanks again for your detailed explanation!