stateIn, shareIn y los flujos calientes

Al terminar esta lección Marcador tendrá dos partidos a la vez que todos los espectadores ven por igual: uno pinta los marcadores, otro avisa de cada gol y un tercero cuenta los minutos, y los tres se paran cuando suena el pitido final. Para eso aprendes a convertir un flujo frío en uno caliente con stateIn y shareIn, y a decidir cuándo se pone en marcha con WhileSubscribed. Tampoco aquí instalas nada: todo sale de la librería de corrutinas que ya tienes en :torneo.

Compartir un flujo frío: stateIn y shareIn

A veces ya tienes un flujo frío y lo quieres compartir: que se ejecute una sola vez y que todos reciban lo mismo. stateIn lo convierte en StateFlow y shareIn en SharedFlow, dentro de un alcance de corrutinas (la zona de la que cuelga el trabajo y que decide cuándo se cancela: Alcance, Job y cancelación). El parámetro started decide cuándo se pone en marcha el flujo de origen.

Antes de los ejemplos, una línea que verás en ellos: CoroutineScope(coroutineContext + Job()). Dentro de runBlocking, coroutineContext es el contexto de la corrutina actual (su Job y su dispatcher: lo que decide cómo corre y quién la cancela); el + le añade un Job() nuevo, y el resultado es un alcance propio que puedes cancelar con alcance.cancel() sin tocar a runBlocking.

Primero shareIn, con una radio. El flujo radio() es frío: cada vez que alguien lo recoge, se enciende la radio y suenan cuatro canciones. Compartido con Eagerly, se enciende una vez, ya, y quien llega tarde oye lo que suene desde que llega:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
// Un flujo FRÍO: no hace nada hasta que alguien lo recoge.
fun radio(): Flow<String> = flow {
    println("(se enciende la radio)")          // se imprime cada vez que alguien arranca el flujo
    for (n in 1..4) {
        delay(100)
        emit("canción $n")
    }
}
 
fun main() = runBlocking {
    val alcance = CoroutineScope(coroutineContext + Job())     // un alcance propio, que cancelamos al final
    // shareIn: lo vuelve SharedFlow. Eagerly = arranca ya, aunque no haya nadie escuchando
    val emisora = radio().shareIn(alcance, SharingStarted.Eagerly)
 
    delay(250)                                                  // suenan dos canciones sin oyentes
    val oyente = launch { emisora.collect { println("oigo: $it") } }
    delay(250)
    oyente.cancel()                                             // sin cancelar, el oyente no terminaría nunca
    alcance.cancel()
}

Al ejecutarlo en el Playground sale:

(se enciende la radio)
oigo: canción 3
oigo: canción 4

La radio se encendió una vez, las canciones 1 y 2 sonaron sin nadie y el oyente solo oyó la 3 y la 4. Ahora stateIn, que además pide un valor inicial (un StateFlow siempre tiene valor) y se lee con value. Aquí con WhileSubscribed, que arranca el origen solo cuando alguien mira:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun termometro(): Flow<Int> = flow {
    println("(arranca el termómetro)")
    for (grados in 20..22) {
        delay(100)
        emit(grados)
    }
}
 
fun main() = runBlocking {
    val alcance = CoroutineScope(coroutineContext + Job())
    // stateIn: lo vuelve StateFlow. Necesita un valor inicial (0) por si nadie ha medido todavía.
    // WhileSubscribed: solo mide mientras alguien mira.
    val temperatura = termometro().stateIn(alcance, SharingStarted.WhileSubscribed(), 0)
 
    println("Sin mirar: ${temperatura.value}")               // nadie mira: no ha arrancado
    val lector = launch { temperatura.collect { } }           // alguien mira: arranca el termómetro
    delay(250)
    println("Mirando: ${temperatura.value}")
    lector.cancel()
    alcance.cancel()
}

Sin mirar: 0
(arranca el termómetro)
Mirando: 21

Las tres políticas de started:

Los 5 segundos existen para que un corte breve no reinicie el origen. Un ejemplo de Android, que verás más adelante: al girar el móvil la pantalla se recrea, y la suscripción se corta un instante y vuelve. Con 5 s de margen, el flujo no se entera.

Error típico: WhileSubscribed sobre la fuente

// El partido, compartido con stateIn y WhileSubscribed.
val marcador = enVivo.marcadores()
    .stateIn(alcance, SharingStarted.WhileSubscribed(5_000), Marcador(0, 0))

