Kotlin Flow 缺少许多琐碎的函数，例如 any()、distinct()

Question

我在 Kotlin 中使用反应式编程，并尝试使用 Flow 等同于 Flux（带有挂起函数）

我注意到那里缺少许多琐碎的功能。

Kotin List、Sequensnce 以及 Flux 都有它们

Flow 只有distinctUntilChanged( 1 1 1 2 2 1 -> 1 2 1) 但为什么没有distinct 我错过了什么吗？

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

我错过了什么吗？

Answer 1

不使用任何组合器*的命令式编写distnct不会花费很长时间。

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

至于为什么不包括在内，我可以提供一个猜测。
Flow对象通常代表长期运行的数据流，通常与应用程序一样长。distinct要么停止发射（如果元素数量有限），要么必须保留一组不断增长的可见值。我想不出现实生活中需要这种行为的用例。由于自己编写它并不难，因此该函数可能不属于库中。

这里可变集被限制在flow. 对于每个新collect调用，都会创建一个新的可变集。该流对象是线程安全的。
在另一个答案中，该集合是在流程之外创建的。**因此它将在多个collect离子上共享。在流中具有相同值的情况下，第二个collect离子将为空。

any其他答案中的实现是正确的。
由于已被删除，我在此引用一下。

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

写成 可能很诱人firstOrNull(predicate) != null，但转换成Unit是必要的。由于T可能可为空，我们将无法判断返回null的是真实元素还是 ，因为没有元素与 . 匹配predicate。

参考 的实现firstOrNull，我们可以编写以下命令式版本。

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

* 我找不到表达这种模式的组合器。我为此创建了一个问题。

** 越来越计算机科学了，它被流程封闭了。