Mastering Kotlin Asynchronous Streams: The Core of Android Reactive Programming

Cold Flow Deep Dive

Cold Flow is a suspendable code block that runs only when a collector invokes collect. Each collector triggers a fresh execution of the flow { … } block, so no data exists before collection.

Core mental model: Cold Flow is a recipe executed on demand; it holds no history.

Independent execution

val coldNumberFlow = flow {
    println("▶️ Starting recipe…")
    for (i in 1..3) {
        delay(500)
        println("📤 Emitting: $i")
        emit(i)
    }
    println("⏹️ Recipe finished.")
}

runBlocking {
    launch { // Collector A
        println("👤 Collector A start")
        coldNumberFlow.collect { println("📥 A received: $it") }
    }
    delay(700)
    launch { // Collector B joins later
        println("👥 Collector B start")
        coldNumberFlow.collect { println("📥 B received: $it") }
    }
}

Output shows that Collector B receives the sequence from the beginning, proving that each collector gets an independent execution and that no values are replayed.

Memory usage with billions of emissions

The emit function suspends the producer, hands the value to the collector, and then the value becomes eligible for garbage collection as soon as the collector finishes processing it (unless the collector stores it). The emission cycle is:

Producer yields: emit suspends.

Collector processes: the collect lambda runs synchronously on the received value.

Collector yields back: after processing (e.g., println), the coroutine resumes.

Producer continues: the flow block resumes and emits the next value.

Because the collector does not retain references, memory consumption stays constant regardless of the number of emitted items.

Buffered operators insert an internal Channel

val bufferedFlow = hugeFlow
    .buffer(capacity = 64) // buffer
    .flowOn(Dispatchers.IO) // thread switch

Both .buffer() and .flowOn() create a real Channel between producer and consumer. The buffer belongs to the operator, not to the flow itself, and each collector gets its own isolated channel instance.

Hot Flow Deep Dive

Hot flows emit regardless of collectors, making them suitable for shared state such as user profiles or playback status.

Core mental model: A hot flow is like a live radio station; data is emitted and disappears unless a replay buffer is configured.

StateFlow – latest‑value holder

val state = MutableStateFlow("Init")
state.value = "A"
state.value = "B"
state.value = "B" // no emission because value unchanged
state.collect { println("Received State: $it") }
// Output: Received State: B

StateFlow always stores a single latest value and deduplicates identical emissions. New collectors receive only the most recent value.

SharedFlow – configurable replay

val shared = MutableSharedFlow<Int>(
    replay = 0,               // history slots
    extraBufferCapacity = 0, // extra buffer for producers
    onBufferOverflow = BufferOverflow.SUSPEND
)

Key configurations: replay = 3 keeps the last three items in RAM for new collectors. replay = 0 disables replay; emitted items are lost if no collector is present.

Converting Cold Flow to Hot Flow

val hotShared = coldFlow.shareIn(
    scope = viewModelScope,
    started = SharingStarted.WhileSubscribed(stopTimeoutMillis = 5000),
    replay = 1
)

Mechanism:

Background collector: shareIn launches a hidden coroutine that continuously collects the cold flow.

Broadcast distribution: each emitted value is forwarded to an internal SharedFlow.

Startup strategy: WhileSubscribed(5000) keeps the upstream collection alive for 5 s after the last subscriber disappears, preventing unnecessary re‑execution on configuration changes.

Channel Deep Dive

Channels are coroutine‑safe unicast FIFO queues. Each element can be consumed only once.

Core mental model: Flow is a broadcast system; Channel is a single‑consumer pipeline.

Apple‑grabbing demonstration

runBlocking {
    val channel = Channel<Int>(capacity = Channel.BUFFERED) // default 64
    launch { // producer
        for (i in 1..6) {
            println("📤 Send apple $i")
            channel.send(i)
            delay(100)
        }
        channel.close()
    }
    launch { // consumer A (slow)
        for (item in channel) {
            println("👤 A got apple: $item 🍎")
            delay(150)
        }
    }
    launch { // consumer B (fast)
        for (item in channel) {
            println("👥 B got apple: $item 🍏")
            delay(120)
        }
    }
}

Each number is received by exactly one consumer; the other consumer never sees it, illustrating the unicast nature.

