Kotlin Flow: emitAll is never collected

Question

I am trying to write a UnitTest for the kotlin-version of networkBoundResource that can be found on serveral sources with several features

Here is my version of it with marker-comments for the following question.

inline fun <ResultType, RequestType> networkBoundResource(
    ...
    coroutineDispatcher: CoroutineDispatcher
) = flow {

    emit(Resource.loading(null)) // emit works!

    val data = queryDatabase().firstOrNull()

    val flow = if (shouldFetch(data)) {
        
        emit(Resource.loading(data)) // emit works!

        try {
            saveFetchResult(fetch())
            query().map { Resource.success(it) }
            
        } catch (throwable: Throwable) {
            onFetchFailed(throwable)
            query().map { Resource.error(throwable.toString(), it) }
            
        }
    } else {
        query().map { Resource.success(it) }
    }

    emitAll(flow) // emitAll does not work!

}.catch { exception ->
    emit(Resource.error("An error occurred while fetching data! $exception", null))

}.flowOn(coroutineDispatcher)

This is one of my UnitTests for this code. The code is edited a bit to focus on my question:

@get:Rule
val testCoroutineRule = TestCoroutineRule()

private val coroutineDispatcher = TestCoroutineDispatcher()

@Test
fun networkBoundResource_noCachedData_shouldMakeNetworkCallAndStoreUserInDatabase() = testCoroutineRule.runBlockingTest {
    ...

    // When getAuthToken is called
    val result = networkBoundResource(..., coroutineDispatcher).toList()

    result.forEach {
        println(it)
    }    
}

The problem is that println(it) is only printing the Resource.loading(null) emissions. But if you have a look at the last line of the flow {} block, you will see that there should be another emission of the val flow. But this emission never arrives in my UnitTest. Why?

Answer 1

As @MarkKeen suggested, I now created my own implementation and it works quite well. Compared to the code that is going around on SO, this version now injects the coroutineDispatcher for easier testing, it lets flow take care of error handling, it does not contain nested flows and is imho easier to read and understand, too. There is still the side-effect of storing updated data to the database, but I am too tired now to tackle this.

import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.flow.*

inline fun <ResultType, RequestType> networkBoundResource(
    crossinline query: () -> Flow<ResultType?>,
    crossinline fetch: suspend () -> RequestType,
    crossinline saveFetchResult: suspend (RequestType) -> Unit,
    crossinline shouldFetch: (ResultType?) -> Boolean = { true },
    coroutineDispatcher: CoroutineDispatcher
) = flow<Resource<ResultType>> {
    
    // check for data in database
    val data = query().firstOrNull()
    
    if (data != null) {
        // data is not null -> update loading status
        emit(Resource.loading(data))
    }
    
    if (shouldFetch(data)) {
        // Need to fetch data -> call backend
        val fetchResult = fetch()
        // got data from backend, store it in database
        saveFetchResult(fetchResult)        
    }

    // load updated data from database (must not return null anymore)
    val updatedData = query().first()

    // emit updated data
    emit(Resource.success(updatedData))    
    
}.onStart { 
    emit(Resource.loading(null))
    
}.catch { exception ->
    emit(Resource.error("An error occurred while fetching data! $exception", null))
    
}.flowOn(coroutineDispatcher)

One possible UnitTest for this inline fun, which is used in an AuthRepsitory:

@ExperimentalCoroutinesApi
class AuthRepositoryTest {

    companion object {
        const val FAKE_ID_TOKEN = "FAkE_ID_TOKEN"
    }

    @get:Rule
    val testCoroutineRule = TestCoroutineRule()

    private val coroutineDispatcher = TestCoroutineDispatcher()

    private val userDaoFake = spyk<UserDaoFake>()

    private val mockApiService = mockk<MyApi>()

    private val sut = AuthRepository(
        userDaoFake, mockApiService, coroutineDispatcher
    )

    @Before
    fun beforeEachTest() {
        userDaoFake.clear()
    }

    @Test
    fun getAuthToken_noCachedData_shouldMakeNetworkCallAndStoreUserInDatabase() = testCoroutineRule.runBlockingTest {
        // Given an empty database
        coEvery { mockApiService.getUser(any()) } returns NetworkResponse.Success(UserFakes.getNetworkUser(), null, HttpURLConnection.HTTP_OK)

        // When getAuthToken is called
        val result = sut.getAuthToken(FAKE_ID_TOKEN).toList()

        coVerifyOrder {
            // Then first try to fetch data from the DB
            userDaoFake.get()

            // Then fetch the User from the API
            mockApiService.getUser(FAKE_ID_TOKEN)

            // Then insert the user into the DB
            userDaoFake.insert(any())

            // Finally return the inserted user from the DB
            userDaoFake.get()
        }
        
        assertThat(result).containsExactly(
            Resource.loading(null),
            Resource.success(UserFakes.getAppUser())
        ).inOrder()
    }
}

Answer 2

I'm not too sure of the complete behaviour, but essentially you want to get a resource, and current flow is all lumped into the FlowCollector<T> which makes it harder to reason and test.

I have never used or seen the Google code before and if I'm honest only glanced at it. My main take away was it had poor encapsulation and seems to break separations of concern - it manages the resource state, and handles all io work one one class. I'd prefer to have 2 different classes to separate that logic and allows for easier testing.

As simple pseudo code I would do something like this :

class ResourceRepository {

    suspend fun get(r : Request) : Resource {
        // abstract implementation details network request and io 
        // - this function should only fulfill the request 
        // can now be mocked for testing
        delay(3_000)
        return Resource.success(Any())
    }
}

data class Request(val a : String)

sealed class Resource {

    companion object {
        val loading : Resource get() = Loading
        fun success(a : Any) : Resource = Success(a)
        fun error(t: Throwable) : Resource = Error(t)
    }

    object Loading : Resource()

    data class Success(val a : Any) : Resource()

    data class Error(val t : Throwable) : Resource()
}

fun resourceFromRequest(r : Request) : Flow<Resource> =
    flow { emit(resourceRepository.get(r)) }
        .onStart { emit(Resource.loading) }
        .catch { emit(Resource.error(it)) }

This allows you to massively simplify the actual testing of the resourceFromRequest() function as you only have to mock the repository and one method. This allows you to abstract and deal with the networking and io work elsewhere, independently which again can be tested in isolation.