如何修改 Kotlin Flow uniqueUntilChanged 以添加到期时间

Question

我如何使用它并为其添加过期时间，这意味着如果流中存在相同的值，我们仍然会收集它，因为在发出前一个重复值之后distinctUntilChanged()它的时间超过了毫秒。expiry

flow { 
  emit("A")    // printed
  emit("B")    // printed
  emit("A")    // printed
  emit("A")    // NOT printed because duplicate
  delay(5000)
  emit("A")    // printed because 5 seconds elapsed which is more than expiry
}
  .distinctUntilChanged(expiry = 2000)
  .collect {
    println(it)
  }

我想打印这个：

A
B
A
A

这是测试它的代码：

  @Test
  fun `distinctUntilChanged works as expected`(): Unit = runBlocking {
    flow {
      emit("A")    // printed
      emit("B")    // printed
      emit("A")    // printed
      emit("A")    // NOT printed because duplicate
      delay(5000)
      emit("A")    // printed because 5 seconds elapsed which is more than expiry
    }
      .distinctUntilChanged(expiry = 2000)
      .toList().also {
        assertEquals("A", it[0])
        assertEquals("B", it[1])
        assertEquals("A", it[2])
        assertEquals("A", it[3])
      }
  }

Answer 1

我认为这会起作用，但我没有进行太多测试。我认为逻辑是不言自明的。存在的原因havePreviousValue是 caseT可为空且第一个发出的值为null。

fun <T> Flow<T>.distinctUntilChanged(expiry: Long) = flow {
    var havePreviousValue = false
    var previousValue: T? = null
    var previousTime: Long = 0
    collect {
        if (!havePreviousValue || previousValue != it || previousTime + expiry < System.currentTimeMillis()) {
            emit(it)
            havePreviousValue = true
            previousValue = it
            previousTime = System.currentTimeMillis()
        }
    }
}