Flow frío

Al terminar esta lección, el partido dejará de ser «una función que devuelve el resultado al final» y pasará a ser un flujo de eventos que llegan uno a uno: el gol del 12, la tarjeta del 30, el cambio del 85. Podrás narrarlo en directo por consola, quedarte solo con los goles y cortar la emisión al primero. Y jugarPartido, que ya escribiste, se reescribirá encima del flujo sin cambiar lo que devuelve (a eso se le llama refactorizar: cambiar el interior de un código sin cambiar lo que hace por fuera).

Importa porque un Flow es la forma de Kotlin de representar varios valores a lo largo del tiempo, y en Android lo verás en pantallas, bases de datos y red. Esta lección es la parte 1: el flujo frío y sus operadores básicos (flow, collect, map, filter, filterIsInstance, flowOn); el resto llega en Flow: scan y combine.

Un flujo que no arranca hasta que alguien lo recoge

Una función suspend devuelve un valor, al final. Un Flow<T> entrega cero, uno o muchos valores de tipo T, a medida que se producen (T es el tipo de cada valor, como en List<Jugador>; lo viste en Genéricos). Se construye con flow { … }, y dentro se entrega cada valor con emit. Quien lo consume lo recoge (collect).

La palabra clave es frío: el bloque de flow { } no corre al crearlo. Corre cuando alguien lo recoge, y cada recogida lo ejecuta entero desde el principio.

Un operador es una función que se aplica a un flujo con un punto (flujo.map { … }), como las de las listas que ya conoces. Hay dos tipos:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun numeros(): Flow<Int> = flow {
    println("  (el flow arranca)")
    for (n in 1..5) {
        delay(100)          // trabajo que tarda
        emit(n)             // entrega un valor y espera a que el colector lo procese
    }
}
 
fun main() = runBlocking {
    val pares = numeros()
        .filter { it % 2 == 0 }        // se queda solo con los pares
        .map { "número $it" }          // transforma cada uno
    println("Flow creado")             // hasta aquí, nada ha corrido
    pares.collect { println(it) }
}

Puedes probarlo tal cual en el Kotlin Playground: import kotlinx.coroutines.* trae todo lo de ese paquete de una vez (runBlocking, delay) y el segundo import …* trae Flow, flow, filter y map. Imprime:

Flow creado
  (el flow arranca)
número 2
número 4

El número 2 sale hacia los 200 ms y el número 4 hacia los 400 ms, contados desde que arranca el flow. Fíjate en el orden: Flow creado sale antes que (el flow arranca).

flowOn: en qué hilo corre el flujo

Por defecto, el flujo corre donde corre quien lo recoge. Con flowOn(Dispatchers.Default) pasas lo que está antes de él (el flow { } y los operadores anteriores) a un hilo del grupo Default (Dispatchers y withContext); lo que va después sigue donde estaba quien recoge:

flow { … }  →  map { … }  →  flowOn(Default)  →  collect { … }
\_____ corre en Default _____/                    \_ corre donde recoges _/

Aparte, una regla: no uses withContext dentro de un flow { } para cambiar de hilo, porque falla en ejecución con una IllegalStateException. Para eso está flowOn.

Error típico: creer que el flow corre una sola vez

Guardas la narración en una variable y la recoges dos veces, esperando que la segunda sea «gratis». (PartidoEnVivo y narracion() los escribirás en el paso de abajo.)

val narracion = enVivo.narracion()
narracion.collect { println("  $it") }                                // un partido completo
val ms = measureTimeMillis { narracion.collect { } }                  // ¡otro partido completo!
println("Segundo collect: $ms ms")

No da error. El segundo collect vuelve a jugar el partido entero: tarda algo más de un segundo, casi lo mismo que el primero (la cifra exacta varía de una ejecución a otra y depende de tu máquina; en el punto de control verás una medida). Si el flow hiciera una llamada de red, harías dos llamadas.

val lineas = enVivo.narracion().toList()      // se juega el partido una vez
lineas.forEach { println("  $it") }
println("Otra vez, sin volver a jugar: ${lineas.size} líneas")

Si necesitas repartir un mismo flujo entre varios espectadores sin repetirlo, hace falta un flujo caliente: es el tema de StateFlow y SharedFlow.

