Android：在另一个不发射的内部收集一个 Kotlin Flow答案

【问题标题】：Android: collecting a Kotlin Flow inside another not emittingAndroid：在另一个不发射的内部收集一个 Kotlin Flow
【发布时间】：2021-01-04 21:46:56
【问题描述】：

我有以下方法：

    operator fun invoke(query: String): Flow<MutableList<JobDomainModel>> = flow {
        val jobDomainModelList = mutableListOf<JobDomainModel>()
        jobListingRepository.searchJobs(sanitizeSearchQuery(query))
            .collect { jobEntityList: List<JobEntity> ->
                for (jobEntity in jobEntityList) {
                    categoriesRepository.getCategoryById(jobEntity.categoryId)
                        .collect { categoryEntity ->
                            if (categoryEntity.categoryId == jobEntity.categoryId) {
                                jobDomainModelList.add(jobEntity.toDomainModel(categoryEntity))
                            }
                        }
                }
                emit(jobDomainModelList)
            }
    }

它在调用search 方法的存储库中搜索，该方法返回Flow<List<JobEntity>>。然后对于流程中的每个JobEntity，我需要从数据库中获取该作业所属的类别。获得该类别和作业后，我可以将作业转换为域模型对象 (JobDomainModel) 并将其添加到列表中，该列表将作为方法的返回对象在流中返回。

我遇到的问题是什么都没有发出。我不确定在 Kotlin 中使用流时是否遗漏了一些东西，但我没有按 ID (categoriesRepository.getCategoryById(jobEntity.categoryId)) 获取类别，然后它可以正常工作并发出列表。

提前非常感谢！

【问题讨论】：

只是为了确定：searchJobs() 实际上是值流（例如 WebSocket）还是单次请求？
它也是一个 Flow，因为它通过 DAO 使用 Room 在 DB 中搜索
我也尝试过使用地图而不是收集（在获取工作类别时），但它仍然不起作用

标签： android kotlin kotlin-flow

【解决方案1】：

我认为问题在于您正在收集无限长度的流，因此collect 永远不会返回。您应该在收集之前使用.take(1) 来获取有限流，或者使用first()。

您的 DAO 返回的流是无限长的。第一个值是第一个查询，但 Flow 将永远持续下去，直到被取消。 Flow 中的每一项都是数据库内容发生变化时的新查询。

类似这样的：

operator fun invoke(query: String): Flow<MutableList<JobDomainModel>> =
    jobListingRepository.searchJobs(sanitizeSearchQuery(query))
        .map { jobEntityList: List<JobEntity> ->
            jobEntityList.mapNotNull { jobEntity ->
                categoriesRepository.getCategoryById(jobEntity.categoryId)
                    .first()
                    .takeIf { it.categoryId == jobEntity.categoryId }
            }
        }

或者，在您的 DAO 中，您可以创建一个 getCategoryById() 的 suspend 函数版本，它只返回列表。

【讨论】：