Waiting for all Flows in a loop finished

user_5599807 0
Viewed 2392

I have an API that return data as flows:

suspend fun loadData(...): Flow<List<Items>> 
suspend fun loadDetails(...): Flow<ItemDetails>

When I get data I need to load details for few items and convert result to live data:

job = SupervisorJob()
val context = CoroutineScope(job + Dispatchers.Main.immediate).coroutineContext
liveData(context = context) {
  emitSource(
        api.loadData(...)
           .flatMapConcat { items ->
              items.forEach { item -> 
                 if (item is Details){
                    api.loadDetails(item).flatMapConcat{ details ->
                        item.details = details
                    }
                 }
              }
             flow {
               emit(items)
             }
            }
   )

the issue here that emit(items) called before loadDetails completed, so item.details = details newer called.

How to wait forEach updates all items?

kotlin-flow kotlin-coroutines kotlin
1 Answers

Alright I am making a few assumptions here, so if I got anything wrong, correct me in the comments and I will update my answer.

In general the use of flatMapConcat is discouraged if you don't absolutely need it (it's actually written in the docs of the function)

I will assume loadDetails can be simply represented as something like this:

suspend fun loadDetails(item: Details) = flow<Any>{
    emit(Loading()) // some sort of sealed wrapper class
    val value = someAsyncOperation()
    emit(Success(value))
}

Now we define a simple helper function to fetch the first Success value emitted by loadDetails

suspend fun simplerLoadDetails(item: Details): Any { 
    // this will collect items of the flow until one matches the lambda
    val wrappedSuccessValue = loadDetails(item)
        .first { it is Success } as Success 
    return wrappedSuccessValue.value
}

And another one to process the entire list

suspend fun populateDetails(items: List<Any>) {
    items.forEach {
        if (it is Details) {
            it.details = simplerLoadDetails(it)
        }  
    }

}

A note here: this will process all elements sequential, if you need it in parallel use

suspend fun populateDetails(items: List<Any>) {
    coroutineScope {
        items.forEach {
            if (it is Details) {
                launch {
                    it.details = simplerLoadDetails(it)
                }  
            }
        }
    }
}

Now for each list of elements emitted by loadData, all we need to do is call populateDetails

suspend fun itemsWithDetails(...) = loadData(...)
    .onEach { itemsList ->
        populateDetails(itemsList)
    }
user_90159420
Related