FlowAssert example

FlowAssert.kt

import com.google.common.truth.Truth.assertThat
import kotlinx.coroutines.*
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.toList
import java.lang.Long.min

fun <T> synchronizedList(vararg elements: T): MutableList<T> = Collections.synchronizedList(mutableListOf(*elements))

class FlowAssert<T>(flow: Flow<T>) {
    companion object {
        private const val AWAIT_DURATION = 5000L
    }

    val values: MutableList<T> = synchronizedList()
    var exception: Throwable? = null
    private val job: Job
    private val monitor = Any()

    init {
        job = GlobalScope.async(Dispatchers.Default) {
            try {
                flow.toList(values)
            } catch (e: Exception) {
                exception = e
            }
        }
    }

    suspend fun awaitValueCount(count: Int, delayTime: Long = AWAIT_DURATION) = withContext(Dispatchers.Default) {
        val startTime = System.currentTimeMillis()
        while (values.size < count && System.currentTimeMillis() - startTime < delayTime) {
            synchronized(monitor) {
                if (exception != null)
                    return@withContext
            }
            delay(min(1, delayTime / 100))
        }
    }

    suspend fun awaitException(delayTime: Long = AWAIT_DURATION) = withContext(Dispatchers.Default) {
        val startTime = System.currentTimeMillis()
        while (System.currentTimeMillis() - startTime < delayTime) {
            synchronized(monitor) {
                if (exception != null)
                    return@withContext
            }
            delay(min(1, delayTime / 100))
        }
    }

    fun assertValueCount(count: Int) {
        assertThat(values).hasSize(count)
    }

    fun <T> assetValue(value: T) {
        assertThat(values.component1()).isEqualTo(value)
    }
}

fun <T> Flow<T>.test(): FlowAssert<T> {
    return FlowAssert(this)
}