如何使用协程 Flow<T> DROP_LATEST？

Question

请帮忙。我试图这样做：

    2sec     2sec     2sec
------[A]------[B]------[C]------...----------------> InitailFlow
       \        |        | 
        \      drop      drop
         \
     5sec \    5sec        5sec
----------[1]---------[2]---------[3]-----|> AnotherFlow
result: [A1, A2, A3]

所以我有 InitailFlow，它会发出很短的时间（2 秒），然后转换为 AnotherFlow，它需要更长的时间才能完成（总共 15 秒）...我想删除由InitialFlow 而 AnotherFlow 尚未完成...

我试过了：

flow{
    delay(2000)
    emit("A")
    delay(2000)
    emit("B")
    delay(2000)
    emit("C")
}.buffer(0, BufferOverflow.DROP_LATEST)
    .onEach {
       println("Event for $it")
    }
    .flatMapConcat {
       flow {
           delay(5000)
           emit("${it}1")
           delay(5000)
           emit("${it}2")
           delay(5000)
           emit("${it}3")
        }
     }
     .onEach {
         println(it)
     }
     .launchIn(scope)

但由于某种原因，结果如下：

Event for A
A1
A2
A3
Event for B
B1
B2
B3

即使我有一个.buffer(0, BufferOverflow.DROP_LATEST)，它仍然会处理Event B。

为什么还要处理事件 B？

有没有办法做到这一点？我希望输出只有：

Event for A
A1
A2
A3

提前致谢。

Answer 1

这应该适合你：

fun <T> Flow<T>.dropIfBusy(): Flow<T> = flow {
    coroutineScope {
        val channel = produce(capacity = Channel.RENDEZVOUS) {
            collect { offer(it) }
        }
        channel.consumeEach { emit(it) }
    }
}

这基本上是来自kotlin docs 的“幼稚”buffer 实现
这里唯一的区别是我们使用channel.offer 而不是channel.send 当与 RENDEZVOUS 通道一起使用时，提供给通道的所有值（在暂停时）都会被丢弃，从而创建您想要的行为。

Answer 2

在玩了一会儿@AdrianK 的解决方案之后，我实际上找到了一个使用channelFlow 的更简单的解决方案。由于channelFlow 目前是实验性 API，您必须选择使用它。

像这样：

fun <T> Flow<T>.dropIfBusy(): Flow<T> = channelFlow {
    collect { offer(it) }
}.buffer(0)

Answer 3

每次调用 emit 时都会暂停，因此 buffer 运算符无效。

我不确定您期望的输出，但我认为您正在寻找 conflate 运算符：

flow{
    delay(2000)
    emit("A")
    delay(2000)
    emit("B")
    delay(2000)
    emit("C")
}
.conflate()
.onEach {
    println("Event for $it")
}
.flatMapConcat {
    flow {
        delay(5000)
        emit("${it}1")
        delay(5000)
        emit("${it}2")
        delay(5000)
        emit("${it}3")
    }
}
.onEach {
    println(it)
}
.launchIn(scope)

conflate 运营商官方文档指出：

通过合并通道合并流量排放并在一个集合中运行收集器 单独的协程。这样做的效果是发射器永远不会 由于收集器速度慢而暂停，但收集器总是得到最多 最近发出的值。

您可以找到conflate运营商官方文档here。