Kotlin Flow: flujos de datos asincronos

1. Resumen

Una función suspend devuelve un único valor. Cuando necesitas recibir varios valores a lo largo del tiempo, por ejemplo nuevos pedidos, estado de pagos o precios de acciones, necesitas un flujo de datos. Kotlin Flow resuelve exactamente eso.

Este artículo explica qué es Flow y cómo transformar datos usando operadores. Luego aprenderás a cambiar de thread con flowOn, manejar errores, usar StateFlow y SharedFlow, y escribir tests para Flow.

2. Instalación

Flow está en la librería kotlinx-coroutines. Este artículo asume que tu proyecto Maven ya tiene configurado kotlin-maven-plugin. Solo necesitas añadir las siguientes dependencias:

<dependencies>
    <dependency>
        <groupId>org.jetbrains.kotlinx</groupId>
        <artifactId>kotlinx-coroutines-core</artifactId>
        <version>1.10.2</version>
    </dependency>

    <!-- Test -->
    <dependency>
        <groupId>org.jetbrains.kotlinx</groupId>
        <artifactId>kotlinx-coroutines-test</artifactId>
        <version>1.10.2</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.junit.jupiter</groupId>
        <artifactId>junit-jupiter</artifactId>
        <version>5.11.4</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.assertj</groupId>
        <artifactId>assertj-core</artifactId>
        <version>3.27.3</version>
        <scope>test</scope>
    </dependency>
</dependencies>

Si usas Spring Boot 3.x, la versión de coroutines ya se gestiona mediante BOM, así que puedes omitir la etiqueta <version>.

3. Tu primer Flow

3.1. Crear y recolectar Flow

La forma más simple de crear un Flow es usar el builder flow { } e invocar emit() para cada valor:

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking

data class Order(val id: Int, val amount: Long)

fun orderStream(): Flow<Order> = flow {
    println("Comenzando a emitir pedidos")
    for (id in 1..3) {
        delay(100)                      // simular espera de datos del sistema
        emit(Order(id, id * 50_000L))
    }
}

fun main() = runBlocking {
    val orders = orderStream()
    println("Flow creado")
    orders.collect { println("Recibido: $it") }
}

Explicación:

• orderStream() no es una función suspend. Solo devuelve una descripción del flujo de datos.
• Dentro de flow { } puedes invocar funciones suspend como delay().
• collect es un terminal operator. Solo cuando lo invocas, el código dentro de flow { } se ejecuta.

Resultado:

Flow creado
Comenzando a emitir pedidos
Recibido: Order(id=1, amount=50000)
Recibido: Order(id=2, amount=100000)
Recibido: Order(id=3, amount=150000)

La línea “Flow creado” aparece antes de “Comenzando a emitir pedidos”. Esto demuestra que crear un Flow no dispara nada.

3.2. Flow es “frío”

El Flow creado con flow { } es un cold stream. Cada vez que haces collect se ejecuta nuevamente todo el bloque de código desde el principio:

fun main() = runBlocking {
    val orders = orderStream()
    orders.collect { }
    orders.collect { }
}

Comenzando a emitir pedidos
Comenzando a emitir pedidos

Nota: Si el bloque flow { } hace llamadas a API o escribe en la base de datos, cada collect invocará de nuevo desde el principio. Esta es la causa de muchos bugs “¿por qué la API se invoca dos veces?”. Si necesitas compartir una fuente entre múltiples collectors, consulta la sección 6.

4. Operadores de transformación

4.1. Intermediate operator

Flow tiene operadores familiares como las colecciones: map, filter, take, onEach… Estos operadores son intermediate. Devuelven un nuevo Flow y son igual de lazy que el Flow original.

import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.map

fun main() = runBlocking {
    orderStream()
        .filter { it.amount >= 100_000 }
        .map { "Pedido #${it.id}: ${it.amount} VND" }
        .collect { println(it) }
}

Comenzando a emitir pedidos
Pedido #2: 100000 VND
Pedido #3: 150000 VND

Cada elemento pasa por toda la cadena filter → map → collect antes de que se emita el siguiente. Flow se ejecuta secuencialmente por defecto en la misma coroutine.

A diferencia de map en colecciones, el lambda en map de Flow es suspend. Por eso puedes invocar funciones asincronizadas directamente:

suspend fun fetchCustomerName(orderId: Int): String {
    delay(50)
    return "Cliente $orderId"
}

orderStream()
    .map { fetchCustomerName(it.id) }
    .collect { println(it) }

4.2. Otros terminal operators

Además de collect, puedes usar otros terminal operators para obtener resultados directamente:

