Hello, I'm trying to make a dynamic `sample`ing of...

# coroutines

[Nino]

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
    }
}

[ephemient]

there is a debounce overload taking a (T) -> timeout lambda, does that do what you want?

[Nino]

No, since there will be multiple flow update during the 100ms duration, debounce(100ms) will basically never emit 🙈

[ephemient]

if not you can implement your own operator using it as the starting point https://github.com/Kotlin/kotlinx.coroutines/issues/1216

[Nino]

debounce is not what I want. I want my flow to emit at most once every 100ms. This is sampling. Debounce "restart the waiting time" at every new upstream value, and since the upstream will emit a lot a values, debounce will wait forever

[Francis Reynders]

Maybe a naive way to achieve this, but since it is a shared flow, why not split it and merge it back? Something like this:

val fooFlow = merge(
        fooMutableStateFlow
            .filterNotNull()
            .sample<String>(SAMPLE_DURATION),
        fooMutableStateFlow
            .filter { it == null })
        .onStart { emit(fooMutableStateFlow.value) }
        .distinctUntilChanged()

[Nino]

val fooFlow : Flow<String?> = fooMutableStateFlow
    .distinctUntilChangedBy { it == null }
    .flatMapLatest {
        if (it == null) {
            flowOf(null)
        } else {
            fooMutableStateFlow.sample(SAMPLE_DURATION)
        }
    }

I did something similar but with flatMap in the end...

[Francis Reynders]

Interesting approach, that took a while to understand 🙂