Paso 12 / 16

Flows calientes: SharedFlow, StateFlow y el timeout WhileSubscribed

📖 Curso de lectura — copia los snippets de Kotlin a tu proyecto; los ejecutables abren en el Arena IDE.

LA IDEA CLAVE

StateFlow es un valor actual, SharedFlow es un bus de eventos, y el timeout de shareIn es la diferencia entre un websocket y uno por cada rotación de pantalla.

01 Entiende02 Explora el código03 Comprueba

Explorar la explicación completa 8 ideas

Un `Flow` frío corre una vez por colector: dos colectores significan dos ejecuciones independientes, dos consultas a la base, dos conexiones websocket. Ese es el default correcto para un pipeline de transformación y el equivocado para una fuente compartida. Los flows calientes existen para el segundo caso — corren **una vez**, independientemente de quién escuche, y transmiten a cada suscriptor activo. `SharedFlow` y `StateFlow` son los dos tipos calientes incorporados, y `shareIn`/`stateIn` son los operadores que convierten cualquier flow frío en uno.

`MutableSharedFlow<T>(replay, extraBufferCapacity, onBufferOverflow)` es el bus de eventos. Cada `emit` se entrega a cada colector activo; las últimas `replay` emisiones se guardan y se entregan a los colectores que llegan después. `replay = 0` es un feed de eventos puro — el suscriptor que no estaba se lo perdió, para siempre. Los parámetros de buffering importan más de lo que parece: `emit` **suspende** cuando el buffer está lleno y la política de desborde es `SUSPEND`, lo que significa que un colector lento aplica back-pressure sobre tu productor, y un suscriptor trabado puede frenar todo el bus. `tryEmit` más una política `DROP_OLDEST` es la forma que querés cuando el productor nunca debe ser bloqueado por un consumidor.

`StateFlow<T>` es otra cosa vestida con una interfaz parecida: es un **valor actual**, siempre presente, y tiene dos comportamientos que un `SharedFlow` no tiene. Está **conflado** — los valores intermedios de un escritor rápido nunca llegan a un colector que todavía no se reanudó, así que escribir `1, 2, 3` sin suspensión entremedio deja observable solo el `3`. Y **deduplica por igualdad** — setear el mismo valor dos veces emite una, porque `StateFlow` compara con `equals`. Es aproximadamente `SharedFlow(replay = 1, onBufferOverflow = DROP_OLDEST)` más `distinctUntilChanged` más un valor inicial obligatorio.

Esos dos comportamientos son la fuente de un bug clásico. Modelá un *evento* de una sola vez — «mostrá un toast», «navegá al checkout» — como un `StateFlow` y dos eventos idénticos seguidos disparan una vez, porque el segundo se deduplica; y un evento emitido mientras nadie recolectaba o se reproduce después en el momento equivocado o se pierde. La regla que sigue: **el estado es un `StateFlow`, los eventos son un `SharedFlow`**. Si podés preguntar con sentido «¿cuál es el valor actual?», es estado. Si la respuesta es «no hay, simplemente ocurrió», es un evento.

`shareIn(scope, started, replay)` convierte un flow frío en uno caliente, y su parámetro `started` es donde viven las decisiones reales. `SharingStarted.Eagerly` arranca el upstream de inmediato y nunca lo para — tu websocket queda abierto cuando nadie mira. `Lazily` arranca con el primer colector y nunca para. `WhileSubscribed(stopTimeoutMillis)` arranca con el primer colector y para el upstream `stopTimeoutMillis` después de que se va el último — la única política que es a la vez barata en reposo y estable frente a huecos breves.

Vale entender ese timeout en concreto. Con `WhileSubscribed(5000)`, una rotación de pantalla o una navegación que deja caer brevemente al último colector **no** destruye el upstream: la suscripción sobrevive el hueco y se reusa la misma conexión. Con `WhileSubscribed(0)` el upstream se cancela y se reinicia, así que una fuente fría vuelve a correr todo su setup — reconectar, reautenticar, reconsultar, y cualquier contador que tuviera se resetea. Cinco segundos es el valor al que convergió la comunidad de Android exactamente por esto, y el mismo razonamiento aplica a cualquier upstream caro y con estado: un change stream, un poll a una API paga, un stream gRPC de larga vida.