import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.fold
import kotlinx.coroutines.flow.toList

fun main() = runBlocking {
    val all = orderStream().toList()
    val firstBig = orderStream().first { it.amount > 60_000 }
    val revenue = orderStream().fold(0L) { acc, order -> acc + order.amount }

    println("Pedidos: ${all.size}, primer pedido grande: #${firstBig.id}, ingresos: $revenue")
}

...
Pedidos: 3, primer pedido grande: #2, ingresos: 300000

first { } cancela el Flow tan pronto como encuentra un elemento que coincide. Por eso el pedido #3 nunca se emite en ese collect.

5. Cambiar de thread con flowOn

Por defecto, el código dentro de flow { } se ejecuta en el contexto de la coroutine que invoca collect. Para ejecutar la parte de emisión en otro thread, por ejemplo leer un archivo o base de datos, usa flowOn:

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.flowOn

data class Customer(val name: String)

fun loadCustomers(): Flow<Customer> = flow {
    println("Leyendo datos en ${Thread.currentThread().name}")
    listOf("Ana", "Benito").forEach { emit(Customer(it)) }
}.flowOn(Dispatchers.IO)

fun main() = runBlocking {
    loadCustomers().collect {
        println("Procesando ${it.name} en ${Thread.currentThread().name}")
    }
}

Leyendo datos en DefaultDispatcher-worker-1
Procesando Ana en main
Procesando Benito en main

flowOn solo afecta la parte upstream, es decir, los operadores anteriores a él. La parte collect aún se ejecuta en el contexto del llamador. Este principio se llama context preservation.

Nota: No uses withContext para envolver emit() dentro de flow { }. El siguiente código lanzará IllegalStateException: Flow invariant is violated:

flow {
    withContext(Dispatchers.IO) { emit(1) } // INCORRECTO
}

Siempre usa flowOn. Si realmente necesitas emitir desde múltiples coroutines diferentes, usa channelFlow { }.

6. Manejo de errores

6.1. catch y onCompletion

Las excepciones en Flow se propagan hacia collect como código normal. El operador catch te permite interceptar errores y emitir un valor alternativo:

import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.onCompletion

fun paymentStatus(): Flow<String> = flow {
    emit("PENDING")
    throw IllegalStateException("Puerta de pago agotada por tiempo")
}

fun main() = runBlocking {
    paymentStatus()
        .catch { e -> emit("FAILED: ${e.message}") }
        .onCompletion { cause -> println("Completado, cause=$cause") }
        .collect { println(it) }
}

PENDING
FAILED: Puerta de pago agotada por tiempo
Completado, cause=null

Explicación:

• catch solo captura errores del upstream. Las excepciones lanzadas en el lambda de collect no serán capturadas.
• onCompletion está después de catch. El error ya fue manejado antes, así que cause es null. Si intercambias los dos operadores, verás que cause contiene la excepción.

6.2. retry

Para errores transitorios como pérdida de red, puedes usar retry. Este operador recolecta el Flow upstream desde el principio:

import java.io.IOException
import kotlinx.coroutines.flow.retry

fun main() = runBlocking {
    var attempts = 0
    flow {
        attempts++
        if (attempts < 3) throw IOException("Conexión perdida")
        emit("OK después de $attempts intentos")
    }
        .retry(3) { e -> e is IOException }
        .collect { println(it) }
}

OK después de 3 intentos

El predicado { e is IOException } te permite reintentar solo con errores de red. Los errores de negocio como IllegalArgumentException no se reintentan. Dentro del predicado, puedes invocar delay() para implementar backoff.

7. StateFlow y SharedFlow: hot streams

flow { } es frío. Pero a menudo necesitas una fuente de datos que existe independientemente del collector, por ejemplo el estado del carrito o eventos del sistema. En ese caso necesitas un hot stream.

7.1. StateFlow

StateFlow siempre mantiene un valor actual. Un nuevo collector recibirá ese valor inmediatamente:

import kotlinx.coroutines.cancel
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.launch

class CartService {
    private val _total = MutableStateFlow(0L)
    val total: StateFlow<Long> = _total.asStateFlow()

    fun add(price: Long) = _total.update { it + price }
}

fun main() = runBlocking {
    val cart = CartService()
    val job = launch {
        cart.total.collect { println("Total carrito: $it") }
    }
    delay(50)
    cart.add(100_000)
    delay(50)
    cart.add(250_000)
    delay(50)
    job.cancel()
}

Total carrito: 0
Total carrito: 100000
Total carrito: 350000

Puntos importantes:

