等待循环中的所有流程完成

Question

我有一个 API return 数据流：

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

当我获取数据时，我需要加载一些项目的详细信息并将结果转换为实时数据：

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)
             }
            }
   )

这里 emit(items) 在 loadDetails 之前调用的问题已完成，因此 item.details = details 较新调用。

如何等待 forEach 更新所有项目？

Answer 1

好吧，我在这里做了一些假设，所以如果我有任何错误，请在评论中纠正我，我会更新我的答案。

一般情况下，如果您不是绝对需要 flatMapConcat，则不鼓励使用它（它实际上写在函数的文档中）

我假设 loadDetails 可以简单地表示为这样的东西：

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

现在我们定义一个简单的辅助函数来获取 loadDetails

发出的第一个 Success 值

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
}

还有一个处理整个列表

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

}

此处注意：这将按顺序处理所有元素，如果您需要并行使用

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

现在对于 loadData 发出的每个元素列表，我们需要做的就是调用 populateDetails

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

等待循环中的所有流程完成

Waiting for all Flows in a loop finished

kotlin

kotlin-coroutines

kotlin-flow