Flow — qué es, streams cold y hot en corrutinas de Kotlin

¿Qué es Flow en Kotlin?

Flow es un tipo del paquete kotlinx.coroutines.flow que representa un flujo de datos asíncrono cold. En esencia, Flow es una secuencia de corrutina que emite valores mediante la función emit() y finaliza ya sea con éxito o con una excepción. La recolección del flujo se realiza mediante el operador terminal collect(), que es una función suspend.

Semántica cold

Stream cold significa que el código dentro del builder flow se ejecuta de nuevo para cada suscriptor. Observable.fromIterable en RxJava se comporta de manera similar: un nuevo suscriptor recibe todos los valores desde el principio. En Flow, esto se implementa mediante la función suspend collect, que bloquea la corrutina durante toda la recolección de datos.

Constructores de Flow

Kotlin proporciona varias formas de crear Flow: flow { } — la construcción básica con emit(), flowOf(vararg values) — para un conjunto fijo de valores, .asFlow() — una extensión para colecciones y Sequence. Todos los constructores son cold — los datos se generan solo al llamar al operador terminal.

Streams cold y hot

La distinción entre streams cold y hot es un concepto clave de la programación reactiva. Stream cold (Flow, Observable) inicia la generación de datos al suscribirse. Stream hot (Channel, SharedFlow) emite datos independientemente — el suscriptor solo recibe lo que ocurre después de la suscripción, sin el inicio de la secuencia.

SharedFlow es un Flow hot que puede tener múltiples suscriptores y reproducir valores recientes cuando se configura replay. SharedFlow es adecuado para eventos (notificaciones únicas). StateFlow es su variante con un valor de estado fijo, que almacena en caché el último valor para nuevos suscriptores.

ChannelFlow usa Channel internamente, combinando propiedades de Flow y Channel. Soporta buffering y backpressure mediante capacity. ChannelFlow es útil al convertir APIs callback en un flujo reactivo, donde los valores se emiten desde diferentes corrutinas.

Conversión entre cold y hot

Para convertir cold Flow a hot SharedFlow se usa el operador shareIn(scope, started, replay). El parámetro started controla el momento de inicio: SharingStarted.WhileSubscribed() — activo mientras haya suscriptores, Lazily — inicio con el primer suscriptor, Eagerly — inicio inmediato. La conversión inversa — hot a cold: StateFlow.asFlow() devuelve un Flow cold que al hacer collect emite el valor actual del StateFlow. Esto es útil para pruebas.

Operadores de Flow

Flow proporciona un rico conjunto de operadores que funcionan como funciones suspend dentro de una corrutina. Los operadores no tienen estado y devuelven un nuevo Flow — el flujo original permanece sin cambios. Esto permite construir cadenas de transformación seguras sin efectos secundarios.

El operador map transforma cada valor del flujo mediante una transformación asíncrona o síncrona. filter solo pasa los valores que cumplen la condición. catch captura excepciones antes del operador terminal y permite recuperar el flujo. flatMapLatest cancela la emisión anterior cuando llega un nuevo valor — similar a switchMap en Rx.

El operador debounce en Flow retrasa la publicación del valor durante un tiempo de espera especificado. Si llega un nuevo valor durante ese tiempo, el temporizador se reinicia. En Android, debounce se usa para búsqueda: la solicitud se envía solo después de una pausa de 300-400 ms, lo que reduce las llamadas a la API entre 3 y 5 veces.

Operadores terminales

Además de collect(), Flow soporta otros operadores terminales: toList() recolecta todos los valores en una lista — útil para pruebas, first() devuelve el primer elemento y cancela el flujo, single() espera exactamente un elemento. fold(initial) acumula valores mediante una función pasada. Todos los operadores terminales son funciones suspend y deben llamarse dentro de una corrutina u otra función suspend.

Ejemplos de código Flow

El primer ejemplo — un Flow básico que genera números con transformación mediante el operador map:

val numberFlow = flow {
    for (i in 1..5) {
        delay(500)
        emit(i)
    }
}

scope.launch {
    numberFlow
        .map { "Número: $it" }
        .collect { value ->
            println(value)
        }
}

El segundo ejemplo — transformación de flujo con filtrado y manejo de errores mediante catch:

flow {
    emit("data1")
    emit("data2")
    throw RuntimeException("network error")
}
    .catch { e ->
        emit("fallback_data")
    }
    .collect { value ->
        println(value)
    }

El tercer ejemplo — uso de StateFlow en ViewModel para UI reactiva en Jetpack Compose:

class SearchViewModel : ViewModel() {
    private val _query = MutableStateFlow("")
    val results: StateFlow<List<Result>> = _query
        .debounce(300)
        .flatMapLatest { query ->
            repository.search(query)
        }
        .catch { emit(emptyList()) }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5000), emptyList())

    fun onQueryChanged(query: String) {
        _query.value = query
    }
}