Channel capacity modes

Rendezvous (capacity = 0): No buffer; send suspends until a receiver calls receive.

Buffered (default 64): Fixed‑size FIFO; send returns immediately while space exists, otherwise the producer suspends.

Unlimited (capacity = Int.MAX_VALUE): Unbounded linked list; send never suspends but can cause OOM if the consumer is slow.

Conflated (capacity = Channel.CONFLATED): Single‑slot buffer that always holds the latest value, similar to StateFlow.

Data‑Loss Pitfalls and Remedies

Cold Flow loss caused by operator buffering

coldFlow
    .buffer(capacity = 2, onBufferOverflow = BufferOverflow.DROP_OLDEST)
    .collect { delay(1000); println("Processed: $it") }

When the producer is fast and the collector is slow, the internal channel fills. With DROP_OLDEST, the oldest buffered element is discarded as new elements arrive.

Hot Flow loss – Android lifecycle gap

val singleEvent = MutableSharedFlow<UiEvent>(replay = 0)
singleEvent.emit(UiEvent.ShowToast("Saved")) // emitted before any collector
// After rotation a new Activity starts collecting
singleEvent.collect { /* never receives */ }

Because replay = 0, the event is dropped when no collector exists. During an Activity rotation the old Activity is destroyed (collectors = 0) before the new one starts collecting, causing the UI event to disappear.

Solution A: Emit on Dispatchers.Main.immediate so the emission runs synchronously on the main thread, eliminating the tiny scheduling gap. Solution B: For one‑time UI events, prefer a buffered Channel (or receiveAsFlow() ) instead of SharedFlow , because a channel guarantees delivery to a single consumer.

Channel loss with consumeAsFlow()

When a collector of channel.consumeAsFlow() is cancelled, the underlying channel is closed and all buffered items are lost, leading to ClosedReceiveChannelException on the next collection. Golden rule: Never use consumeAsFlow() when you need multiple collection cycles. Use receiveAsFlow() instead; it does not close the original channel on cancellation.

Design Philosophy: Committed‑Delivery vs. Fire‑and‑Forget

Committed‑Delivery (Channel send ): Guarantees that a value is stored or the producer suspends until a consumer receives it.

Fire‑and‑Forget (Flow emit , Channel trySend , SharedFlow with replay = 0 ): Sends the value without waiting; if no consumer is ready, the value is dropped.

Real‑World Refactor: From Cold Flow to Hot StateFlow

Before refactor, each UI component called a cold flow from the repository, causing duplicate network requests:

class UserRepository(private val api: UserApi) {
    fun users(): Flow<List<User>> = flow {
        println("📡 Requesting users…")
        emit(api.fetchUsers())
    }
}

class UsersViewModel(private val repo: UserRepository) : ViewModel() {
    val usersFlow = repo.users()
}

Both UI A and UI B collect usersFlow , triggering two identical API calls. After refactor, the cold flow is converted to a shared StateFlow that is collected once and cached:

class UsersViewModel(private val repo: UserRepository) : ViewModel() {
    val uiState: StateFlow<UsersUiState> = repo.users()
        .map { UsersUiState.Success(it) }
        .catch { emit(UsersUiState.Error(it)) }
        .stateIn(
            scope = viewModelScope,
            started = SharingStarted.WhileSubscribed(stopTimeoutMillis = 5000),
            initialValue = UsersUiState.Loading
        )
}

sealed interface UsersUiState {
    object Loading : UsersUiState
    data class Success(val data: List<User>) : UsersUiState
    data class Error(val throwable: Throwable) : UsersUiState
}

Benefits:

Both UI A and UI B share the same state; the network request runs only once.

The WhileSubscribed(5000) strategy keeps the upstream coroutine alive for 5 s after the last subscriber disappears, so a rotated Activity instantly receives the cached Success state without a second request.

Key Take‑aways

Cold flows never retain history: they re‑run the recipe on each collection, so emitting billions of items does not increase memory usage.

Hot flows are fire‑and‑forget unless you configure replay or state: to keep history you must set replay (for SharedFlow) or use StateFlow for the latest value.

Channels deliver each item exactly once: they are ideal for single‑event communication; for multi‑consumer scenarios use Flow.