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

Question

我在 Kotlin 中使用响应式(Reactive)编程，并尝试将 Flow 用作等效于 Flux(带有挂起功能)
我注意到那里缺少许多琐碎的功能。
Kotin List、Sequesnce 和 Flux 都有它们
流量只有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 共享。 ionic 。流中的值相同时，第二个 collect ionic 将是空的。
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
}

* 我找不到表达这种模式的组合器。我创建了 an issue为了这。
** 获得更多的计算机科学，是 closed over随流。