Add method awaitUntil(predicate: (item: T) -> Boolean): T

[fr-gonzalez-globant]

Add method awaitUntil(predicate: (item: T) -> Boolean): T

Sometimes you just want to await for a certain item to emit when testing a flow but you don't know how many items to skip in order to catch that item.

flowOf(1, 2, 3).test {
  val item = awaitUntil { it == 3 }
  assertEquals(3, item)
  awaitComplete()
}

[JakeWharton]

We need to agree upon, document, and then test the timeout behavior. Unlike other functions, this can suspend forever so long as events are flowing. Is that what we want? Or should an overall timeout be enforced?

[fr-gonzalez-globant]

We need to agree upon, document, and then test the timeout behavior. Unlike other functions, this can suspend forever so long as events are flowing. Is that what we want? Or should an overall timeout be enforced?

I never thought of that scenario, but yes, it is possible to suspend forever with a continous flow that never matches the predicate. I tried modifying the function to receive a duration and use withAppropriateTimeout like:

// ReceiveTurbine.kt
public suspend fun awaitUntil(timeout: Duration = 3.seconds, predicate: (item: T) -> Boolean): T

// channel.kt
public suspend fun <T> ReceiveChannel<T>.awaitUntil(
  name: String? = null,
  timeout: Duration,
  predicate: (item: T) -> Boolean): T = withAppropriateTimeout(timeout) {
    //try to find and return the item
}

This makes this test pass:

@Test
fun awaitUntilThrowsOnTimeout() = runTest {
  var item = 0
  flow {
    while (true) {
      emit(item++)
      delay(1.seconds)
    }
  }.test {
    val message = assertFailsWith<TurbineTimeoutCancellationException> {
      awaitUntil { it < 0 }
    }.message
    assertEquals("Timed out waiting for 3s", message)
  }
}

But I don't know if this is the right way to tackle the problem or if this solution suits the way this library was designed.