Hey all! I'm learning how state flows work and trying to test something with the

StateFlow

. My test looks like this:

@Test
fun `test state flow`() = runBlocking {
    val stateFlow = flowOf(1, 2, 3)
        .onEach {
            println(it)
        }
        .stateIn(
            scope = TestCoroutineScope(),
            started = SharingStarted.WhileSubscribed(),
            initialValue = 0
        )
    
    val results = mutableListOf<Int>()
    val job = stateFlow.collectAsync(TestCoroutineScope()) {
        results.add(it)
    }

    assertEquals(listOf(0, 1, 2, 3, 4), results)
    
    job.cancel()
}

I was expecting this test to pass but it is failing. The actual size of results is

1

and the value it contains is

3

. I debugged my test with

onEach

and in console every value from the source flow is printed which means the value was emitted but it was never received in the collector. Does anybody know why am I only receiving the last value in my collector in the above code?

Use a

StateFlow

when you only care about the latest value. If you need every value, you should use a

SharedFlow

. I think 1, 2, 3 are all getting processed before the

collect

lambda runs so only the 3 is left because it's the latest.

flowOf

doesn't suspend while looping through values and sending values into the

StateFlow

never suspends (it just throws away old unprocessed values), so the initial code populating the

StateFlow

never yields and finishes without the

collect

code having a chance to run.

Why does this happen only on the initial code? I tried the following code and it is working as expected:

@Test
fun `test state flow`() = runBlocking {
    val channel = Channel<Int>()
    val stateFlow = channel.receiveAsFlow()
        .onEach {
            println(it)
        }
        .stateIn(
            scope = TestCoroutineScope(),
            started = SharingStarted.WhileSubscribed(),
            initialValue = 0
        )
    
    val results = mutableListOf<Int>()
    val job = stateFlow.collectAsync(TestCoroutineScope()) {
        results.add(it)
    }

    channel.trySend(1)
    channel.trySend(2)
    channel.trySend(3)

    assertEquals(listOf(0, 1, 2, 3, 4), results)
    
    job.cancel()
}

I'm not sure exactly why this one prints all of them, apparently

trySend

is directly resuming the

collect

code in

stateIn

. But these two examples are quite different. Consider

flowOf(1,2,3).collect { printLn(it) }

. It never suspends. It all runs synchronously inside the collecting coroutine. With the

Channel

you have two coroutines, one `trySend`ing and the other `collect`ing.

That's is a fair suggestion, but using

runBlockingTest

does not change results of this test. But I understand your point here. It would provide us with the predictable behavior where we can understand what is actually happening. I'll have to dig deeper into this. But that you for your help @Nick Allen