如何启动 Kotlin Flow，但如果提前完成则将其返回的结果延迟一定的时间

Question

我正在尝试创建一个 Flow，通过在返回第一个结果之前等待至少一定时间来“修改”当前 Flow。

当然，我想立即开始处理当前流程，但要“保留”结果，直到时间过去。

我在当前正在编写的 Android 应用程序上需要这个，因为如果加载速度足够快，流程将在 NavHost 输入转换期间完成，从而导致卡顿。

当然，另一种方法是立即启动请求（不是 HTTP 请求，抱歉）并等待转换结束，但恐怕解决方案将与此类似（除了“等待”）流程与

delay

) 不同。

我发现了

zip

流程，因为它说：

使用应用于每对值的提供的变换函数将当前流（此）中的值与其他流压缩。其中一个流程完成后，生成的流程就会完成，并对剩余流程调用取消。

他们给出的例子是：

val flow = flowOf(1, 2, 3).onEach { delay(10) }
val flow2 = flowOf("a", "b", "c", "d").onEach { delay(15) }
flow.zip(flow2) { i, s -> i.toString() + s }.collect {
    println(it) // Will print "1a 2b 3c"
}

在示例中，剩余的

"d"

项被丢弃。

最后一句话是我在代码中看到的情况。 压缩完成后，“等待”流程不会自动取消。 由于我还在拉动刷新时重新启动此流程，因此下次它会卡住（没有这个

delayResultAtMost

就不会发生）。

这是我的代码，从LogCat，我只能看到：

开始
延迟：L：252

但我看不到“END”...

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.zip
import kotlin.time.Duration

/**
 * Waits [d] to produce a result if the result of the current flow is ready before [d].
 * For example, if [d] is 5 seconds and the result of the current flow is ready at `t = 2 seconds`,
 * this "modifier" will wait another 3 seconds before returning the result.
 *
 * If, instead, the result is ready at `t = 7 seconds`, return it when it's ready.
 *
 * @param d the minimum time to wait for the result.
 */
fun <T> Flow<T>.delayResultAtMost(d: Duration): Flow<T> {
    val waitFlows = flow {
        val ctx = currentCoroutineContext()

        Log.d("FLOW", "START")
        val start = System.currentTimeMillis()

        delay(d)

        val end = System.currentTimeMillis()
        val delta = end - start
        Log.d("FLOW", "DELAY: L: $delta")

        while(ctx.isActive) {
            emit(Unit)
        }

        Log.d("FLOW", "END")
    }

    return this.zip(waitFlows) { a, _ -> a }
}

有没有办法在第一个流程完成时自动取消另一个流程？ 还是这种方法完全错误？ （顺便说一句，这可能是因为这是我第一次使用 Kotlin Flow 和 Jetpack Compose）。

Answer 1

您可能应该使用

combine

而不是 zip。

zip

的问题是，它在发送每个流中的值之前仅组合一次，因此您的“延迟流”最终会非常尴尬，因为它需要发出与您所需的流一样多的“虚拟”值。

另一方面，Combine 会组合最新的值，在这种情况下，可以使用仅发出一次的“延迟流”。

试试这个：

fun <T> Flow<T>.delayResultAtMost(d: Duration) = combine(
    flowOf(Unit).onStart { delay(d) }
) { v, _ -> v }