Tengo un List<Flow<T>> y me gustaría generar un Flow<List<T>> . Esto es casi lo que hace combine , excepto que Combine espera que todos y cada uno de los Flow emitan un valor inicial, que no es lo que quiero. Tome este código por ejemplo:
val a = flow { repeat(3) { emit("a$it") delay(100) } } val b = flow { repeat(3) { delay(150) emit("b$it") } } val c = flow { delay(400) emit("c") } val flows = listOf(a, b, c) runBlocking { combine(flows) { it.toList() }.collect { println(it) } } Con combine (y por lo tanto tal cual), esta es la salida:
[a2, b1, c] [a2, b2, c]Mientras que también estoy interesado en todos los pasos intermedios. Esto es lo que quiero de esos tres flujos:
[] [a0] [a1] [a1, b0] [a2, b0] [a2, b1] [a2, b1, c] [a2, b2, c]En este momento tengo dos soluciones alternativas, pero ninguna de ellas es excelente... La primera es simplemente fea y no funciona con tipos anulables:
val flows = listOf(a, b, c).map { flow { emit(null) it.collect { emit(it) } } } runBlocking { combine(flows) { it.filterNotNull() }.collect { println(it) } } Al obligar a todos los flujos a emitir un primer valor irrelevante, se llama al transformador combine y me permite eliminar los valores nulos que sé que no son valores reales. Iterando sobre eso, más legible pero más pesado:
sealed class FlowValueHolder { object None : FlowValueHolder() data class Some<T>(val value: T) : FlowValueHolder() } val flows = listOf(a, b, c).map { flow { emit(FlowValueHolder.None) it.collect { emit(FlowValueHolder.Some(it)) } } } runBlocking { combine(flows) { it.filterIsInstance(FlowValueHolder.Some::class.java) .map { it.value } }.collect { println(it) } }Ahora este funciona bien, pero todavía se siente como si estuviera exagerando. ¿Hay algún método que me falta en la biblioteca de rutinas?
Todavía me gustaría evitar la asignación a un tipo de contenedor intermediario, y como alguien mencionó en los comentarios, el comportamiento es ligeramente incorrecto (esto emite una lista vacía al principio si aún no se emitieron argumentos), pero esto es un poco mejor que las soluciones Lo tenía en mente cuando escribí la pregunta (todavía muy similar) y funciona con tipos anulables:
inline fun <reified T> instantCombine( flows: Iterable<Flow<T>> ): Flow<List<T>> = combine(flows.map { flow -> flow.map { @Suppress("USELESS_CAST") // Required for onStart(null) Holder(it) as Holder<T>? } .onStart { emit(null) } }) { it.filterNotNull() .map { holder -> holder.value } }Y aquí hay un conjunto de pruebas que pasa con esta implementación:
class InstantCombineTest { @Test fun `when no flows are merged, nothing is emitted`() = runBlockingTest { assertThat(instantCombine(emptyList<Flow<String>>()).toList()) .isEmpty() } @Test fun `intermediate steps are emitted`() = runBlockingTest { val a = flow { delay(20) repeat(3) { emit("a$it") delay(100) } } val b = flow { repeat(3) { delay(150) emit("b$it") } } val c = flow { delay(400) emit("c") } assertThat(instantCombine(a, b, c).toList()) .containsExactly( emptyList<String>(), listOf("a0"), listOf("a1"), listOf("a1", "b0"), listOf("a2", "b0"), listOf("a2", "b1"), listOf("a2", "b1", "c"), listOf("a2", "b2", "c") ) .inOrder() } @Test fun `a single flow is mirrored`() = runBlockingTest { val a = flow { delay(20) repeat(3) { emit("a$it") delay(100) } } assertThat(instantCombine(a).toList()) .containsExactly( emptyList<String>(), listOf("a0"), listOf("a1"), listOf("a2") ) .inOrder() } @Test fun `null values are kept`() = runBlockingTest { val a = flow { emit("a") emit(null) emit("b") } assertThat(instantCombine(a).toList()) .containsExactly( emptyList<String?>(), listOf("a"), listOf(null), listOf("b") ) .inOrder() } }Qué tal esto:
inline fun <reified T> instantCombine(vararg flows: Flow<T>) = channelFlow { val array= Array(flows.size) { false to (null as T?) // first element stands for "present" } flows.forEachIndexed { index, flow -> launch { flow.collect { emittedElement -> array[index] = true to emittedElement send(array.filter { it.first }.map { it.second }) } } } }Resuelve algunos problemas:
[] no está en el flujo resultantePor lo tanto, no notará ninguna solución alternativa específica de implementación, porque no tiene que lidiar con eso durante la recopilación:
runBlocking { instantCombine(a, b, c).collect { println(it) } }Producción:
[a0]
[a1]
[a1, b0]
[a2, b0]
[a2, b1]
[a2, b1, c]
[a2, b2, c]
Editar: respuesta actualizada para manejar flujos que también emiten valores nulos.
* La matriz de bajo nivel utilizada es segura para subprocesos. Es como si estuvieras tratando con variables individuales.