`stateIn` es `shareIn` con `replay = 1`, conflación y un valor inicial — produce un `StateFlow`, así que los suscriptores tardíos ven de inmediato el valor actual en vez de esperar la próxima emisión. `stateIn(scope, WhileSubscribed(5000), inicial)` es la línea canónica de un ViewModel, y vale poder decir por qué están los tres argumentos: el scope posee la vida del upstream, la política decide cuándo corre, y el valor inicial es lo que ve un suscriptor antes de que llegue la primera emisión real.

Dos notas operativas. Primero, el scope que pasás posee la corrutina de sharing, así que tiene que sobrevivir a los suscriptores — un scope de aplicación, un `viewModelScope`, el `backgroundScope` de `runTest` — nunca el scope del que casualmente está recolectando ahora. Segundo, los flows calientes nunca completan, así que un colector sobre uno nunca retorna: `toList()` sobre un `SharedFlow` cuelga para siempre, y un test que recolecte uno tiene que hacer `launch` y cancelarlo. Eso no es una aspereza; es la definición de caliente, y recordarlo te ahorra una tarde.

DOJO PETTu pequeño compañero de código

DOJO / TU GUÍA

Aprendamos juntos.

Pregúntame sobre este curso o por dónde empezar.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*

@OptIn(ExperimentalCoroutinesApi::class)
fun main() = runTest {
    val values = listOf(1, 2, 2, 3)

    val state = MutableStateFlow(0)
    val stateSeen = mutableListOf<Int>()
    val j1 = launch { state.collect { stateSeen.add(it) } }
    runCurrent()                                // let the collector subscribe
    for (v in values) state.value = v           // no suspension between writes
    runCurrent(); j1.cancel()

    val shared = MutableSharedFlow<Int>(replay = 2)
    val sharedSeen = mutableListOf<Int>()
    val j2 = launch { shared.collect { sharedSeen.add(it) } }
    runCurrent()
    for (v in values) shared.emit(v)
    runCurrent(); j2.cancel()

    println("stateFlow   = " + stateSeen)       // conflated + deduped
    println("sharedFlow  = " + sharedSeen)      // every emission, duplicate too
    println("replayCache = " + shared.replayCache)
}

The two behaviours that separate StateFlow from SharedFlow: conflation (a fast writer's intermediates never arrive) and equality-based dedup.

Arena IDE

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.test.*

@OptIn(ExperimentalCoroutinesApi::class)
suspend fun TestScope.probe(stopMs: Long): String {
    var starts = 0
    val upstream = flow {
        starts++                                 // one per upstream START
        var i = 0
        while (true) { emit(i++); delay(100) }
    }
    val shared = upstream.shareIn(
        backgroundScope,                         // must outlive the collectors
        SharingStarted.WhileSubscribed(stopTimeoutMillis = stopMs),
        replay = 0,
    )
    val seen = mutableListOf<Int>()
    var job = launch { shared.collect { seen.add(it) } }
    advanceTimeBy(250); job.cancel()
    advanceTimeBy(300)                           // the gap
    job = launch { shared.collect { seen.add(it) } }
    advanceTimeBy(250); job.cancel()
    advanceTimeBy(1000)
    return "seen=" + seen + " upstreamStarts=" + starts
}

@OptIn(ExperimentalCoroutinesApi::class)
fun main() = runTest {
    println("WhileSubscribed(500) -> " + probe(500))
    println("WhileSubscribed(0)   -> " + probe(0))
}

The stop timeout, measured. Same collectors, same gap — the only difference is whether the expensive upstream survives it.

Arena IDE

class OrdersViewModel(repo: OrderRepository) : ViewModel() {

    // STATE: there is always a current value, and duplicates are noise.
    val orders: StateFlow<List<Order>> =
        repo.observeOrders()                       // a cold Flow from the driver
            .stateIn(
                scope = viewModelScope,            // owns the upstream's lifetime
                started = SharingStarted.WhileSubscribed(5_000), // survive rotation
                initialValue = emptyList(),        // what a new subscriber sees first
            )

    // EVENTS: no "current value", and two identical ones must both fire.
    private val _effects = MutableSharedFlow<UiEffect>(extraBufferCapacity = 8)
    val effects: SharedFlow<UiEffect> = _effects.asSharedFlow()

    fun refresh() = viewModelScope.launch {
        runCatching { repo.refresh() }
            .onFailure { _effects.emit(UiEffect.ShowToast("Refresh failed")) }
    }
}

The canonical ViewModel shape: state as a StateFlow with stateIn, one-shot events as a SharedFlow. Reference only — the point is the two different types.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val bus = MutableSharedFlow<String>()
    val seen = mutableListOf<String>()

    // WRONG: bus.toList() would suspend forever — a SharedFlow has no end.
    val job = launch { bus.collect { seen.add(it) } }
    yield()                                   // let the collector subscribe

    bus.emit("a")
    bus.emit("b")
    yield()
    job.cancel()                              // the collector must be stopped

    println("seen = " + seen)
    println("a SharedFlow is never 'done' — you cancel the collector instead")
}