StateFlow y SharedFlow

StateFlow es un Flow hot con un único valor actual. Almacena en caché el último valor y lo pasa inmediatamente a un nuevo suscriptor. StateFlow es un contenedor observable para el estado, soporta comparación equals — si el nuevo valor coincide con el actual, no se produce emisión. Jetpack Compose usa StateFlow a través de collectAsState().

SharedFlow es un Flow hot más flexible sin valor inicial obligatorio. SharedFlow se configura mediante replay (número de valores para nuevos suscriptores), extraBufferCapacity (búfer más allá de replay) y onBufferOverflow (estrategia en caso de desbordamiento). SharedFlow es ideal para eventos únicos: navegación, Snackbar, analítica.

Flow en la arquitectura Android es recomendado por Google como fuente principal de datos (Capa: Repository → UseCase → ViewModel). LiveData es inferior a Flow en flexibilidad: Flow soporta corrutinas, operadores, backpressure y funciona fuera de la capa UI. La migración de LiveData a Flow es una práctica estándar en proyectos Android modernos.

Al usar Flow en ViewModel, es importante elegir el tipo correcto. StateFlow es ideal para el estado UI que debe sobrevivir a la rotación de pantalla. SharedFlow es adecuado para eventos donde el reprocesamiento es inaceptable — por ejemplo, navegación. Flow con collect() en lifecycleScope da el máximo control sobre el contexto de ejecución pero requiere cancelación manual al salir de la pantalla.

Las pruebas de Flow se realizan mediante kotlinx-coroutines-test. La librería proporciona TestDispatcher — tiempo virtual que permite acelerar retrasos (delay) y controlar el orden de ejecución de las corrutinas. TestScope.runTest { } crea un entorno aislado para probar Flow. El operador toList() se usa a menudo en pruebas para recolectar todos los valores del flow con un tiempo de espera, para verificar que el flujo emitió la secuencia de datos correcta.

Flow se integra bien con Room (librería Android para BD): los métodos DAO pueden devolver Flow<List<Entity>>. Room emite automáticamente un nuevo valor ante cualquier cambio en la tabla — la UI se actualiza sin disparador manual. Esto se implementa mediante InvalidationTracker, que internamente usa Flow con callbackFlow. Este enfoque elimina la necesidad de LiveData y hace que la capa de datos esté completamente orientada a corrutinas. Jetpack Compose a través de collectAsState() se suscribe a StateFlow y redibuja solo aquellos componentes cuyos datos han cambiado — esto proporciona un rendimiento inalcanzable con arquitecturas orientadas a LiveData. DataStore (reemplazo de SharedPreferences) también devuelve Flow<Preferences>, proporcionando lectura reactiva de la configuración de la aplicación sin disparadores de actualización manuales.

Flow soporta comunicación entre procesos mediante kotlinx-coroutines-core en JVM sin librerías adicionales. Por ejemplo, en aplicaciones de servidor con Ktor, Flow puede representar un flujo de mensajes WebSocket entrantes. Cada mensaje se emite en el flujo, pasa por filtrado y agregación mediante operadores, y el resultado se envía al cliente. Este enfoque reemplaza librerías reactivas como Reactor o RxJava en proyectos Kotlin.

La compatibilidad de Flow con código RxJava existente se proporciona mediante el módulo kotlinx-coroutines-rx3. La función de extensión Flow.asObservable() convierte Flow a Observable de RxJava 3. La conversión inversa — CompletableSource.asFlow(), Observable.asFlow(). Esto simplifica la migración de RxJava a corrutinas: se puede reescribir el proyecto por etapas, dejando algunas capas en RxJava. Al convertir, hay que considerar la diferencia en semántica cold/hot: Observable puede ser tanto cold como hot, Flow siempre es cold para Flow normal y hot para SharedFlow.

Manejo de errores y pruebas de Flow

El manejo de errores en Flow tiene una particularidad: si ocurre una excepción dentro del builder flow antes del operador terminal, se propaga a catch. Si ocurre una excepción en un operador después del builder, la captura catch después de ese operador. retryWhen permite reintentar la suscripción con una condición: reintentar en error de red hasta 3 veces, pero no reintentar en CancellationException. Flow elimina errores dependientes del estado porque no almacena estado — esto simplifica la depuración en comparación con Observable, donde Subject almacena estado interno.

Las pruebas de Flow con kotlinx-coroutines-test usan TestDispatcher para simular retrasos. Turbine es una librería popular de la comunidad para probar Flow: test { } lanza Flow, awaitItem() espera el siguiente valor, awaitComplete() espera la finalización. Turbine añade un tiempo de espera por defecto, evitando que las pruebas se cuelguen. Para probar StateFlow, use .testIn(scope) con verificación de valores en orden cronológico.