Kotlin Flow missing many trivial functions like any(), distinct()

user_4489305 0
Viewed 1450

I'm using reactive programming in Kotlin and trying to use Flow as equivalent to Flux (with suspend functions)

I have noticed that many trivial functions are missing there.

Kotin List, Sequesnce have them as well as Flux

Flow has only distinctUntilChanged ( 1 1 1 2 2 1 -> 1 2 1) but why no distinct Am I missing something ?

// distinct
flow.toList().distinct()
// is working but probably not ideal for performance

Am I missing something ?

project-reactor kotlin-coroutines kotlin
1 Answers

Writing distnct imperatively without using any combinators* does not take long.

fun <T> Flow<T>.distinct(): Flow<T> = flow {
    val past = mutableSetOf<T>()
    collect {
        val isNew = past.add(it)
        if (isNew) emit(it)
    }
}

As to why this is not included, I can offer a speculation.
Flow objects usually represent long running flows of data, often living as long as the application. distinct either stops emitting (if the number of elements is finite), or has to keep an ever-growing set of seen values. I cannot think of an real-life use-case where this behaviour is needed. As writing it yourself is not hard at-all, this function probably does not belong in the library.

Here the mutable set is confined within the flow. For every new collect call, a new mutable set will be created. This flow object is thread-safe.
In the other answer, the set is created outside the flow.** Therefore it will be shared over multiple collections. With the same values in the flow, the second collection will be empty.

The any implementation in the other answer is correct.
As it was deleted, I will quote it here.

suspend fun <T> Flow<T>.any(predicate: (T) -> Boolean): Boolean =
    transform { if (predicate(it)) emit(Unit) }.firstOrNull() != null

It may be tempting to write firstOrNull(predicate) != null, but transforming to Unit is necessary. As T maybe nullable, we won't be able to tell whether the null returned is a real element or the because no element matches the predicate.

Referring to the implementation of firstOrNull, we can write the following imperative version.

suspend fun <T> Flow<T>.any(predicate: (T) -> Boolean): Boolean {
    var result = false
    collectWhile {
        if (predicate(it)) {
            result = true
            false
        } else {
            true
        }
    }
    return result
}

* I cannot find a combinator that expresses this pattern. I have created an issue for this.

** Getting more computer-sciency, it is closed over by the flow.

user_53630110
Related