// La fuente se recoge UNA vez y alimenta un MutableStateFlow.
private val marcadorActual = MutableStateFlow(Marcador(0, 0))
val marcador: StateFlow<Marcador> = marcadorActual.asStateFlow()
 
fun iniciar(alcance: CoroutineScope): Job = alcance.launch {
    eventos().collect { evento -> /* actualiza marcadorActual */ }
}

Falla por cómo funcionan los flujos fríos: marcadores() es un flow { } y cada vez que se vuelve a recoger, el partido empieza de cero. Si la pantalla se queda sin espectadores más de 5 s y luego vuelve alguien, WhileSubscribed habría parado la recogida y la reanudaría con un partido nuevo. WhileSubscribed es para lo que se puede reiniciar sin perder nada, como combinar dos StateFlow; no para la fuente de la verdad.

Por eso Marcador lo reparte así: el partido lo mueve iniciar(alcance), que recoge eventos() una vez, y WhileSubscribed va en Jornada, que solo combina flujos ya vivos. Si nadie mira la jornada, no calcula nada; cuando alguien mira, se pone al día con el estado actual.

Las piezas de la demo: merge, UNDISPATCHED y SupervisorJob

• merge(a, b) junta varios flujos en uno, en el orden en que llegan los valores. combine los empareja (un resultado con el último de cada uno); merge los mezcla (todos los valores, uno detrás de otro).
• launch(start = CoroutineStart.UNDISPATCHED). Normalmente, una corrutina recién lanzada no arranca al instante: espera su turno en la cola de su dispatcher, y quien la lanzó sigue con lo suyo. Con UNDISPATCHED arranca ya, y corre hasta su primera espera (aquí, el collect, que se queda esperando valores). Es lo que deja a cada espectador suscrito antes de que empiece el partido: si no, el partido podría emitir el primer gol mientras el espectador aún espera en la cola, y se lo perdería.
• CoroutineScope(SupervisorJob() + Dispatchers.Default) crea un alcance propio, igual que el de antes pero con el dispatcher Default (un grupo de hilos) y un SupervisorJob, que hace que una hija que falla no cancele a las demás (Errores en corrutinas).

Como los espectadores corren en Dispatchers.Default (un grupo de hilos), aquí sí hay paralelismo: el reloj, el aviso de goles y la pantalla pueden ejecutarse en el mismo instante en núcleos distintos. Por eso el orden de las líneas que ocurren a la vez puede variar de una ejecución a otra.

Y ahora, a practicar lo importante de la lección: compartir un flujo para que se ejecute una sola vez.

(arranca el sensor)
Ana ve 22
Leo ve 22
Sin compartir:
(arranca el sensor)
(arranca el sensor)
Ana ve 22, Leo ve 22

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun sensor(): Flow<Int> = flow {
    println("(arranca el sensor)")
    for (grados in 20..22) { delay(100); emit(grados) }
}
 
fun main() = runBlocking {
    val alcance = CoroutineScope(coroutineContext + Job())   // un alcance propio para compartir
    val compartido = sensor().stateIn(alcance, SharingStarted.WhileSubscribed(), 0)
    val ana = async { compartido.first { it == 22 } }
    val leo = async { compartido.first { it == 22 } }
    println("Ana ve ${ana.await()}")
    println("Leo ve ${leo.await()}")
    alcance.cancel()
 
    println("Sin compartir:")
    val a = async { sensor().first { it == 22 } }
    val b = async { sensor().first { it == 22 } }
    println("Ana ve ${a.await()}, Leo ve ${b.await()}")
}

Paso · Dos partidos, muchos espectadores

Objetivo. Nace Jornada en PartidoEnVivo.kt, con stateIn y shareIn; Main.kt gana la demo demoEnVivo; y se copia FlowsCalientesTest. Al terminar, ./gradlew :torneo:run muestra dos partidos a la vez, con goles, marcadores y reloj, y todo se cancela al pitar el final.

Dónde va cada cosa. Despliega torneo → kotlin+java → com.sebastiannarvaez.marcador.torneo y abre cada fichero con doble clic:

1. PartidoEnVivo.kt: los imports arriba y Jornada al final del fichero, fuera de la clase.
2. Main.kt: cuatro imports, la llamada a demoEnVivo() dentro de main después de la última demo que hay (la de la jornada) y la función demoEnVivo() debajo de main.

