StateFlow y SharedFlow

Al terminar esta lección Marcador tendrá un partido que no depende de quién lo mire: guardará su marcador en un valor que cualquiera puede leer y avisará de cada gol a quien esté escuchando. Todavía no habrá nada nuevo en pantalla: es la pieza que la lección siguiente (stateIn, shareIn y los flujos calientes) usa para que dos partidos se vean a la vez.

Hasta ahora cada espectador recibía su propio partido, porque un flow { } es frío. Aquí aprendes los dos flujos que existen aunque nadie mire: StateFlow y SharedFlow. Tampoco instalas nada: todo sale de la librería de corrutinas que ya tienes en :torneo.

Flujos calientes: StateFlow y SharedFlow

Un flujo caliente existe tenga o no suscriptores: no arranca cuando alguien lo recoge, ya está en marcha. Un suscriptor (en la documentación de Kotlin, un colector) es simplemente el código que llama a collect sobre el flujo; en esta lección las dos palabras significan lo mismo. Kotlin trae dos flujos calientes, para dos necesidades distintas: un estado (algo que tiene un valor ahora, como el marcador) y un evento (algo que ocurre en un instante, como un gol).

Los dos se crean mutables (que se pueden cambiar) para poder emitir y se exponen de solo lectura para que nadie de fuera emita:

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
class Reloj {
 
    // Quien tiene el Reloj puede LEER el minuto, pero no cambiarlo.
    private val _minuto = MutableStateFlow(0)
    val minuto: StateFlow<Int> = _minuto.asStateFlow()
 
    // Un evento suelto: sin valor actual y sin repetición.
    private val _pitidos = MutableSharedFlow<String>()
    val pitidos: SharedFlow<String> = _pitidos.asSharedFlow()
 
    suspend fun avanzar() {
        _minuto.update { it + 1 }       // cambio atómico
        _pitidos.emit("minuto ${_minuto.value}")
    }
}
 
fun main() = runBlocking {
    val r = Reloj()                  // un reloj nuevo: el minuto vale 0
    r.avanzar()                      // el minuto pasa a 1 (nadie escucha los pitidos)
    println(r.minuto.value)          // lee el valor actual desde fuera
}

Al ejecutarlo en el Playground sale:

1

Fíjate en el par _minuto / minuto: por convención, el nombre con guion bajo es la versión privada y mutable y el nombre normal es la que ven los demás, de solo lectura. asStateFlow() y asSharedFlow() hacen esa conversión.

update { it + 1 } cambia el valor de forma atómica, es decir, de un solo golpe: si dos corrutinas lo llaman a la vez, ninguna pisa a la otra. (it es el valor de ahora; lo que devuelve la llave es el valor nuevo.) Y asignar un valor igual al actual no emite nada: por eso un StateFlow no repite 1-0 dos veces.

Ninguno de los dos deja de recoger por su cuenta: quien lo recoge sigue suscrito hasta que alguien cancele su corrutina. Un collect sobre un StateFlow o un SharedFlow no termina nunca por sí solo.

Un ejemplo completo: una caja registradora

Los dos flujos juntos, en una clase pequeña. La caja guarda el total de lo vendido (un estado) y avisa de cada venta grande (un evento). Léelo línea a línea:

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asSharedFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
 
class Caja(private val nombre: String, private val ventas: List<Int>) {
    private var abierta = false
    private val _total = MutableStateFlow(0)                        // estado: siempre tiene valor
    val total: StateFlow<Int> = _total.asStateFlow()
    // extraBufferCapacity = 16: hasta 16 avisos pendientes sin que emit se suspenda
    private val _avisos = MutableSharedFlow<String>(extraBufferCapacity = 16)
    val avisos: SharedFlow<String> = _avisos.asSharedFlow()
 
    // Recoge la fuente UNA vez y alimenta los dos flujos
    fun abrir(alcance: CoroutineScope): Job {
        check(!abierta) { "La caja ya está abierta" }               // si es false, IllegalStateException
        abierta = true
        return alcance.launch {
            flow { ventas.forEach { delay(20); emit(it) } }.collect { euros ->
                _total.update { it + euros }                        // cambio atómico
                if (euros >= 50) _avisos.emit("$nombre: venta grande de $euros")
            }
        }
    }
}
 
