SharedFlow: repetición, capacidad extra y las dos formas de emitir

SharedFlow<T> suele presentarse como el flujo para eventos, y esa etiqueta esconde lo único que hay que entender de él: es una estructura con un buffer compartido por todos los suscriptores, en el que cada suscriptor mantiene su propio índice de lectura, y en el que el emisor solo puede avanzar mientras quepa por delante del lector más rezagado. De esa mecánica se derivan, sin excepción, todos sus comportamientos: por qué un suscriptor tardío ve historia, por qué un suscriptor lento frena a todos los demás, por qué tryEmit devuelve false en unos montajes y nunca en otros, y por qué el valor por defecto —cero repetición, cero colchón, suspender— es el más estricto de todos los posibles y también el más honesto.

El buffer real es la suma de dos números

El constructor tiene tres parámetros y ninguno es decorativo. replay es cuántos de los valores ya emitidos se le entregan de golpe a un suscriptor nuevo en el momento de suscribirse. extraBufferCapacity es cuántos valores adicionales puede aceptar el flujo por delante del suscriptor más lento antes de tener que suspender al emisor. Y onBufferOverflow decide qué ocurre cuando ese espacio se agota.

val eventos = MutableSharedFlow<Evento>(
    replay = 0,
    extraBufferCapacity = 0,
    onBufferOverflow = BufferOverflow.SUSPEND,   // valores por defecto
)

La primera consecuencia, que es la que sorprende, es que la capacidad total del buffer es la suma de los dos primeros. Un replay de tres no solo da historia: da también tres huecos de holgura al emisor, porque esos valores están físicamente en el mismo buffer. Por eso un MutableSharedFlow con repetición tres y sin capacidad extra tiene un colchón efectivo de tres, y por eso mucha gente cree haber configurado solo historia cuando en realidad ha configurado también tolerancia a ráfagas.

La segunda consecuencia es que con los valores por defecto —suma cero— el flujo se comporta como una cita: emit suspende hasta que todos los suscriptores actuales han recibido el valor. Es un comportamiento de contrapresión total, correcto para eventos que no se pueden perder, y también la explicación de por qué un solo suscriptor lento arrastra al sistema entero: no hay múltiples colas independientes, hay un buffer con un puntero por suscriptor, y el emisor mira siempre al más rezagado.

Cuándo suspende emit y qué se pierde si no

La regla es corta: emit suspende cuando la política es SUSPEND y el nuevo valor no cabe sin desalojar un valor que el suscriptor más lento todavía no ha leído. Con las otras dos políticas no suspende jamás, y lo que cambia es cuál de los dos extremos del buffer se sacrifica.

Hay un detalle que rompe expectativas y conviene fijar: las políticas de descarte solo pueden usarse si hay algún buffer, es decir, si la suma de replay y extraBufferCapacity es mayor que cero. Pedir DROP_OLDEST con buffer cero es una contradicción —no hay nada viejo que tirar— y la biblioteca lo rechaza.

// Estado de conexión: al último le importa el presente, no la historia.
val conexion = MutableSharedFlow<Estado>(
    replay = 1,
    onBufferOverflow = BufferOverflow.DROP_OLDEST,
)

// Comandos del usuario: ninguno puede perderse, el emisor espera.
val comandos = MutableSharedFlow<Comando>(extraBufferCapacity = 8)

Un segundo detalle, que se descubre siempre depurando: replay no es un caché consultable. No hay una propiedad valor actual en un SharedFlow genérico, aunque expone replayCache como lista de solo lectura para inspección. Si lo que necesitas es el último valor, siempre disponible, sincrónicamente, no estás describiendo un SharedFlow con repetición uno: estás describiendo un StateFlow, y la diferencia entre ambos no es de forma sino de conflación, como veremos en la lección siguiente.

flowchart TD
E[emit de un valor nuevo] --> Q{Cabe por delante del suscriptor mas lento}
Q -->|Si| OK[Se guarda y se entrega]
Q -->|No| P{Politica de desbordamiento}
P -->|SUSPEND| W[El emisor espera al mas lento]
P -->|DROP_OLDEST| D1[Se desaloja el mas antiguo no leido]
P -->|DROP_LATEST| D2[Se descarta el valor recien emitido]

emit frente a tryEmit, y quién los observa

