FlowRecorder for recording emissions from flows.

FlowRecorder.kt

/**
 * Starts subscribing / collecting the given Flow and records it's emission to verify them later
 */
suspend fun <T> Flow<T>.recordEmissions(
    recordingScope: CoroutineScope = GlobalScope,
    verify: suspend SharedFlow<T>.() -> Unit
) {
    val recording = shareIn(recordingScope, SharingStarted.Eagerly, Int.MAX_VALUE)
    verify(recording)
}

suspend fun <T> SharedFlow<T>.verifyAll(vararg emissions: T, emissionTimeoutMilliseconds: Long = 1000L) {
    verifyAllList(emissions.toList(), emissionTimeoutMilliseconds)
}

suspend fun <T> SharedFlow<T>.verifyAllList(nextEmissions: List<T>, emissionTimeoutMilliseconds: Long) {
    val result = waitForIndex(emissionTimeoutMilliseconds, nextEmissions.size)
    if (result == null) {
        error(
            "\nWaiting for     : $nextEmissions \n" +
                    "Emissions so far: $replayCache"
        )
    } else {
        assertThat(nextEmissions).isEqualTo(replayCache)
        assertThat(nextEmissions).hasSameSizeAs(replayCache)
    }
}

suspend fun <T> SharedFlow<T>.verityItemAt(emissionTimeoutMilliseconds: Long, index: Int, verifyBlock: (a: T) -> Unit) {
    val result = waitForIndex(emissionTimeoutMilliseconds, index + 1)

    if (result == null) {
        error(
            "Waiting for $index but no new emission within " +
                    "${emissionTimeoutMilliseconds}ms. Emissions so far: $replayCache"
        )
    } else {
        verifyBlock(replayCache[index])
    }
}

suspend fun <T> SharedFlow<T>.waitForIndex(
    emissionTimeoutMilliseconds: Long,
    size: Int
): Unit? {
    return withTimeoutOrNull(emissionTimeoutMilliseconds) {
        while (replayCache.size < size) {
            delay(100)
        }
    }
}