• La propiedad _total es private y mutable. Externamente solo se ve el StateFlow de solo lectura. Este es el patrón estándar para encapsular estado.
• update { } actualiza de forma atómica (atomic), seguro cuando múltiples threads invocan add.
• StateFlow nunca termina. Por eso debes hacer cancel del job, si no runBlocking esperará para siempre.
• StateFlow ignora valores iguales al valor actual. También conflate: un collector lento solo ve el valor más reciente, no los pasos intermedios.

7.2. SharedFlow

SharedFlow es apropiado para eventos, por ejemplo “el pedido ha sido pagado”. Aquí cada emisión tiene significado y no hay concepto de “valor actual”:

import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow

class PaymentService {
    private val _events = MutableSharedFlow<String>()
    val events: SharedFlow<String> = _events

    suspend fun pay(orderId: Int) {
        _events.emit("Pedido #$orderId pagado")
    }
}

fun main() = runBlocking {
    val payments = PaymentService()
    payments.pay(0)                       // aún no hay listeners -> se pierde

    val job = launch { payments.events.collect { println("Email: $it") } }
    delay(50)
    payments.pay(1)
    payments.pay(2)
    delay(50)
    job.cancel()
}

Email: Pedido #1 pagado
Email: Pedido #2 pagado

Con la configuración por defecto (replay = 0), los eventos emitidos cuando no hay subscribers se pierden. Si un collector llega tarde y necesita recibir algunos eventos recientes, usa MutableSharedFlow<String>(replay = 1).

También puedes convertir un Flow frío en un Flow caliente usando stateIn(scope, SharingStarted.WhileSubscribed(5_000), initialValue) o shareIn(...). Entonces múltiples collectors comparten una única ejecución del upstream.

8. Escribir tests para Flow

La librería kotlinx-coroutines-test proporciona runTest. Esta función salta los delay usando tiempo virtual, así que los tests se ejecutan casi instantáneamente:

import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.test.runTest
import org.assertj.core.api.Assertions.assertThat
import org.junit.jupiter.api.Test

class OrderFlowTest {

    @Test
    fun `mantener solo pedidos de 100 mil en adelante`() = runTest {
        val ids = orderStream()
            .filter { it.amount >= 100_000 }
            .map { it.id }
            .toList()

        assertThat(ids).containsExactly(2, 3)
    }

    @Test
    fun `error de pago se convierte a estado FAILED`() = runTest {
        val states = paymentStatus()
            .catch { emit("FAILED") }
            .toList()

        assertThat(states).containsExactly("PENDING", "FAILED")
    }

    @Test
    fun `carrito acumula valores`() = runTest {
        val cart = CartService()
        cart.add(100_000)
        cart.add(250_000)

        assertThat(cart.total.value).isEqualTo(350_000L)
    }
}

Con Flows finitos, toList() es la forma más simple de verificar todos los valores emitidos. Con StateFlow, lee .value directamente. No invoques toList() en un StateFlow, porque este Flow nunca termina así que el test se quedará colgado.

9. Errores comunes

• El código después de collect no se ejecuta. collect es una función suspend que solo retorna cuando el Flow termina. Con StateFlow/SharedFlow, eso nunca ocurre. Coloca cada collect infinito en su propio launch.
• Llamar a la API múltiples veces sin intención. Cada collector de un Flow frío re-ejecuta el upstream. Usa shareIn/stateIn cuando necesites compartir.
• Usar withContext alrededor de emit. Reemplázalo con flowOn o channelFlow.
• Pensar que catch captura todos los errores. Este operador solo captura errores del upstream. Si quieres capturar errores en la parte de procesamiento, mueve la lógica a onEach { } colocado antes de catch, luego invoca collect() sin parámetros.
• Usar StateFlow para eventos. Dos eventos iguales consecutivos se fusionan en uno. Usa SharedFlow para eventos.

Nota: No uses Flow cuando tienes solo un resultado. Una simple función suspend fun findOrder(id: Int): Order es mucho más legible que Flow<Order> que emite un único elemento.

10. Conclusión

En este artículo, viste que Flow es un flujo de datos asincronizado, frío y lazy. Aprendiste a transformar datos con map/filter, cambiar de thread de forma segura con flowOn y manejar errores con catch, retry, onCompletion. Para necesidades de compartir estado y eventos, StateFlow y SharedFlow son las dos herramientas principales. runTest te ayuda a escribir tests rápidos y estables.

A partir de aquí, puedes explorar operadores más avanzados como combine, flatMapLatest, debounce o buffer para construir pipelines de datos más complejos.