En el módulo de Swift verás su equivalente, AsyncSequence y AsyncStream: for await recoge los valores como aquí lo hace collect.

Creada
(arranca la cuenta atrás)
(arranca la cuenta atrás)
(arranca la cuenta atrás)
par 4
par 2
(arranca la cuenta atrás)
Primero: 5

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun cuentaAtras(desde: Int): Flow<Int> = flow {
    println("(arranca la cuenta atrás)")
    for (n in desde downTo 1) {
        delay(100)
        emit(n)
    }
}
 
fun main() = runBlocking {
    val cuenta = cuentaAtras(5)
    println("Creada")                              // El aviso NO sale: es frío
 
    cuenta.collect { }                             // El aviso sale...
    cuenta.collect { }                             //    ...y sale otra vez: dos veces en total
 
    cuenta.filter { it % 2 == 0 }
        .map { "par $it" }
        .collect { println(it) }                   // par 4, par 2
 
    println("Primero: ${cuenta.first()}")          // 5, en unos 100 ms: corta el resto
}

Paso · El partido como flujo de eventos

Objetivo. Crear PartidoEnVivo.kt, con eventos() como un Flow, y tres derivados: narracion(), goles() y primerGol(). Después, reescribir jugarPartido encima de eventos(). Al terminar, ./gradlew :torneo:run narra los cinco eventos del partido y demuestra que el segundo collect vuelve a jugarlo.

Esta lección no instala nada: Flow y sus operadores (kotlinx.coroutines.flow) vienen en la misma librería de corrutinas que añadiste en Corrutinas y suspend.

Qué hacer en cada fichero (el código completo está más abajo):

1. Crea PartidoEnVivo.kt. En torneo → kotlin+java, clic derecho sobre el paquete com.sebastiannarvaez.marcador.torneo → New → Kotlin Class/File → File → escribe PartidoEnVivo y pulsa Intro. Dentro, class PartidoEnVivo(val guion: Partido, val msPorMinuto: Long = 10): guion es el partido con todo lo que va a pasar y msPorMinuto cuánto dura un minuto de juego en tiempo real. Los tests dependen de esa forma.
2. Abre Simulacion.kt (doble clic en su nodo, en el mismo paquete): añade el import kotlinx.coroutines.flow.fold junto a los de arriba y sustituye la función jugarPartido, con el comentario que lleva encima, por la nueva (con el comentario ya ajustado para ti). El resto del fichero no cambia.
3. Abre Main.kt y haz dos cambios: la llamada demoFlowFrio() al final del bloque runBlocking de main, tras demoErrores(), y la función demoFlowFrio() justo debajo de la llave que cierra main. Escribe la versión de más abajo.
4. El test se copia con el botón, al final.

Cosas que verás y conviene aclarar:

• filterIsInstance<Gol>() en lugar de filter { it is Gol }: se queda con los elementos de ese tipo y cambia el tipo del flow de Flow<EventoDePartido> a Flow<Gol>. Con filter { it is Gol }, seguiría siendo Flow<EventoDePartido>.
• fold, el mismo de las listas de Colecciones, ahora sobre un flujo, es un operador terminal que acumula: recibe un valor inicial (un Partido vacío) y, por cada evento, una lambda { jugado, evento -> … } que devuelve el acumulado nuevo. Aquí es lo que reconstruye el Partido.
• first() recoge el primer valor y cancela el resto del flujo: el partido no se sigue jugando por nada.
• jugarPartido cambia por dentro y no por fuera. Los tests de las lecciones anteriores siguen en verde: es una refactorización.
• emit dentro de forEach { emit(it) } funciona sin más: entrega cada elemento de la lista, uno tras otro.

Piezas nuevas de este paso: flow { … emit(x) … } dentro de una clase, filterIsInstance, map(::funcion) (una referencia a función, que pasa describir de Sealed: estados con datos como lambda), los terminales collect, toList, first y fold(inicial) { acumulado, elemento -> … }. (measureTimeMillis se vio en Corrutinas y suspend.) Todas en una estación del tiempo:

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.filterIsInstance
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.fold
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.runBlocking
import kotlin.system.measureTimeMillis
 
sealed interface Aviso { val hora: Int }
data class Lluvia(override val hora: Int, val litros: Int) : Aviso
data class Viento(override val hora: Int) : Aviso
 
