Kotlin Flow: fluxos de dados assíncronos

1. Visão geral

Uma função suspend retorna apenas um valor. Quando você precisa receber múltiplos valores ao longo do tempo, como novos pedidos, status de pagamento ou preço de ações, você precisa de um fluxo de dados. Kotlin Flow resolve exatamente isso.

Este artigo explica o que é Flow e como transformar dados usando operadores. Depois, você aprenderá como mudar de thread com flowOn, tratar erros, usar StateFlow e SharedFlow, e escrever testes para Flow.

2. Instalação

Flow está na biblioteca kotlinx-coroutines. Este artigo pressupõe que seu projeto Maven já tem o kotlin-maven-plugin configurado. Você só precisa adicionar as dependências a seguir:

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

Se você usar Spring Boot 3.x, a versão do coroutines já é gerenciada via BOM, então você pode omitir a tag <version>.

3. Primeiro Flow

3.1. Criar e coletar Flow

A maneira mais simples de criar um Flow é usar o builder flow { } e chamar 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("Começando a emitir pedidos")
    for (id in 1..3) {
        delay(100)                      // simular aguardando dados do sistema
        emit(Order(id, id * 50_000L))
    }
}

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

Explicação:

• orderStream() não é uma função suspend. Ela apenas retorna uma descrição do fluxo de dados.
• Dentro de flow { } você pode chamar funções suspend como delay().
• collect é um terminal operator. Apenas quando você o chama é que o código dentro de flow { } é executado.

Resultado:

Flow criado
Começando a emitir pedidos
Recebido: Order(id=1, amount=50000)
Recebido: Order(id=2, amount=100000)
Recebido: Order(id=3, amount=150000)

A linha “Flow criado” aparece antes de “Começando a emitir pedidos”. Isso prova que a criação de um Flow não ativa nada.

3.2. Flow é “cold”

Um Flow criado com flow { } é um cold stream. Cada vez que você chama collect, todo o bloco de código é executado novamente do início:

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

Começando a emitir pedidos
Começando a emitir pedidos

Nota: Se o bloco flow { } chamar uma API ou escrever no banco de dados, cada collect chamará tudo novamente do início. Esta é a origem de muitos bugs “por que a API foi chamada duas vezes?”. Se você quiser compartilhar uma fonte entre múltiplos coletadores, veja a seção 6.

4. Operadores de transformação

4.1. Intermediate operator

Flow tem um conjunto de operadores familiares como collection: map, filter, take, onEach… Esses operadores são intermediate. Eles retornam um novo Flow e também são lazy como o 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) }
}

Começando a emitir pedidos
Pedido #2: 100000 VND
Pedido #3: 150000 VND

Cada elemento passa por toda a cadeia filter → map → collect antes que o próximo elemento seja emitido. Flow executa sequencialmente por padrão no mesmo coroutine.

Diferentemente do map de uma collection, o lambda em map de um Flow é suspend. Portanto, você pode chamar uma função assíncrona diretamente nele:

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

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

4.2. Outros terminal operators

Além de collect, você pode usar outros terminal operators para obter resultados diretamente:

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("Número de pedidos: ${all.size}, primeiro grande pedido: #${firstBig.id}, receita: $revenue")
}

...
Número de pedidos: 3, primeiro grande pedido: #2, receita: 300000

first { } cancela o Flow assim que encontra um elemento que corresponde. Portanto, o pedido #3 nunca é emitido nessa coleta.

5. Trocar de thread com flowOn

Por padrão, o código dentro de flow { } executa no contexto do coroutine que chama collect. Para empurrar a parte de emissão de dados para outra thread, por exemplo, ao ler um arquivo ou banco de dados, você usa flowOn:

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

data class Customer(val name: String)

fun loadCustomers(): Flow<Customer> = flow {
    println("Lendo dados em ${Thread.currentThread().name}")
    listOf("Ana", "Bruno").forEach { emit(Customer(it)) }
}.flowOn(Dispatchers.IO)

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

Lendo dados em DefaultDispatcher-worker-1
Processando Ana em main
Processando Bruno em main

flowOn apenas afeta a parte upstream, ou seja, os operadores que vêm antes dele. A parte collect ainda executa no contexto do chamador. Este princípio é chamado de context preservation.

Nota: Não use withContext para envolver emit() dentro de flow { }. O código abaixo lançará IllegalStateException: Flow invariant is violated:

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

Sempre use flowOn. Se você realmente precisar emitir de múltiplos coroutines diferentes, use channelFlow { }.

6. Tratamento de erros

6.1. catch e onCompletion

Exceções em Flow propagam para collect como em código normal. O operador catch permite que você intercepte erros e emita um valor de substituição:

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

fun paymentStatus(): Flow<String> = flow {
    emit("PENDING")
    throw IllegalStateException("Portal de pagamento timeout")
}

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

PENDING
FAILED: Portal de pagamento timeout
Completo, cause=null

Explicação:

• catch apenas intercepta erros de upstream. Exceções lançadas dentro do lambda de collect não serão capturadas por ele.
• onCompletion vem depois de catch. O erro já foi tratado na etapa anterior, então cause é null. Se você inverter a ordem dos dois operadores, verá que cause contém a exceção.