tryEmit es la variante no suspendida y devuelve un Boolean que indica si el valor entró. Su regla es tan estricta que resuelve casi todas las dudas: tryEmit devuelve true si y solo si el valor pudo aceptarse sin suspender. Con la política SUSPEND y buffer cero, eso significa que devolverá false siempre que haya un suscriptor que no esté esperando activamente. Con cualquiera de las políticas de descarte, en cambio, devolverá true siempre, porque descartar es una forma legítima de aceptar.

// Un callback del sistema no puede suspender: hay que comprar el derecho.
private val _pulsaciones = MutableSharedFlow<Pulsacion>(
    extraBufferCapacity = 64,
    onBufferOverflow = BufferOverflow.DROP_OLDEST,
)

fun alPulsar(p: Pulsacion) {
    _pulsaciones.tryEmit(p)   // aquí siempre entra: se pagó con buffer
}

De ahí sale la única heurística sólida sobre el asunto: si escribes tryEmit sobre un flujo con los parámetros por defecto y no compruebas el resultado, has escrito una pérdida silenciosa con apariencia de emisión. Y si lo escribes ignorando el booleano porque siempre da true, entonces el diseño real está en el constructor y no en el punto de llamada, así que el comentario que explica la política pertenece al constructor.

La otra pieza que completa el cuadro es subscriptionCount, un StateFlow<Int> que publica cuántos suscriptores hay en cada momento. Sirve para dos cosas legítimas y una discutible. Legítimas: arrancar la fuente cara solo cuando aparece el primer suscriptor y pararla cuando se va el último —que es justamente lo que automatiza WhileSubscribed—, y esperar activamente a que haya alguien antes de emitir, cerrando la ventana de carrera del arranque.

// Cerrar la ventana: no emitas al vacío durante el arranque.
suspend fun emitirCuandoHayaPublico(e: Evento) {
    eventos.subscriptionCount.first { it > 0 }
    eventos.emit(e)
}

Lo discutible es usar subscriptionCount para tomar decisiones de negocio, porque es una cantidad inherentemente de carrera: entre que la lees y actúas, puede haber cambiado. Como señal de ciclo de vida es excelente; como condición de corrección, no.

Queda una propiedad estructural que conviene grabar: MutableSharedFlow no completa nunca y no propaga excepciones del productor. No existe cerrar un SharedFlow. Si tu dominio necesita señalar el fin o el fallo, tienes que modelarlo dentro del tipo emitido —un tipo suma con un caso terminal— porque el canal fuera de banda que ofrecía Channel con su close aquí sencillamente no está.

Difundir no es repartir

Hay una confusión que sobrevive a todo lo anterior y que conviene cortar de raíz, porque lleva a usar SharedFlow donde hacía falta un Channel. Un SharedFlow difunde: cada suscriptor recibe todos los valores, y dos suscriptores ven la misma secuencia. Un Channel reparte: cada elemento lo recibe exactamente un receptor, y dos receptores se dividen el trabajo. Son semánticas opuestas y ninguna emula bien a la otra.

flowchart LR
P[Productor] --> SF[SharedFlow difunde]
SF --> A[Suscriptor A ve todo]
SF --> B[Suscriptor B ve todo]
P2[Productor] --> CH[Channel reparte]
CH --> C[Receptor C ve la mitad]
CH --> D[Receptor D ve la otra mitad]

La consecuencia práctica es que un evento que debe procesarse una sola vez —cerrar sesión, navegar, enviar una petición— no pertenece a un SharedFlow si existe la posibilidad de que haya dos suscriptores, porque entonces se procesará dos veces. El síntoma clásico es una navegación duplicada tras una reconstrucción de pantalla en la que el suscriptor viejo aún no se había ido.

// Difusión: varios observadores del mismo hecho, todos lo ven.
val sesionExpirada: SharedFlow<Unit> = _sesionExpirada.asSharedFlow()

// Reparto: exactamente un consumidor debe atender cada trabajo.
private val trabajos = Channel<Trabajo>(capacity = 64)

La segunda diferencia es la terminación, ya mencionada pero decisiva aquí: un Channel se cierra y ese cierre viaja al receptor como información; un SharedFlow no tiene cierre. Si tu dominio necesita decir se acabó o falló, con SharedFlow tienes que codificarlo dentro del valor. Elegir entre ambos tipos, por tanto, se decide con dos preguntas encadenadas: cuántos deben ver cada elemento, y si existe un final que alguien tenga que observar.