fun leer(aviso: Aviso): String = "${aviso.hora}h: $aviso"
 
class Estacion(private val avisos: List<Aviso>, private val msPorHora: Long = 10) {
    // flow { … }: no corre al llamar a la función, sino al recogerlo
    fun avisos(): Flow<Aviso> = flow {
        for (hora in 0..5) {
            delay(msPorHora)                                    // una hora de reloj
            avisos.filter { it.hora == hora }.forEach { emit(it) }   // emit dentro de forEach: entrega cada aviso de esa hora
        }
    }
    fun lluvias(): Flow<Lluvia> = avisos().filterIsInstance<Lluvia>()   // filtra Y cambia el tipo a Flow<Lluvia>
    fun lectura(): Flow<String> = avisos().map(::leer)       // ::leer: la función leer pasada como lambda
}
 
fun main() = runBlocking {
    val estacion = Estacion(listOf(Lluvia(1, 3), Viento(2), Lluvia(4, 5)))
 
    estacion.lectura().collect { println(it) }               // collect: recoge cada valor
    println(estacion.lluvias().toList())                     // toList: recoge todos en una lista
    println(estacion.avisos().first())                       // first: el primero, y cancela el resto
    // fold: valor inicial 0 y, por cada lluvia, el acumulado nuevo
    val total = estacion.lluvias().fold(0) { litros, lluvia -> litros + lluvia.litros }
    println("Litros: $total")
    val ms = measureTimeMillis { estacion.avisos().toList() }   // cada recogida vuelve a correr el flow
    println("Otra recogida, otras 6 horas: ${ms >= 60}")
}

Al ejecutarlo, imprime:

1h: Lluvia(hora=1, litros=3)
2h: Viento(hora=2)
4h: Lluvia(hora=4, litros=5)
[Lluvia(hora=1, litros=3), Lluvia(hora=4, litros=5)]
Lluvia(hora=1, litros=3)
Litros: 8
Otra recogida, otras 6 horas: true

Intenta escribirlo tú, en este orden:

1. eventos() entrega los eventos del guion uno a uno, a medida que pasan los minutos: los del minuto 0 al empezar y, después, minuto a minuto con su espera, como hacía jugarPartido. Corre en el grupo de hilos Default.
2. Tres derivados: narracion() entrega cada evento ya descrito como texto; goles() entrega solo los goles; primerGol() espera al primer gol, lo devuelve y no deja que el partido siga jugándose.
3. jugarPartido, por dentro, pasa a construir el partido jugado acumulando los eventos de eventos(); por fuera no cambia, y sus tests siguen en verde.
4. demoFlowFrio() prepara la narración del Rayo FC-Toros y avisa de que aún no ha corrido nada; la recoge mostrando cada línea; la vuelve a recoger, la misma, midiendo cuánto tarda; y por último muestra el primer gol.

Tras las demos anteriores, debe verse (los milisegundos varían; lo importante es que el aviso sale antes de la narración y que el segundo recorrido vuelve a durar más de un segundo):

== un Flow frío, la narración del partido ==
Flow creado; nada ha corrido todavía
  12' Gol de Ana (Rayo FC)
  30' Tarjeta amarilla para Iván
  55' Gol de Iván (Toros)
  80' Gol de Ana (Rayo FC)
  85' Cambio: sale Ana, entra Luis
Segundo collect de la misma variable: otro partido completo (1284 ms)
Primer gol: 12' Gol de Ana (Rayo FC) (el resto del partido se canceló)

Cuando termines, abre el bloque y compara.

package com.sebastiannarvaez.marcador.torneo
 
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.filterIsInstance
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.flowOn
import kotlinx.coroutines.flow.map
 
// Un partido que se está jugando. `guion` es todo lo que va a pasar.
class PartidoEnVivo(val guion: Partido, val msPorMinuto: Long = 10) {
 