6.2. retry

Para erros temporários como perda de conexão, você pode usar retry. Este operador coleta o Flow upstream novamente do início:

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

fun main() = runBlocking {
    var attempts = 0
    flow {
        attempts++
        if (attempts < 3) throw IOException("Conexão perdida")
        emit("OK após $attempts tentativas")
    }
        .retry(3) { e -> e is IOException }
        .collect { println(it) }
}

OK após 3 tentativas

O predicado { e is IOException } ajuda você a tentar novamente apenas com erros de rede. Erros comerciais como IllegalArgumentException não serão retentados. Dentro do predicado, você pode chamar delay() para criar um backoff.

7. StateFlow e SharedFlow: hot stream

flow { } é cold. Mas frequentemente você precisa de uma fonte de dados que existe independentemente dos coletadores, como o status de um carrinho de compras ou eventos do sistema. Nesses casos, você precisa de um hot stream.

7.1. StateFlow

StateFlow sempre mantém um valor atual. Um novo coletor receberá esse valor imediatamente:

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 do carrinho: $it") }
    }
    delay(50)
    cart.add(100_000)
    delay(50)
    cart.add(250_000)
    delay(50)
    job.cancel()
}

Total do carrinho: 0
Total do carrinho: 100000
Total do carrinho: 350000

Pontos importantes:

• A propriedade _total é private e mutável. Fora dela, você vê apenas StateFlow de leitura. Este é o padrão padrão para encapsular estado.
• update { } atualiza atomicamente, seguro quando múltiplas threads chamam add simultaneamente.
• StateFlow nunca termina. Portanto, você deve cancel a tarefa, senão runBlocking esperará para sempre.
• StateFlow ignora valores que são iguais ao valor atual. Ele também conflate: um coletor lento vê apenas o valor mais recente, não os valores intermediários.

7.2. SharedFlow

SharedFlow é adequado para eventos, como “pedido foi pago”. Aqui, cada emissão tem significado e não há conceito de “valor atual”:

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 foi pago")
    }
}

fun main() = runBlocking {
    val payments = PaymentService()
    payments.pay(0)                       // ninguém está ouvindo ainda -> perdido

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

Email: Pedido #1 foi pago
Email: Pedido #2 foi pago

Com a configuração padrão (replay = 0), eventos emitidos quando não há subscriber são descartados. Se um coletor chegar tarde e precisar receber os últimos eventos, use MutableSharedFlow<String>(replay = 1).

Você também pode transformar um Flow cold em hot usando stateIn(scope, SharingStarted.WhileSubscribed(5_000), initialValue) ou shareIn(...). Dessa forma, múltiplos coletores compartilham uma única execução upstream.

8. Testando Flow

A biblioteca kotlinx-coroutines-test fornece runTest. Esta função ignora delay usando tempo virtual, portanto os testes são executados quase instantaneamente:

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 `mantém apenas pedidos de 100 mil ou mais`() = runTest {
        val ids = orderStream()
            .filter { it.amount >= 100_000 }
            .map { it.id }
            .toList()

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

    @Test
    fun `erro de pagamento é convertido para status FAILED`() = runTest {
        val states = paymentStatus()
            .catch { emit("FAILED") }
            .toList()

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

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

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

Para Flows finitos, toList() é a maneira mais simples de verificar todos os valores emitidos. Com StateFlow, você lê .value diretamente. Não chame toList() em StateFlow, pois esse Flow nunca termina e o teste ficará pendurado.

9. Erros comuns

• Código após collect não executa. collect é uma função suspend que retorna apenas quando o Flow termina. Com StateFlow/SharedFlow, isso nunca acontece. Coloque cada collect infinito em um launch separado.
• Chamar API múltiplas vezes involuntariamente. Cada coletor de um Flow cold executa upstream novamente. Use shareIn/stateIn quando precisar compartilhar.
• Usar withContext em torno de emit. Substitua por flowOn ou channelFlow.
• Pensar que catch captura todos os erros. Este operador apenas captura erros upstream. Se você quiser capturar erros na parte de tratamento, mova a lógica para onEach { } colocado antes de catch, depois chame collect() sem parâmetros.
• Usar StateFlow para eventos. Dois eventos idênticos consecutivos serão mesclados em um. Use SharedFlow para eventos.

Nota: Não use Flow quando você tiver apenas um resultado. Uma função suspend fun findOrder(id: Int): Order é muito mais simples e legível do que Flow<Order> emitindo apenas um elemento.

10. Conclusão

Neste artigo, você viu que Flow é um fluxo de dados assíncrono, cold e lazy. Você aprendeu como transformar dados com map/filter, trocar de thread com segurança usando flowOn e tratar erros com catch, retry, onCompletion. Para a necessidade de compartilhar estado e eventos, StateFlow e SharedFlow são as duas principais ferramentas. runTest ajuda você a testá-los rápido e de forma estável.

A partir daqui, você pode experimentar operadores avançados como combine, flatMapLatest, debounce ou buffer para construir pipelines de dados mais complexos.