Flow — o que é, streams cold e hot em corrotinas Kotlin

O que é Flow em Kotlin?

Flow é um tipo do pacote kotlinx.coroutines.flow que representa um fluxo de dados assíncrono cold. Em essência, Flow é uma sequência de corrotina que emite valores através da função emit() e termina com sucesso ou com uma exceção. A coleta do fluxo é feita através do operador terminal collect(), que é uma função suspend.

Semântica cold

Stream cold significa que o código dentro do builder flow é executado novamente para cada assinante. Observable.fromIterable no RxJava se comporta de forma semelhante: um novo assinante recebe todos os valores desde o início. No Flow, isso é implementado através da função suspend collect, que bloqueia a corrotina durante toda a coleta de dados.

Construtores Flow

Kotlin fornece várias formas de criar Flow: flow { } — a construção básica com emit(), flowOf(vararg values) — para um conjunto fixo de valores, .asFlow() — uma extensão para coleções e Sequence. Todos os construtores são cold — os dados são gerados apenas quando o operador terminal é chamado.

Streams cold e hot

A distinção entre streams cold e hot é um conceito chave da programação reativa. Stream cold (Flow, Observable) inicia a geração de dados ao se inscrever. Stream hot (Channel, SharedFlow) emite dados independentemente — o assinante só recebe o que acontece após a inscrição, sem o início da sequência.

SharedFlow é um Flow hot que pode ter múltiplos assinantes e reproduzir valores recentes quando o replay está configurado. SharedFlow é adequado para eventos (notificações únicas). StateFlow é sua variante com um valor de estado fixo, que armazena em cache o último valor para novos assinantes.

ChannelFlow usa Channel internamente, combinando propriedades de Flow e Channel. Suporta buffer e backpressure através de capacity. ChannelFlow é útil ao converter APIs callback em um fluxo reativo, onde os valores são emitidos de diferentes corrotinas.

Conversão entre cold e hot

Para converter cold Flow em hot SharedFlow usa-se o operador shareIn(scope, started, replay). O parâmetro started controla o momento de início: SharingStarted.WhileSubscribed() — ativo enquanto houver assinantes, Lazily — início no primeiro assinante, Eagerly — início imediato. A conversão inversa — hot para cold: StateFlow.asFlow() retorna um Flow cold que ao fazer collect emite o valor atual do StateFlow. Isso é útil para testes.

Operadores Flow

Flow fornece um rico conjunto de operadores que funcionam como funções suspend dentro de uma corrotina. Os operadores não têm estado e retornam um novo Flow — o fluxo original permanece inalterado. Isso permite construir cadeias de transformação seguras sem efeitos colaterais.

O operador map transforma cada valor do fluxo através de uma transformação assíncrona ou síncrona. filter só passa os valores que satisfazem a condição. catch captura exceções antes do operador terminal e permite recuperar o fluxo. flatMapLatest cancela a emissão anterior quando um novo valor chega — similar ao switchMap no Rx.

O operador debounce no Flow atrasa a publicação do valor por um tempo limite especificado. Se um novo valor chegar durante esse tempo, o temporizador reinicia. No Android, debounce é usado para pesquisa: a solicitação é enviada apenas após uma pausa de 300-400 ms, reduzindo as chamadas de API em 3-5 vezes.

Operadores terminais

Além de collect(), o Flow suporta outros operadores terminais: toList() coleta todos os valores em uma lista — útil para testes, first() retorna o primeiro elemento e cancela o fluxo, single() espera exatamente um elemento. fold(initial) acumula valores através de uma função passada. Todos os operadores terminais são funções suspend e devem ser chamados dentro de uma corrotina ou outra função suspend.

Exemplos de código Flow

O primeiro exemplo — um Flow básico gerando números com transformação através do 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)
        }
}

O segundo exemplo — transformação de fluxo com filtragem e tratamento de erros através de catch:

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

O terceiro exemplo — uso de StateFlow no ViewModel para UI reativa em 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 e SharedFlow

StateFlow é um Flow hot com um único valor atual. Ele armazena em cache o último valor e o passa imediatamente para um novo assinante. StateFlow é um contêiner observável para estado, suporta comparação equals — se o novo valor coincidir com o atual, nenhuma emissão ocorre. Jetpack Compose usa StateFlow através de collectAsState().