fun main() = runBlocking {
    val norte = Caja("Norte", listOf(10, 60))
    println("Antes de abrir: ${norte.total.value}")
    // `this` es el alcance de runBlocking: la caja lanza su corrutina dentro de él
    norte.abrir(this).join()                  // abre la caja y espera a que termine
    println("Después de abrir: ${norte.total.value}")
    try {
        norte.abrir(this)                     // ya estaba abierta: check falla
        println("esto no debería verse")
    } catch (e: IllegalStateException) {
        println(e.message)
    }
}

Al ejecutarlo en el Playground sale:

Antes de abrir: 0
Después de abrir: 70
La caja ya está abierta

Tres cosas nuevas:

• abrir recoge la fuente una sola vez y alimenta a los dos flujos: el total sube con cada venta y, si la venta es grande, sale un aviso. Nadie escucha los avisos en este main, y no pasa nada: un SharedFlow sin oyentes los deja pasar.
• MutableSharedFlow(extraBufferCapacity = 16) reserva sitio para 16 avisos pendientes, de modo que emit no se queda esperando si un oyente va lento. replay sigue en 0: quien llega tarde no ve los avisos anteriores.
• check(condición) { "mensaje" } lanza un IllegalStateException con ese mensaje si la condición es falsa (lo viste en Excepciones: try, catch y require). Aquí impide abrir la misma caja dos veces, que duplicaría las ventas.

En la lección siguiente pondrás dos cajas a la vez y sumarás sus totales. Ahora, a practicar.

marcador: 2

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
 
fun main() = runBlocking {
    val marcador = MutableStateFlow(0)
    val avisos = MutableSharedFlow<String>()
 
    marcador.update { 1 }     // la llave devuelve el valor nuevo
    marcador.update { 1 }
    marcador.update { 2 }
    avisos.emit("gol")       // sin suscriptores: nadie lo oye
    avisos.emit("tarjeta")
 
    val a = launch { marcador.collect { println("marcador: $it") } }
    val b = launch { avisos.collect { println("aviso: $it") } }
    delay(100)
    a.cancel(); b.cancel()
}

Paso · Un partido que se comparte

Objetivo. PartidoEnVivo gana marcador (un StateFlow), goles (un SharedFlow) e iniciar; Marcador.kt gana despuesDe; y goles() pasa a llamarse golesEnFrio(). No se ve nada nuevo al ejecutar la app, y hoy no hay tests nuevos: solo compruebas que compila y que los tests de antes siguen en verde. Los tests de estas piezas llegan en la lección siguiente.

Todo se hace sobre ficheros que ya tienes. Dos decisiones a tener en cuenta:

• goles() pasa a golesEnFrio(). Ahora PartidoEnVivo tendrá una propiedad goles (el SharedFlow, caliente). Si la función de ahora siguiera llamándose goles(), dos cosas muy distintas (un flujo frío y uno caliente) se distinguirían solo por unos paréntesis, y es muy fácil leer o escribir una por la otra. Con golesEnFrio() el nombre ya dice qué es. Es el único cambio a algo ya escrito, y afecta a dos sitios: primerGol y FlowFrioTest.
• despuesDe se extrae de Marcador.kt para que el flujo frío (marcadores()) y el caliente (iniciar) calculen el marcador con la misma función.

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; los flujos calientes y iniciar al final de la clase, antes de la llave que la cierra.
2. Marcador.kt: despuesDe va antes de marcadores(), que pasa a usarla.
3. FlowFrioTest.kt (en el paquete que lleva (test)): una línea con el nombre nuevo.

Piezas nuevas de PartidoEnVivo:

• val marcador: StateFlow<Marcador>, con un MutableStateFlow(Marcador(0, 0)) privado detrás, y val goles: SharedFlow<Gol>, con un MutableSharedFlow<Gol> privado. Es el mismo patrón de la caja.
• fun iniciar(alcance: CoroutineScope): Job, que recoge eventos() una vez y alimenta los dos. Un CoroutineScope es el alcance, la «zona» de la que cuelgan las corrutinas y que decide cuándo se cancelan (Alcance, Job y cancelación); un Job es el mando de una corrutina.
• check(!iniciado) { "El partido ya empezó" }: como en la caja, impide llamar dos veces a iniciar, que duplicaría los goles.

Intenta escribirlo tú, en este orden:

1. El cambio de nombre: goles() pasa a golesEnFrio(), en PartidoEnVivo y en FlowFrioTest.
2. despuesDe(evento, partido): el marcador que queda tras un evento (un gol suma al equipo que marcó; cualquier otro evento lo deja igual). marcadores() pasa a usarlo.
3. Dentro de PartidoEnVivo: marcador es el estado actual del partido, que empieza en 0-0 y que todos pueden leer pero solo el partido cambia; goles avisa de cada gol a quien esté escuchando en ese momento, sin repetir los anteriores a quien llegue tarde. iniciar(alcance) juega el partido una sola vez, actualizando los dos, y se niega a empezar dos veces («El partido ya empezó»).

    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()

    fun golesEnFrio(): 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 = golesEnFrio().first()

    …
 
    // `eventos()` es frío: cada espectador ve SU partido. Un partido de verdad es
    // uno solo y todos ven lo mismo, y para eso hacen falta flujos calientes.
 
    // StateFlow: siempre tiene un valor (`value`), el estado actual, y quien se
    // suscribe recibe primero ese valor. Si llegan dos valores muy seguidos, un
    // espectador lento solo ve el último, y los valores iguales al anterior se
    // descartan. Es la elección para ESTADO que se pinta: el marcador.
    private val marcadorActual = MutableStateFlow(Marcador(0, 0))
    val marcador: StateFlow<Marcador> = marcadorActual.asStateFlow()
 
    // SharedFlow: un canal de EVENTOS. Sin valor actual y sin repetición
    // (`replay = 0`): quien se suscribe tarde NO ve los goles que ya pasaron, y
    // cada gol llega a todos los suscriptores, uno por uno. Es la elección para
    // cosas que ocurren (un gol, un aviso), no para estado.
    private val golesEmitidos = MutableSharedFlow<Gol>(extraBufferCapacity = 16)
    val goles: SharedFlow<Gol> = golesEmitidos.asSharedFlow()
 
    private var iniciado = false
 
    // Arranca el partido: recoge `eventos()` UNA vez y alimenta a los dos flujos
    // calientes. Devuelve el Job, que termina al pitar el final. Se ejecuta en el
    // alcance que le den, así que quien lo da decide cuándo se cancela todo.
    fun iniciar(alcance: CoroutineScope): Job {
        check(!iniciado) { "El partido ya empezó" }
        iniciado = true
        return alcance.launch {
            eventos().collect { evento ->
                if (evento is Gol) {
                    marcadorActual.update { it.despuesDe(evento, guion) }
                    golesEmitidos.emit(evento)
                }
            }
        }
    }
}

// El marcador tras un evento: solo un gol lo cambia.
fun Marcador.despuesDe(evento: EventoDePartido, partido: Partido): Marcador = when {
    evento !is Gol -> this
    evento.equipo == partido.local -> copy(local = local + 1)
    else -> copy(visitante = visitante + 1)
}
 
// `scan` es como `fold` pero emite cada paso: parte de 0-0 y va emitiendo el
// marcador tras cada evento. `distinctUntilChanged` deja pasar solo los cambios
// (una tarjeta no cambia el marcador, así que no emite nada).
fun PartidoEnVivo.marcadores(): Flow<Marcador> =
    eventos()
        .scan(Marcador(0, 0)) { marcador, evento -> marcador.despuesDe(evento, guion) }
        .distinctUntilChanged()
 

        assertEquals(3, enVivo.goles().toList().size)

        assertEquals(3, enVivo.golesEnFrio().toList().size)

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

La lección siguiente, stateIn, shareIn y los flujos calientes, pone dos partidos a la vez con estos flujos y llega a 127 tests.