Piezas nuevas de Jornada(primero, segundo, alcance): marcadores es un StateFlow<MarcadoresDeJornada> (combine + stateIn), goles es un SharedFlow<Gol> (merge + shareIn) e iniciar() lanza los dos partidos. En iniciar(), alcance.launch { … } abre una corrutina y, dentro de las llaves, this es el alcance de esa corrutina: primero.iniciar(this) lanza el partido como hija suya, y por eso su Job termina cuando terminan las dos hijas (Alcance, Job y cancelación).

Intenta escribirlo tú:

1. Jornada: junta dos partidos en vivo. Sus marcadores son los dos marcadores juntos y solo se calculan mientras alguien mira (con 5 s de margen); sus goles mezclan los de los dos partidos y también llevan WhileSubscribed con 5 s de margen; iniciar() lanza los dos.
2. La demo, demoEnVivo(): en un alcance propio, pone en marcha un reloj que avisa cada 30 minutos, un espectador de goles y una pantalla con los dos marcadores, todos suscritos antes de que empiece el partido; inicia la jornada y, al terminar, cancela a los tres espectadores, comprueba que quedaron cancelados y muestra los marcadores finales.

Tras las demos anteriores, debe verse algo así (los goles y marcadores que ocurren en el mismo instante pueden salir en otro orden):

== dos partidos simultáneos con StateFlow y SharedFlow ==
  Rayo FC 0-0 Toros | Lobos 0-0 Águilas
  ⚽ 12' Gol de Ana (Rayo FC)
  Rayo FC 1-0 Toros | Lobos 0-0 Águilas
  ⚽ 20' Gol de Pedro (Lobos)
  Rayo FC 1-0 Toros | Lobos 1-0 Águilas
  ⏱ minuto 30
  ⚽ 33' Gol de Nico (Águilas)
  Rayo FC 1-0 Toros | Lobos 1-1 Águilas
  ⚽ 55' Gol de Iván (Toros)
  Rayo FC 1-1 Toros | Lobos 1-1 Águilas
  ⏱ minuto 60
  ⚽ 70' Gol de Nico (Águilas)
  Rayo FC 1-1 Toros | Lobos 1-2 Águilas
  ⚽ 80' Gol de Ana (Rayo FC)
  Rayo FC 2-1 Toros | Lobos 1-2 Águilas
  ⏱ minuto 90
Pitido final: se cancelan el reloj y los espectadores
Cancelados: [true, true, true]
Marcadores finales: 2-1 y 1-2

Abre el bloque y compara.

El título del println que encabeza la demo de Main.kt puedes escribirlo como prefieras.

// Los dos partidos de la jornada, con sus flujos calientes derivados.
//
// `stateIn` y `shareIn` convierten un flujo en StateFlow / SharedFlow dentro de
// un alcance. `WhileSubscribed(5_000)` significa: el flujo de origen solo se
// recoge mientras haya suscriptores, y sigue 5 s después de que el último se
// vaya (para no reiniciarlo en una rotación de pantalla, por ejemplo).
// Aquí es apropiado porque el origen (`combine` sobre StateFlows) se puede
// reiniciar sin perder nada. NO lo sería para `eventos()`, que empezaría otro
// partido de cero.
class Jornada(val primero: PartidoEnVivo, val segundo: PartidoEnVivo, private val alcance: CoroutineScope) {
    val marcadores: StateFlow<MarcadoresDeJornada> =
        combine(primero.marcador, segundo.marcador, ::MarcadoresDeJornada)
            .stateIn(
                scope = alcance,
                started = SharingStarted.WhileSubscribed(stopTimeoutMillis = 5_000),
                initialValue = MarcadoresDeJornada(Marcador(0, 0), Marcador(0, 0)),
            )
 
    val goles: SharedFlow<Gol> =
        merge(primero.goles, segundo.goles)
            .shareIn(alcance, SharingStarted.WhileSubscribed(stopTimeoutMillis = 5_000))
 
    // Los dos partidos a la vez. El Job termina cuando terminan los dos.
    fun iniciar(): Job = alcance.launch {
        primero.iniciar(this)
        segundo.iniciar(this)
    }
}

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.CoroutineStart
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
 
    …
 
        println("== dos partidos simultáneos con StateFlow y SharedFlow ==")
        demoEnVivo()
 
    …
 