SharedFlow é um Flow hot mais flexível sem valor inicial obrigatório. SharedFlow é configurado através de replay (número de valores para novos assinantes), extraBufferCapacity (buffer além do replay) e onBufferOverflow (estratégia em caso de estouro). SharedFlow é ideal para eventos únicos: navegação, Snackbar, análise.

Flow na arquitetura Android é recomendado pelo Google como fonte primária de dados (Camada: Repository → UseCase → ViewModel). LiveData é inferior ao Flow em flexibilidade: Flow suporta corrotinas, operadores, backpressure e funciona fora da camada UI. A migração de LiveData para Flow é uma prática padrão em projetos Android modernos.

Ao usar Flow no ViewModel, é importante escolher o tipo certo. StateFlow é ideal para o estado UI que deve sobreviver à rotação da tela. SharedFlow é adequado para eventos onde o reprocessamento é inaceitável — por exemplo, navegação. Flow com collect() no lifecycleScope dá o máximo controle sobre o contexto de execução, mas requer cancelamento manual ao sair da tela.

O teste do Flow é feito através de kotlinx-coroutines-test. A biblioteca fornece TestDispatcher — tempo virtual que permite acelerar atrasos (delay) e controlar a ordem de execução das corrotinas. TestScope.runTest { } cria um ambiente isolado para testar Flow. O operador toList() é frequentemente usado em testes para coletar todos os valores do flow com um tempo limite, para verificar se o fluxo emitiu a sequência correta de dados.

Flow integra-se bem com Room (biblioteca Android para BD): métodos DAO podem retornar Flow<List<Entity>>. Room emite automaticamente um novo valor a qualquer alteração na tabela — a UI é atualizada sem acionador manual. Isso é implementado através de InvalidationTracker, que internamente usa Flow com callbackFlow. Esta abordagem elimina a necessidade de LiveData e torna a camada de dados totalmente orientada a corrotinas. Jetpack Compose através de collectAsState() subscreve-se ao StateFlow e redesenha apenas os componentes cujos dados mudaram — isto proporciona um desempenho inatingível com arquiteturas orientadas a LiveData. DataStore (substituto do SharedPreferences) também retorna Flow<Preferences>, fornecendo leitura reativa das configurações do aplicativo sem acionadores de atualização manuais.

Flow suporta comunicação entre processos através de kotlinx-coroutines-core na JVM sem bibliotecas adicionais. Por exemplo, em aplicações servidoras com Ktor, o Flow pode representar um fluxo de mensagens WebSocket de entrada. Cada mensagem é emitida no fluxo, passa por filtragem e agregação através de operadores, e o resultado é enviado ao cliente. Esta abordagem substitui bibliotecas reativas como Reactor ou RxJava em projetos Kotlin.

A compatibilidade do Flow com código RxJava existente é fornecida pelo módulo kotlinx-coroutines-rx3. A função de extensão Flow.asObservable() converte Flow em Observable do RxJava 3. A conversão inversa — CompletableSource.asFlow(), Observable.asFlow(). Isto simplifica a migração de RxJava para corrotinas: pode-se reescrever o projeto em etapas, deixando algumas camadas em RxJava. Ao converter, é necessário considerar a diferença na semântica cold/hot: Observable pode ser tanto cold quanto hot, Flow é sempre cold para Flow normal e hot para SharedFlow.

Tratamento de erros e teste do Flow

O tratamento de erros no Flow tem uma particularidade: se uma exceção ocorrer dentro do builder flow antes do operador terminal, ela é propagada para catch. Se uma exceção ocorrer em um operador após o builder, o catch após esse operador a captura. retryWhen permite tentar novamente a inscrição com uma condição: tentar novamente em erro de rede até 3 vezes, mas não tentar novamente em CancellationException. O Flow elimina erros dependentes de estado porque não armazena estado — isto simplifica a depuração em comparação com Observable, onde o Subject armazena estado interno.

O teste do Flow com kotlinx-coroutines-test usa TestDispatcher para simular atrasos. Turbine é uma biblioteca popular da comunidade para testar Flow: test { } lança Flow, awaitItem() aguarda o próximo valor, awaitComplete() aguarda a conclusão. Turbine adiciona um tempo limite padrão, evitando que os testes travem. Para testar StateFlow, use .testIn(scope) com verificação de valores em ordem cronológica.