Android：在另一个不发射的 Kotlin Flow 中收集

问题描述

我有以下方法：

    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)) 获取类别，然后它可以正常工作并发出列表。

非常感谢！

解决方法

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

您的 DAO 返回的 Flows 是无限长的。第一个值是进行的第一个查询，但 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 中，您可以创建一个 suspend 的 getCategoryById() 函数版本，它只返回列表。