    // Un Flow FRÍO: `flow { ... }` no hace nada hasta que alguien lo recoge con
    // `collect` (o `first`, `toList`, `fold`…). Y cada `collect` ejecuta el bloque
    // ENTERO otra vez, desde el principio: dos espectadores = dos partidos
    // independientes. `emit` entrega un valor y suspende hasta que el que recoge
    // lo ha procesado.
    //
    // `flowOn` cambia el dispatcher de lo que está ANTES (el `flow { }` y sus
    // operadores anteriores); quien recoge sigue en su propio contexto.
    fun eventos(): Flow<EventoDePartido> = flow {
        guion.eventos.filter { it.minuto == 0 }.forEach { emit(it) }
        val ultimoMinuto = maxOf(MINUTOS_DEL_PARTIDO, guion.eventos.maxOfOrNull { it.minuto } ?: 0)
        for (minuto in 1..ultimoMinuto) {
            delay(msPorMinuto)
            guion.eventos.filter { it.minuto == minuto }.forEach { emit(it) }
        }
    }.flowOn(Dispatchers.Default)
 
    // Operadores: cada uno devuelve OTRO Flow, y sigue sin ejecutarse nada hasta
    // que se recoja el resultado.
    fun narracion(): Flow<String> = eventos().map(::describir)
 
    fun goles(): Flow<Gol> = eventos().filterIsInstance<Gol>()
 
    // `first()` es un operador terminal que se queda con el primer valor y CANCELA
    // el resto del flujo: el partido no se sigue jugando por nada.
    suspend fun primerGol(): Gol = goles().first()
}

import kotlinx.coroutines.flow.fold

// `suspend` NO significa «corre en otro hilo». Significa «esta función puede
// pausarse»: dentro puede llamar a otras `suspend` (como `delay`) y, mientras
// espera, el hilo queda libre para hacer otra cosa. Solo se puede llamar desde
// otra `suspend` o desde una corrutina.
//
// `partido` es el GUION: todo lo que va a pasar. La función lo «juega» minuto a
// minuto y devuelve el partido tal como quedó. `msPorMinuto` es cuánto dura un
// minuto de juego en tiempo real (10 ms hace un partido de un segundo).
//
// Ahora el minuto a minuto vive en `PartidoEnVivo.eventos()` (un Flow) y
// aquí solo se acumula lo que emite: `fold` parte de un partido vacío y va
// registrando cada evento.
suspend fun jugarPartido(partido: Partido, msPorMinuto: Long = 10): Partido =
    PartidoEnVivo(partido, msPorMinuto).eventos()
        .fold(Partido(partido.local, partido.visitante)) { jugado, evento -> jugado.registrar(evento) }

        println("== un Flow frío, la narración del partido ==")
        demoFlowFrio()

suspend fun demoFlowFrio() {
    val enVivo = PartidoEnVivo(Ejemplo.rayoContraToros)
    // Aún no ha pasado nada: crear el Flow no arranca el partido.
    val narracion = enVivo.narracion()
    println("Flow creado; nada ha corrido todavía")
    narracion.collect { println("  $it") }
 
    // La MISMA variable, recogida otra vez: el frío se repite entero.
    val ms = measureTimeMillis { narracion.collect { } }
    println("Segundo collect de la misma variable: otro partido completo ($ms ms)")
 
    val gol = enVivo.primerGol()
    println("Primer gol: ${describir(gol)} (el resto del partido se canceló)")
}

El test: se copia con el botón

FlowFrioTest.kt (6 pruebas: orden, narración, solo goles, frío, first y el primer gol) comprueba lo que acabas de escribir y se copia con el botón Copiar fichero, como en Clases y objetos.

1. En torneo → kotlin+java, clic derecho sobre el paquete com.sebastiannarvaez.marcador.torneo (test) → New → Kotlin Class/File → File → escribe FlowFrioTest y pulsa Intro.
2. Pulsa Copiar fichero en el bloque de abajo.
3. En el editor, selecciona todo y pega: Cmd+A / Cmd+V. El texto copiado trae su propia línea package.

El test usa AtomicInteger (un contador seguro entre hilos) y onEach (anota cada elemento que pasa); no hace falta dominarlos, solo copiarlo.

== un Flow frío, la narración del partido ==
Flow creado; nada ha corrido todavía
  12' Gol de Ana (Rayo FC)
  30' Tarjeta amarilla para Iván
  55' Gol de Iván (Toros)
  80' Gol de Ana (Rayo FC)
  85' Cambio: sale Ana, entra Luis
Segundo collect de la misma variable: otro partido completo (1284 ms)
Primer gol: 12' Gol de Ana (Rayo FC) (el resto del partido se canceló)