小编The*_*r42的帖子

我有一个可能永远不会被调用的监听器。但是，如果它至少被调用一次，我有理由确信它会被调用很多次。我是 Flows 的粉丝，因此我将其包装在callbackFlow() 构建器中。为了防止永远等待，我想添加一个超时时间。我正在尝试构建流运算符，如果流的第一个元素发出时间太长，它将抛出某种超时。这是我所拥有的。

fun <T> Flow<T>.flowBeforeTimeout(ms: Long): Flow<T> = flow {
    withTimeout(ms){
        emit(first())
    }
    emitAll(this@flowBeforeTimeout.drop(1))
}

效果还不错，这些 JUnit4 测试都通过了。还有更多通过的测试，但为了简洁起见我省略了它们。

@Test(expected = CancellationException::class)
fun `Throws on timeout`(): Unit = runBlocking {
    val testFlow = flow {
        delay(200)
        emit(1)
    }

    testFlow.flowBeforeTimeout(100).toList()
}

@Test
fun `No distortion`(): Unit = runBlocking {
    val testList = listOf(1,2,3)

    val resultList = testList
        .asFlow()
        .flowBeforeTimeout(100)
        .toList()

    assertThat(testList.size, `is`(resultList.size))
}

然而，这个测试没有通过。

// Fails with: Expected: is <1> but: was <2>
@Test
fun `Starts only once`(): Unit = …