如何为Kotlin Flow的distinctUntilChanged添加过期时间？

我需要给Kotlin Flow的distinctUntilChanged()扩展一个过期逻辑：当相同值距离上次重复发射的时间间隔超过expiry毫秒时，这个值依然会被下游收集。

示例代码 #

flow { 
  emit("A")    // 会被打印
  emit("B")    // 会被打印
  emit("A")    // 会被打印
  emit("A")    // 不会被打印，因为是重复值
  delay(5000)
  emit("A")    // 会被打印，因为间隔超过了设定的2000毫秒过期时间
}
  .distinctUntilChanged(expiry = 2000)
  .collect {
    println(it)
  }

预期输出 #

A
B
A
A

测试代码 #

@Test
fun `distinctUntilChanged works as expected`(): Unit = runBlocking {
  flow {
    emit("A")    // 会被收集
    emit("B")    // 会被收集
    emit("A")    // 会被收集
    emit("A")    // 不会被收集，因为是重复值
    delay(5000)
    emit("A")    // 会被收集，因为间隔超过了2000毫秒过期时间
  }
    .distinctUntilChanged(expiry = 2000)
    .toList().also {
      assertEquals("A", it[0])
      assertEquals("B", it[1])
      assertEquals("A", it[2])
      assertEquals("A", it[3])
    }
}

内容的提问来源于stack exchange，提问作者vovahost