Hi guys. A Flow from a DAO only completes when you cancel the coroutine that is collecting it. I want the flow to be completed when for loop is done. What should I do?

@AndroidEntryPoint
class RescheduleAlarmJobService : JobService() {

    @Inject lateinit var alarmRepository: AlarmRepository

    @Inject lateinit var scheduleAlarmManager: ScheduleAlarmManager

    val jobServiceScope = CoroutineScope(SupervisorJob())

    override fun onStartJob(p0: JobParameters?): Boolean {
        jobServiceScope.launch {
            alarmRepository.alarmsList.onCompletion {
                jobFinished(p0, false)
                jobServiceScope.cancel()
            }.buffer().collect { alarmList ->
                for (alarm in alarmList) {
                    if (alarm.isScheduled) {
                        scheduleAlarmManager.schedule(alarm)
                    }
                }
            }

        }
        return true
    }

    override fun onStopJob(p0: JobParameters?): Boolean {
        jobServiceScope.cancel()
        return false
    }
}

Direct solution is to throw CancellationException, but what you really want is to use first() with filter:

val scheduledAlarms = flow.map { alarmList -> 
   alarmList.filter { alarm -> alarm.isScheduled() }
}.first() 
scheduledAlarms.forEach { schedule(it) }

thanks so much Andrey Mischenko your answer works properly

Glad that it helps, just a small correction to my first snippet, it probably makes sense (depending on required semantics of course) to filter items which do not have any scheduled alarms (using filter after map), so you will get first scheduled element, but depends on your case of course, maybe you want completely even if nothing is scheduled

I appreciate your clarification