Kotlin Flow

Flow is Kotlin's way to handle streams of values over time in a cold, declarative, and composable way. Where a suspend function returns one value, a Flow returns multiple values — one after another. Flows are lazy: they produce values only when collected.

Flow vs Suspend Function

suspend function:
  suspend fun getNumber(): Int = 42       // returns ONE value

Flow:
  fun getNumbers(): Flow = flow {
      emit(1)
      emit(2)
      emit(3)
  }                                       // returns MULTIPLE values over time

Creating a Flow

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

fun countdown(): Flow = flow {
    for (i in 5 downTo 1) {
        delay(500)   // wait between each value
        emit(i)      // emit sends one value downstream
    }
}

fun main() = runBlocking {
    countdown().collect { value ->
        println("T-$value")
    }
    println("Liftoff!")
}

Output:

T-5
T-4
T-3
T-2
T-1
Liftoff!

Flow is Cold

val numbers = flow { emit(1); emit(2); emit(3) }

// Nothing runs yet — Flow is just a blueprint

numbers.collect { println(it) }   // NOW it runs
numbers.collect { println(it) }   // runs AGAIN from scratch (cold)

Flow Operators

fun numbers(): Flow = flow {
    for (i in 1..10) emit(i)
}

fun main() = runBlocking {
    numbers()
        .filter { it % 2 == 0 }        // keep even numbers
        .map { it * it }               // square them
        .take(3)                       // take only 3
        .collect { println(it) }       // 4, 16, 36
}

Terminal Operators

fun nums(): Flow = flow { for (i in 1..5) emit(i) }

fun main() = runBlocking {
    println(nums().toList())              // [1, 2, 3, 4, 5]
    println(nums().first())              // 1
    println(nums().count())              // 5
    println(nums().fold(0) { a, v -> a + v })  // 15
    nums().forEach { print("$it ") }     // 1 2 3 4 5
}

flowOf and asFlow

// flowOf: create a fixed Flow from values
val colors = flowOf("red", "green", "blue")

// asFlow: convert a collection to a Flow
val numbers = listOf(10, 20, 30).asFlow()

// asFlow on a range
val range = (1..5).asFlow()

fun main() = runBlocking {
    colors.collect { println(it) }
}

Transform: flatMapConcat

fun details(id: Int): Flow = flow {
    delay(100)
    emit("Item $id: name")
    emit("Item $id: price")
}

fun main() = runBlocking {
    flowOf(1, 2, 3)
        .flatMapConcat { details(it) }
        .collect { println(it) }
}

Output:

Item 1: name
Item 1: price
Item 2: name
Item 2: price
Item 3: name
Item 3: price

Exception Handling in Flow

fun riskyFlow(): Flow = flow {
    emit(1)
    emit(2)
    throw RuntimeException("Network error")
    emit(3)   // never reached
}

fun main() = runBlocking {
    riskyFlow()
        .catch { e -> println("Caught: ${e.message}") }
        .collect { println(it) }
}

Output:

1
2
Caught: Network error

Practical Example: Live Stock Price Feed

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

fun stockPriceFeed(ticker: String): Flow = flow {
    val basePrices = mapOf("AAPL" to 182.0, "GOOG" to 175.0, "TSLA" to 250.0)
    var price = basePrices[ticker] ?: 100.0
    repeat(5) {
        price += (-2..2).random()   // simulate price fluctuation
        delay(400)
        emit(price)
    }
}

fun main() = runBlocking {
    stockPriceFeed("AAPL")
        .onEach { println("AAPL: $${"%.2f".format(it)}") }
        .filter { it > 181.0 }
        .take(2)
        .collect { println("  → BUY signal at $${"%.2f".format(it)}") }
}