сделать атомик флаг, и  повесить слушатель на окончание исходного флов, вроде ьыл такой

А почему не повесить collect в конец без ланча и сделать на нем try-catch

я про цикл скорей

он ведь никогда не кончится, или это так и задумано ?

А зачем? Флаг-то уже есть isActive назвыается, если уж очень нужно

он обрушится вместе со скоупом

он есть у флов ?

В channelFlow есть

так речь про исходный флов, он ведь может перестать емитить, как узнать каналу что все хватит ?

Вот тут я уже навскидку не отвечу. По хорошему там контекста взад прокидывается, и если корутина упадет, он дожен само-отмениться, но голову на отсечение не дам

или ук них общий скоуп ?

Такс, только что проверил:

fun <T> Flow<T>.timedBuffer(millis:Long):Flow<List<T>> = channelFlow {
   val list = arrayListOf<T>()

   launch {
       while(isActive) {
           delay(millis)

           if(list.isNotEmpty()){
               send(ArrayList(list))
               list.clear()
           }
       }
   }
   collect { list.add(it) }
}

private val job = SupervisorJob()
private val scope = CoroutineScope(Dispatchers.Main + job)

scope.launch {
   foo()
       .timedBuffer(1000)
       .collect {
           print(it)
       }
}


Вообщем при отмене scope все завершается, channel тоже отменяется и isActive становится false

private val job = SupervisorJob()
private val scope = CoroutineScope(Dispatchers.Main + job)
Зачем такая штука? Просто supervisorScope нельзя?

Там косяк с синхронизацией. Есть риск потерять элемент

Может быть ситуация, когда элемент добавился прямо перед clear

Я сейчас думаю как это исправить, у меня такая штука может быть 100%

Возьмите мутекс из моего примера

Поставьте после
`.collect { print(it) }`
любую suspend-функцию и посмотрите, выполнится ли она

А разве аналогичный буфер из rx фильтрует пустые списки?

Функция не выполниться, потому как при отмене, collect выбрасывает kotlinx.coroutines.JobCancellationException: