Kotlin Flow — Reactive Data Streams Guide

In this tutorial, you will learn about Kotlin Flow. We cover key concepts, practical examples, and best practices to help you master this topic.

Kotlin Flow is a cold asynchronous data stream that emits multiple values over time, supporting rich operators like map, filter, flatMapLatest, and combine, with structured concurrency and cancellation propagation.

What You'll Learn

• Create flows with flow builders
• Transform data with flow operators
• Collect flows safely in different scopes
• Understand cold versus hot flows
• Use StateFlow and SharedFlow for UI state
• Combine multiple flows
• Handle errors and retry in flows
• Test flows with Turbine or kotlinx-test

Why It Matters

Modern apps consume continuous data streams: location updates, database changes, sensor readings, and UI events. Flow provides a Kotlin-native way to handle these streams with structured concurrency. Unlike LiveData, Flow works across all platforms and integrates with coroutines. Unlike RxJava, Flow has a simpler API and zero external dependencies.

Real-World Use

DodaTech uses Flow for real-time malware scanning results, database observation with Room, and sensor data from device sensors. StateFlow drives the Compose UI with automatic recomposition. SharedFlow broadcasts system events like "settings changed" to multiple subscribers.

Learning Path

flowchart LR
  A[Coroutines] --> B[Flow\nYou are here]
  B --> C[KMP Basics]
  style B fill:#f90,color:#fff

Creating a Flow

Flow emits values sequentially, completing when the block finishes.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun simpleFlow(): Flow<Int> = flow {
    println("Flow started")
    for (i in 1..3) {
        delay(200)
        emit(i)
    }
}

fun main() = runBlocking {
    simpleFlow().collect { value ->
        println("Received: $value")
    }
}

Output:

Flow started
Received: 1
Received: 2
Received: 3

Flow is cold: the block runs only when collect is called. Each collector triggers a fresh execution.

Flow Builders

fun main() = runBlocking {
    // flowOf: emit fixed values
    flowOf("apple", "banana", "cherry").collect { println(it) }
    
    // asFlow: convert collections and sequences
    listOf(1, 2, 3).asFlow().collect { println(it) }
    
    // IntRange.asFlow
    (1..5).asFlow().collect { println(it) }
    
    // callbackFlow: bridge callback APIs
    // (used for location listeners, sensor events, etc.)
}

Output: Each Builder creates a cold flow that emits the specified values.

Flow Operators

Flow provides declarative operators similar to Kotlin sequences and collections.

fun main() = runBlocking {
    val numbers = (1..10).asFlow()
    
    // Map: transform each value
    numbers
        .map { it * it }
        .collect { print("$it ") }
    println()  // Output: 1 4 9 16 25 36 49 64 81 100
    
    // Filter: keep matching values
    numbers
        .filter { it % 2 == 0 }
        .collect { print("$it ") }
    println()  // Output: 2 4 6 8 10
    
    // Take: limit emissions
    numbers
        .take(3)
        .collect { print("$it ") }
    println()  // Output: 1 2 3
    
    // Drop: skip initial emissions
    numbers
        .drop(7)
        .collect { print("$it ") }
    println()  // Output: 8 9 10
}

Output: Each operator returns a new flow without modifying the original.

FlatMap Operators

flatMapConcat, flatMapMerge, and flatMapLatest handle inner flows.

fun requestData(id: Int): Flow<String> = flow {
    emit("Loading $id")
    delay(500)
    emit("Data for $id")
}

fun main() = runBlocking {
    val ids = (1..3).asFlow()
    
    println("flatMapConcat (sequential):")
    ids.flatMapConcat { requestData(it) }.collect { println(it) }
    // Total time: ~1500ms
    
    println("\nflatMapMerge (concurrent):")
    ids.flatMapMerge { requestData(it) }.collect { println(it) }
    // Total time: ~500ms
    
    println("\nflatMapLatest (cancels previous):")
    ids.flatMapLatest { requestData(it) }.collect { println(it) }
    // Only last ID completes
}

Output: flatMapConcat processes sequentially. flatMapMerge processes concurrently. flatMapLatest cancels the previous inner flow when a new value arrives.

Combining Flows

Combine multiple flows into one.

fun main() = runBlocking {
    val flow1 = (1..3).asFlow().onEach { delay(200) }
    val flow2 = flowOf("A", "B", "C").onEach { delay(300) }
    val flow3 = flowOf("X", "Y", "Z").onEach { delay(150) }
    
    // Combine: emits when any flow emits (with the latest from each)
    flow1.combine(flow2) { a, b ->
        "$a -> $b"
    }.collect { println(it) }
    
    println("---")
    
    // Zip: pairs elements (waits for both)
    flow1.zip(flow2) { a, b ->
        "($a, $b)"
    }.collect { println(it) }
}

Output: combine emits on every emission from any source. zip emits in pairs, waiting for both flows to have a new value.

Error Handling

Handle errors in flow with catch, retry, and onCompletion.

fun unstableFlow(): Flow<Int> = flow {
    emit(1)
    emit(2)
    throw RuntimeException("Network error")
    emit(3)
}

fun main() = runBlocking {
    // Catch errors and emit a fallback
    unstableFlow()
        .catch { e ->
            println("Caught: ${e.message}")
            emit(-1)  // Fallback value
        }
        .collect { println(it) }
    
    println("---")
    
    // Retry on error
    var attempt = 0
    val retryingFlow = flow {
        attempt++
        if (attempt < 3) throw RuntimeException("Attempt $attempt failed")
        emit(42)
    }
    
    retryingFlow
        .retry(3) { cause ->
            println("Retrying after: ${cause.message}")
            true  // true means retry
        }
        .collect { println("Success: $it") }
}

Output: catch replaces the error with a fallback value. retry re-executes the flow block up to the specified number of times.

StateFlow

StateFlow is a hot flow that holds a single current value.

class TemperatureSensor {
    private val _temperature = MutableStateFlow(22.0)
    val temperature: StateFlow<Double> = _temperature.asStateFlow()
    
    fun startSensing() {
        // Simulate temperature changes
        CoroutineScope(Dispatchers.Default).launch {
            while (true) {
                delay(1000)
                _temperature.value = 20.0 + Math.random() * 10
            }
        }
    }
}

fun main() = runBlocking {
    val sensor = TemperatureSensor()
    sensor.startSensing()
    
    val job = launch {
        sensor.temperature.collect { temp ->
            println("Temperature: ${String.format("%.1f", temp)} C")
        }
    }
    
    delay(3000)
    println("Current temp: ${String.format("%.1f", sensor.temperature.value)} C")
    job.cancel()
}

Output: Temperature updates print every second. The current value is accessible via .value without collecting.

StateFlow always has a value. It emits the same value to new collectors immediately.

SharedFlow

SharedFlow is a hot flow that broadcasts events to multiple collectors.

class EventBus {
    private val _events = MutableSharedFlow<String>(
        replay = 0,  // Don't replay old events
        extraBufferCapacity = 10
    )
    val events: SharedFlow<String> = _events.asSharedFlow()
    
    fun sendEvent(event: String) {
        _events.tryEmit(event)
    }
}

fun main() = runBlocking {
    val bus = EventBus()
    
    // Collector 1
    launch {
        bus.events.collect { println("Collector 1: $it") }
    }
    
    // Collector 2 (joins later)
    launch {
        delay(500)
        bus.events.collect { println("Collector 2: $it") }
    }
    
    delay(200)
    bus.sendEvent("Event 1")
    delay(200)
    bus.sendEvent("Event 2")
    delay(200)
    bus.sendEvent("Event 3")
    
    delay(1000)
}

Output: Collector 1 receives all events. Collector 2 receives only events emitted after it started collecting.

Testing Flows

Use Turbine or the built-in kotlinx-coroutines-test for flow testing.

// build.gradle.kts
// testImplementation("app.cash.turbine:turbine:1.1.0")

class FlowTest {
    @Test
    fun testStateFlow() = runTest {
        val viewModel = MyViewModel()
        
        viewModel.state.test {
            assertEquals(UiState.Loading, awaitItem())
            assertEquals(UiState.Success("data"), awaitItem())
            awaitComplete()
        }
    }
}

Common Mistakes

1. Collecting flows from a non-Coroutine context: collect() is a suspend function and must be called from a coroutine scope. Use .launchIn(scope) for one-shot collection.

2. Using StateFlow for event-like one-shot emissions: StateFlow emits the current value to new collectors. Use SharedFlow with replay=0 for one-shot events.

3. Not using conflate or distinctUntilChanged for performance: Emitting duplicate values triggers unnecessary recomposition. Use distinctUntilChanged() to suppress duplicates.

4. Blocking the flow with long operations in operators: Operators like map and filter should not perform blocking operations. Use flatMapLatest with a new coroutine context for heavy work.

5. Forgetting to handle backpressure: If a collector is slower than the emitter, consider using conflate(), buffer(), or collectLatest().

6. Sharing the same flow without sharing Strategy: Multiple collectors trigger independent executions for cold flows. Use shareIn() or stateIn() to share a single execution.

Practice Questions

1. What is the difference between a cold flow and a hot flow?

Answer: A cold flow starts emitting only when collected, and each collector triggers an independent execution. A hot flow emits regardless of collectors and shares emissions among them.

2. What is the difference between StateFlow and SharedFlow?

Answer: StateFlow always has a single current value accessible via .value. SharedFlow can have zero replay and emits events without storing the latest value.

3. How do you handle errors in a flow?

Answer: Use the catch() operator to catch upstream errors and emit fallback values. Use retry() to re-execute the flow on failure.

4. What does flatMapLatest do?

Answer: When a new value arrives from the source, flatMapLatest cancels the previous inner flow and starts a new one with the latest value.

5. Challenge: Implement an auto-saving text editor. The flow emits text changes. After the user stops typing for 500ms, save the text to a simulated database. Use debounce, flatMapLatest, and distinctUntilChanged.

Answer:

fun autoSaveFlow(textChanges: Flow<String>): Flow<String> = textChanges
    .debounce(500)
    .distinctUntilChanged()
    .flatMapLatest { text ->
        flow {
            emit("Saving...")
            delay(1000)  // Simulate save
            emit("Saved: ${text.take(20)}...")
        }
    }

fun main() = runBlocking {
    val textChanges = flowOf(
        "H", "He", "Hel", "Hell", "Hello",
        "Hello ", "Hello W", "Hello Wo", "Hello Wor", "Hello Worl", "Hello World"
    ).onEach { delay(100) }
    
    autoSaveFlow(textChanges).collect { println(it) }
}

Output: Only the final text ("Hello World") triggers a save. Intermediate values are debounced away.

Mini Project

Build a real-time search with Flow. Requirements:

• Emit search query changes as a Flow
• Debounce by 300ms
• Distinct until changed
• FlatMapLatest to cancel previous search requests
• Simulated API delay of 500ms
• StateFlow for results and loading state
• Error handling with catch and retry
• Show loading indicator, results, and error states in console

This project applies flow operators, error handling, and state management in a realistic search scenario.

FAQ

What's Next

After mastering Flow, learn KMP basics for cross-platform development. You can also explore Ktor for server-side networking.