Estoy tratando de crear una ventana móvil de datos usando Kotlin Flows. Se puede lograr en RxKotlin usando un búfer, pero no es lo mismo búfer usando Flujos.
RxKotlin tiene un operador de buffer , recopila periódicamente elementos emitidos por un Observable en paquetes y emite estos paquetes en lugar de emitir los elementos uno a la vez: búfer (contar, omitir)
Kotlin Flow tiene un buffer , pero solo ejecuta un colector en una corrutina separada - búfer
¿Hay algún operador existente en Flows que pueda lograr esto?
Creo que lo que está buscando no está disponible en la biblioteca Coroutines de Kotlinx, pero hay un problema abierto .
También hay una posible implementación en este comentario que también incluiré aquí:
fun <T> Flow<T>.windowed(size: Int, step: Int): Flow<List<T>> = flow { // check that size and step are > 0 val queue = ArrayDeque<T>(size) val toSkip = max(step - size, 0) < if sbd would like to skip some elements before getting another window, by serving step greater than size, then why not? val toRemove = min(step, size) var skipped = 0 collect { element -> if(queue.size < size && skipped == toSkip) { queue.add(element) } else if (queue.size < size && skipped < toSkip) { skipped++ } if(queue.size == size) { emit(queue.toList()) repeat(toRemove) { queue.remove() } skipped = 0 } } }