How we can collect flows from another collect or nested flow?

Question

I am getting Flow of records (Flow<List<Record>) from Dao, now on the basis of these records I have to fetch another list of flows (Flow<List<AnotherRecord>>) , but collect function inside of another collect function is not working .

 allRecords.collect { records ->
    withContext(Dispatchers.IO) {
        records.forEach { record ->
            taskRepository.getDatedTasks(record.createOn).collect {
                datedTasks.add(DatedTasks(record, it))
            }
        }

        // this line executes before of inner collect ,  I don't know why ?
        Log.d(TAG, "getAllDatedTasks: ${datedTasks.size}")
    }
}

Answer 1

A collect call from Room is infinite, so if you collect multiple flows inside this collect block, it will never get past the first one.

You also don't need to use withContext. Flows from Room already properly handle backgrounding their work.

You probably don't want all these collectors adding into the same mutable list, because there's no way to control it so they aren't adding multiple items. The final output should just be a Flow<List<DatedTask>> which you can use elsewhere as needed.

Maybe you can do this using combine.

val datedTasksFlow: Flow<List<DatedTasks>> = allRecords.flatMapLatest { records ->
    val flowsList = records.map { record ->
        taskRepository.getDatedTasks(record.createOn)
            .map { DatedTasks(record, it) }
    }
    combine(flowsList).map { it.toList } // convert the arrays to lists
}

Answer 2

I didn’t check if my code is right but I think it’s better like that :

allRecords.flowOn(Dispatchers.IO).flatMapConcat { record ->
    taskRepository.getDatedTasks(record.createOn).collect {
        datedTasks.add(DatedTasks(record, it))
    }
}.collect {
     Log.d(TAG, "getAllDatedTasks: ${datedTasks.size}")
}