Hot flows never complete — so a collector never returns. Collect them in a launch you cancel, and never call toList() on one.

Arena IDE

🧠 Comprueba tu comprensión

0/1 · 0/1 answered

1. Your ViewModel exposes navigation events as a `StateFlow<NavEvent?>`. A user taps the same button twice in a row and only navigates once. Why?

Back-pressure en Flow: secuencial por defecto, y qué compra buffer()Estado mutable compartido: Mutex, atómicos y confinamiento

import kotlinx.coroutines.*

import kotlinx.coroutines.flow.*

import kotlinx.coroutines.test.*

@OptIn(ExperimentalCoroutinesApi::class)

fun main() = runTest {

val values = listOf(1, 2, 2, 3)

val state = MutableStateFlow(0)

val stateSeen = mutableListOf<Int>()

val j1 = launch { state.collect { stateSeen.add(it) } }

runCurrent() // let the collector subscribe

for (v in values) state.value = v // no suspension between writes

runCurrent(); j1.cancel()

val shared = MutableSharedFlow<Int>(replay = 2)

val sharedSeen = mutableListOf<Int>()

val j2 = launch { shared.collect { sharedSeen.add(it) } }

runCurrent()

for (v in values) shared.emit(v)

runCurrent(); j2.cancel()

println("stateFlow = " + stateSeen) // conflated + deduped

println("sharedFlow = " + sharedSeen) // every emission, duplicate too

println("replayCache = " + shared.replayCache)

}

import kotlinx.coroutines.*

import kotlinx.coroutines.flow.*

import kotlinx.coroutines.test.*

@OptIn(ExperimentalCoroutinesApi::class)

suspend fun TestScope.probe(stopMs: Long): String {

var starts = 0

val upstream = flow {

starts++ // one per upstream START

var i = 0

while (true) { emit(i++); delay(100) }

}

val shared = upstream.shareIn(

backgroundScope, // must outlive the collectors

SharingStarted.WhileSubscribed(stopTimeoutMillis = stopMs),

replay = 0,

)

val seen = mutableListOf<Int>()

var job = launch { shared.collect { seen.add(it) } }

advanceTimeBy(250); job.cancel()

advanceTimeBy(300) // the gap

job = launch { shared.collect { seen.add(it) } }

advanceTimeBy(250); job.cancel()

advanceTimeBy(1000)

return "seen=" + seen + " upstreamStarts=" + starts

}

@OptIn(ExperimentalCoroutinesApi::class)

fun main() = runTest {

println("WhileSubscribed(500) -> " + probe(500))

println("WhileSubscribed(0) -> " + probe(0))

}

class OrdersViewModel(repo: OrderRepository) : ViewModel() {

// STATE: there is always a current value, and duplicates are noise.

val orders: StateFlow<List<Order>> =

repo.observeOrders() // a cold Flow from the driver

.stateIn(

scope = viewModelScope, // owns the upstream's lifetime

started = SharingStarted.WhileSubscribed(5_000), // survive rotation

initialValue = emptyList(), // what a new subscriber sees first

)

// EVENTS: no "current value", and two identical ones must both fire.

private val _effects = MutableSharedFlow<UiEffect>(extraBufferCapacity = 8)

val effects: SharedFlow<UiEffect> = _effects.asSharedFlow()

fun refresh() = viewModelScope.launch {

runCatching { repo.refresh() }

.onFailure { _effects.emit(UiEffect.ShowToast("Refresh failed")) }

}

import kotlinx.coroutines.*

import kotlinx.coroutines.flow.*

fun main() = runBlocking {

val bus = MutableSharedFlow<String>()

val seen = mutableListOf<String>()

// WRONG: bus.toList() would suspend forever — a SharedFlow has no end.

val job = launch { bus.collect { seen.add(it) } }

yield() // let the collector subscribe

bus.emit("a")

bus.emit("b")

yield()

job.cancel() // the collector must be stopped

println("seen = " + seen)

println("a SharedFlow is never 'done' — you cancel the collector instead")

}