// Los dos partidos a la vez, con la pantalla (marcadores) y el aviso de goles
// como espectadores, y un cronómetro. Al pitar el final se cancela todo: los
// espectadores son corrutinas infinitas y, si nadie las cancela, el programa no
// terminaría nunca.
suspend fun demoEnVivo() {
    val alcance = CoroutineScope(SupervisorJob() + Dispatchers.Default)
    val jornada = Jornada(PartidoEnVivo(Ejemplo.rayoContraToros), PartidoEnVivo(Ejemplo.lobosContraAguilas), alcance)
 
    val reloj = alcance.cronometro(msPorMinuto = 10) { minuto -> if (minuto % 30 == 0) println("  ⏱ minuto $minuto") }
    val goles = alcance.launch(start = CoroutineStart.UNDISPATCHED) {
        jornada.goles.collect { println("  ⚽ ${describir(it)}") }
    }
    val pantalla = alcance.launch(start = CoroutineStart.UNDISPATCHED) {
        jornada.marcadores.collect { println("  ${Ejemplo.rayo.nombre} ${it.primero} ${Ejemplo.toros.nombre} | ${Ejemplo.lobos.nombre} ${it.segundo} ${Ejemplo.aguilas.nombre}") }
    }
 
    jornada.iniciar().join()
    println("Pitido final: se cancelan el reloj y los espectadores")
    val espectadores = listOf(reloj, goles, pantalla)
    espectadores.forEach { it.cancelAndJoin() }
    println("Cancelados: ${espectadores.map { it.isCancelled }}")
    println("Marcadores finales: ${jornada.marcadores.value.primero} y ${jornada.marcadores.value.segundo}")
    alcance.cancel()
}

Ahora el test, con el botón Copiar fichero (explicado en Clases y objetos: copia el contenido al portapapeles; el fichero lo creas tú). El orden es siempre el mismo:

1. En torneo → kotlin+java, clic derecho sobre el paquete com.sebastiannarvaez.marcador.torneo (test) → New → Kotlin Class/File → File → FlowsCalientesTest. Se abre con solo la línea package.
2. Pulsa el botón Copiar fichero de abajo. Aparecerá un aviso de que se ha copiado.
3. Haz clic dentro de FlowsCalientesTest.kt, selecciona todo con Cmd+A y pega con Cmd+V. El contenido pegado ya trae su línea package, así que sustituye la que puso el IDE.

Son nueve pruebas. Usan tres piezas que aún no has visto: flujo.take(n) recoge solo los n primeros valores y termina (como el take de las listas); withTimeoutOrNull(ms) { … } devuelve null si el bloque no termina a tiempo; y alcance.coroutineContext[Job]!! le pide al alcance su Job por su tipo, para cancelarlo (el !! afirma que no es null). El test ya viene escrito: basta con que sepas leerlo por encima. La que enseña WhileSubscribed es esta, tal cual sale en el fichero:

    @Test
    fun conWhileSubscribedLaJornadaNoSeCalculaSinEspectadores() = runBlocking {
        val alcance = alcance()
        val jornada = Jornada(PartidoEnVivo(Ejemplo.rayoContraToros, 1), PartidoEnVivo(Ejemplo.lobosContraAguilas, 1), alcance)
        jornada.iniciar().join()
        // Los partidos acabaron, pero nadie miraba: el combine ni arrancó.
        assertEquals(MarcadoresDeJornada(Marcador(0, 0), Marcador(0, 0)), jornada.marcadores.value)
        // En cuanto alguien mira, se reactiva y ve lo de ahora.
        val visto = withTimeoutOrNull(1_000) {
            jornada.marcadores.first { it.golesTotales == 6 }
        }
        assertEquals(MarcadoresDeJornada(Marcador(2, 1), Marcador(1, 2)), visto)
        alcance.cancel()
    }

Se lee por partes. alcance() es una función del propio fichero que crea un alcance nuevo para cada prueba. jornada.iniciar().join() juega los dos partidos enteros y espera a que terminen, sin ningún espectador. Como Jornada.marcadores usa WhileSubscribed, el combine de dentro no ha arrancado nunca, y value sigue siendo el valor inicial, 0-0 y 0-0. Después, first { … } es un espectador que recoge hasta que se cumple la condición (aquí, seis goles en total); withTimeoutOrNull(1_000) le da como máximo un segundo y devuelve null si no lo consigue. Al aparecer un espectador, el combine arranca, se pone al día con el estado actual de los dos StateFlow y ve 2-1 y 1-2.

Ejecuta los tests: en la Terminal de Android Studio, ./gradlew :torneo:test. Debe terminar en BUILD SUCCESSFUL.