emitAll until another operation is done?

Question

I'm using the following code in my Android app to load data from a remote source into a local database cache.

inline fun <ResultType, RequestType> networkBoundResource(
    crossinline query: () -> Flow<ResultType>,
    crossinline fetch: suspend () -> RequestType,
    crossinline saveFetchResult: suspend (RequestType) -> Unit,
    crossinline onFetchSuccess: () -> Unit = { },
    crossinline onFetchFailed: (Throwable) -> Unit = { },
    crossinline shouldFetch: (ResultType) -> Boolean = { true }
) = flow {
    val data = query().first()

    val flow = if (shouldFetch(data)) {
        emit(Resource.(data))

        try {
            // this could take a while, I want to keep getting updates meanwhile
            saveFetchResult(fetch())
            onFetchSuccess()
            query().map { Resource.Success(it) }
        } catch (t: Throwable) {
            onFetchFailed(t)
            query().map { Resource.Error(t, it) }
        }
    } else {
        query().map { Resource.Success(it) }
    }
    emitAll(flow)
}

The query is a database query that keeps emitting database updates through emitAll until we call this method again.

The problem with this setup is that Resource.Loading only contains a "snapshot" of the current data (first()) and we won't receive any database updates until we get to the end of the try/catch block and call emitAll. But I would like to keep receiving database updates while Loading is still in progress. However, I can't just call emitAll on Resource.Loading because it would block the whole Flow.

Is there a way to call emitAll on Loading and then switch to Success/Error once the try block has finished?

N1hk N1hk · Accepted Answer · 2021-01-31T22:18:19

I've only done simple testing on this to validate it, but it looks like you can listen to the query and emit any/all data it propagates in a newly launched coroutine based on the outer Flow's context -- the other work in the function will continue, unblocked. Once that other work is done, the coroutine that's listening to the query can be cancelled. For example:

inline fun <ResultType, RequestType> networkBoundResource(
    crossinline query: () -> Flow<ResultType>,
    crossinline fetch: suspend () -> RequestType,
    crossinline saveFetchResult: suspend (RequestType) -> Unit,
    crossinline onFetchFailed: (Throwable) -> Unit = { },
    crossinline shouldFetch: (ResultType) -> Boolean = { true }
): Flow<Resource<ResultType>> = flow {
    emit(Resource.Loading())
    val data = query().first()

    val flow = if (shouldFetch(data)) {
        val flowContext = currentCoroutineContext()
        val loading: Job = coroutineScope {
            launch(flowContext) {
                query().map { Resource.Loading(it) }
                    .collect { withContext(flowContext) { emit(it) } }
            }
        }

        try {
            val request = fetch()
            loading.cancel()
            saveFetchResult(request)
            query().map { Resource.Success(it) }
        } catch (throwable: Throwable) {
            loading.cancel()
            onFetchFailed(throwable)
            query().map { Resource.Error(throwable, it) }
        }
    } else {
        query().map { Resource.Success(it) }
    }

    emitAll(flow)
}

Let me know if this works out!

emitAll until another operation is done?

1 Answers