Flow — What It Is, Cold and Hot Streams in Kotlin Coroutines

What is Flow in Kotlin?

Flow is a type from the kotlinx.coroutines.flow package, representing a cold asynchronous data stream. At its core, Flow is a coroutine sequence that emits values via the emit() function and completes either successfully or with an exception. Stream collection is performed via the terminal operator collect(), which is a suspend function.

Cold Semantics

Cold stream means the code inside the flow builder runs anew for each subscriber. Observable.fromIterable in RxJava behaves similarly: a new subscriber receives all values from the start. In Flow, this is implemented via the suspend function collect, which blocks the coroutine for the entire data collection duration.

Flow Builders

Kotlin provides several ways to create a Flow: flow { } — the basic construct with emit(), flowOf(vararg values) — for a fixed set of values, .asFlow() — an extension for collections and Sequence. All builders are cold — data is generated only when the terminal operator is called.

Cold and Hot Streams

The distinction between cold and hot streams is a key concept of reactive programming. Cold stream (Flow, Observable) starts generating data upon subscription. Hot stream (Channel, SharedFlow) emits data independently — the subscriber only receives what happens after subscription, without the beginning of the sequence.

SharedFlow is a hot Flow that can have multiple subscribers and replay recent values when replay is configured. SharedFlow is suitable for events (one-shot notifications). StateFlow is its variant with a fixed state value, caching the latest value for new subscribers.

ChannelFlow uses Channel under the hood, combining properties of Flow and Channel. It supports buffering and backpressure via capacity. ChannelFlow is useful when converting callback APIs into a reactive stream, where values are emitted from different coroutines.

Converting Between Cold and Hot

To convert cold Flow to hot SharedFlow, use the shareIn(scope, started, replay) operator. The started parameter controls the launch moment: SharingStarted.WhileSubscribed() — active while there are subscribers, Lazily — launch on the first subscriber, Eagerly — immediate launch. The reverse conversion — hot to cold: StateFlow.asFlow() returns a cold Flow that emits the current StateFlow value upon collect. This is convenient for testing.

Flow Operators

Flow provides a rich set of operators that work as suspend functions inside a coroutine. Operators are stateless and return a new Flow — the original stream remains unchanged. This allows building safe transformation chains without side effects.

The map operator transforms each stream value through an asynchronous or synchronous transformation. filter passes only values that satisfy the condition. catch catches exceptions before the terminal operator and allows stream recovery. flatMapLatest cancels the previous emission when a new value arrives — similar to switchMap in Rx.

The debounce operator in Flow delays value publication by a specified timeout. If a new value arrives during this time, the timer resets. In Android, debounce is used for search: the request is sent only after a 300-400 ms pause, which reduces API calls by 3-5 times.

Terminal Operators

In addition to collect(), Flow supports other terminal operators: toList() collects all values into a list — useful for tests, first() returns the first element and cancels the stream, single() expects exactly one element. fold(initial) accumulates values through a passed function. All terminal operators are suspend functions and must be called inside a coroutine or another suspend function.

Flow Code Examples

The first example — a basic Flow generating numbers with transformation via the map operator:

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

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

The second example — stream transformation with filtering and error handling via catch:

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

The third example — using StateFlow in ViewModel for reactive UI in 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 and SharedFlow

StateFlow is a hot Flow with a single current value. It caches the latest value and passes it to a new subscriber immediately. StateFlow is an observable container for state, supports equals comparison — if the new value matches the current one, no emission occurs. Jetpack Compose uses StateFlow via collectAsState().

SharedFlow is a more flexible hot Flow without a mandatory initial value. SharedFlow is configured via replay (number of values for new subscribers), extraBufferCapacity (buffer beyond replay), and onBufferOverflow (strategy on overflow). SharedFlow is ideal for one-shot events: navigation, Snackbar, analytics.

Flow in Android architecture is recommended by Google as the primary data source (Layer: Repository → UseCase → ViewModel). LiveData falls short of Flow in flexibility: Flow supports coroutines, operators, backpressure and works outside the UI layer. Migration from LiveData to Flow is standard practice in modern Android projects.

When using Flow in ViewModel, it is important to choose the right type. StateFlow is ideal for UI state that should survive screen rotation. SharedFlow is suitable for events where reprocessing is unacceptable — for example, navigation. Flow with collect() in lifecycleScope gives maximum control over the execution context but requires manual cancellation when leaving the screen.

Testing Flow is done via kotlinx-coroutines-test. The library provides TestDispatcher — virtual time that allows accelerating delays (delay) and controlling coroutine execution order. TestScope.runTest { } creates an isolated environment for testing Flow. The toList() operator is often used in tests to collect all flow values with a timeout, to verify that the stream emitted the correct data sequence.

Flow integrates well with Room (Android database library): DAO methods can return Flow<List<Entity>>. Room automatically emits a new value on any table change — the UI updates without a manual trigger. This is implemented via InvalidationTracker, which under the hood uses Flow with callbackFlow. This approach eliminates the need for LiveData and makes the data layer fully coroutine-oriented. Jetpack Compose via collectAsState() subscribes to StateFlow and redraws only those components whose data has changed — this gives performance unattainable with LiveData-oriented architectures. DataStore (replacement for SharedPreferences) also returns Flow<Preferences>, providing reactive reading of app settings without manual update triggers.

Flow supports inter-process communication via kotlinx-coroutines-core on JVM without additional libraries. For example, in Ktor server applications, Flow can represent an incoming WebSocket message stream. Each message is emitted into the stream, goes through filtering and aggregation via operators, and the result is sent to the client. This approach replaces reactive libraries like Reactor or RxJava in Kotlin projects.

Flow compatibility with existing RxJava code is provided by the kotlinx-coroutines-rx3 module. The extension function Flow.asObservable() converts Flow to RxJava 3 Observable. Reverse conversion — CompletableSource.asFlow(), Observable.asFlow(). This simplifies migration from RxJava to coroutines: you can rewrite the project incrementally, leaving some layers on RxJava. When converting, consider the difference in cold/hot semantics: Observable can be both cold and hot, Flow is always cold for regular Flow and hot for SharedFlow.

Flow Error Handling and Testing

Error handling in Flow has a specific behavior: if an exception occurs inside the flow builder before the terminal operator, it is propagated to catch. If an exception occurs in an operator after the builder, it is caught by the catch after that operator. retryWhen allows retrying the subscription with a condition: retry on network error up to 3 times, but do not retry on CancellationException. Flow eliminates state-dependent errors because it does not store state — this simplifies debugging compared to Observable, where Subject stores internal state.

Testing Flow with kotlinx-coroutines-test uses TestDispatcher to simulate delays. Turbine is a popular community library for testing Flow: test { } launches Flow, awaitItem() waits for the next value, awaitComplete() waits for completion. Turbine adds a default timeout, preventing test hangs. For testing StateFlow, use .testIn(scope) with value verification in chronological order.