Kotlin Flow: Asynchronous data streams

1. Overview

A suspend function only returns one value. When you need to receive multiple values over time, for example new orders, payment status or stock prices, you need a data stream. Kotlin Flow solves exactly that.

This article explains what Flow is and how to transform data with operators. Then you’ll learn thread switching with flowOn, error handling, using StateFlow and SharedFlow, and writing tests for Flow.

2. Installation

Flow is in the kotlinx-coroutines library. This article assumes your Maven project already has kotlin-maven-plugin configured. You just need to add the following dependencies:

<dependencies>
    <dependency>
        <groupId>org.jetbrains.kotlinx</groupId>
        <artifactId>kotlinx-coroutines-core</artifactId>
        <version>1.10.2</version>
    </dependency>

    <!-- Test -->
    <dependency>
        <groupId>org.jetbrains.kotlinx</groupId>
        <artifactId>kotlinx-coroutines-test</artifactId>
        <version>1.10.2</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.junit.jupiter</groupId>
        <artifactId>junit-jupiter</artifactId>
        <version>5.11.4</version>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>org.assertj</groupId>
        <artifactId>assertj-core</artifactId>
        <version>3.27.3</version>
        <scope>test</scope>
    </dependency>
</dependencies>

If you’re using Spring Boot 3.x, the coroutines version is already managed via BOM, so you can omit the <version> tag.

3. Your first Flow

3.1. Create and collect Flow

The simplest way to create a Flow is to use the flow { } builder and call emit() for each value:

import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking

data class Order(val id: Int, val amount: Long)

fun orderStream(): Flow<Order> = flow {
    println("Starting to emit orders")
    for (id in 1..3) {
        delay(100)                      // simulate waiting for data from system
        emit(Order(id, id * 50_000L))
    }
}

fun main() = runBlocking {
    val orders = orderStream()
    println("Created flow")
    orders.collect { println("Received: $it") }
}

Explanation:

• orderStream() is not a suspend function. It just returns a description of the data stream.
• Inside flow { } you can call suspend functions like delay().
• collect is a terminal operator. Only when you call it does the code inside flow { } run.

Result:

Created flow
Starting to emit orders
Received: Order(id=1, amount=50000)
Received: Order(id=2, amount=100000)
Received: Order(id=3, amount=150000)

The line “Created flow” appears before “Starting to emit orders”. This proves that creating a Flow doesn’t trigger anything.

3.2. Flow is “cold”

Flow created with flow { } is a cold stream. Each time you collect, the entire code block runs from the start:

fun main() = runBlocking {
    val orders = orderStream()
    orders.collect { }
    orders.collect { }
}

Starting to emit orders
Starting to emit orders

Note: If the flow { } block calls an API or writes to a database, each collect will call it again from the start. This is the source of many “why is the API called twice” bugs. If you want to share a source with multiple collectors, see section 6.

4. Transformation operators

4.1. Intermediate operators

Flow has familiar operators like collections: map, filter, take, onEach… These operators are intermediate. They return a new Flow and are also lazy like the original Flow.

import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.map

fun main() = runBlocking {
    orderStream()
        .filter { it.amount >= 100_000 }
        .map { "Order #${it.id}: ${it.amount} VND" }
        .collect { println(it) }
}

Starting to emit orders
Order #2: 100000 VND
Order #3: 150000 VND

Each element goes through the entire chain filter → map → collect before the next element is emitted. Flow by default runs sequentially in the same coroutine.

Unlike collection’s map, the lambda in Flow’s map is a suspend function. So you can call async functions directly inside:

suspend fun fetchCustomerName(orderId: Int): String {
    delay(50)
    return "Customer $orderId"
}

orderStream()
    .map { fetchCustomerName(it.id) }
    .collect { println(it) }

4.2. Other terminal operators

Besides collect, you can use other terminal operators to get results directly:

import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.fold
import kotlinx.coroutines.flow.toList

fun main() = runBlocking {
    val all = orderStream().toList()
    val firstBig = orderStream().first { it.amount > 60_000 }
    val revenue = orderStream().fold(0L) { acc, order -> acc + order.amount }

    println("Number of orders: ${all.size}, first big order: #${firstBig.id}, revenue: $revenue")
}

...
Number of orders: 3, first big order: #2, revenue: 300000

first { } cancels the Flow as soon as it finds a matching element. So order #3 is never emitted in that collect call.

5. Switch thread with flowOn

By default, code inside flow { } runs on the context of the coroutine that calls collect. To push the data emission to another thread, for example reading a file or database, you use flowOn:

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.flowOn

data class Customer(val name: String)

fun loadCustomers(): Flow<Customer> = flow {
    println("Reading data on ${Thread.currentThread().name}")
    listOf("Alice", "Bob").forEach { emit(Customer(it)) }
}.flowOn(Dispatchers.IO)

fun main() = runBlocking {
    loadCustomers().collect {
        println("Processing ${it.name} on ${Thread.currentThread().name}")
    }
}

Reading data on DefaultDispatcher-worker-1
Processing Alice on main
Processing Bob on main

flowOn only affects the upstream part, that is, the operators before it. The collect part still runs on the context of the caller. This principle is called context preservation.

Note: Don’t use withContext to wrap emit() inside flow { }. The code below will throw IllegalStateException: Flow invariant is violated:

flow {
    withContext(Dispatchers.IO) { emit(1) } // WRONG
}

Always use flowOn. If you really need to emit from multiple different coroutines, use channelFlow { }.

6. Error handling

6.1. catch and onCompletion

Exceptions in Flow propagate down to collect like normal code. The catch operator lets you intercept errors and emit a replacement value:

import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.onCompletion

fun paymentStatus(): Flow<String> = flow {
    emit("PENDING")
    throw IllegalStateException("Payment gateway timeout")
}

fun main() = runBlocking {
    paymentStatus()
        .catch { e -> emit("FAILED: ${e.message}") }
        .onCompletion { cause -> println("Completed, cause=$cause") }
        .collect { println(it) }
}

PENDING
FAILED: Payment gateway timeout
Completed, cause=null

Explanation:

• catch only intercepts errors from upstream. Exceptions thrown inside the lambda of collect won’t be caught by it.
• onCompletion comes after catch. The error was already handled in the previous step, so cause is null. If you swap the two operators, you’ll see cause contains the exception.

6.2. retry

For temporary errors like network loss, you can use retry. This operator collects the upstream Flow from the start:

import java.io.IOException
import kotlinx.coroutines.flow.retry

fun main() = runBlocking {
    var attempts = 0
    flow {
        attempts++
        if (attempts < 3) throw IOException("Lost connection")
        emit("OK after $attempts attempts")
    }
        .retry(3) { e -> e is IOException }
        .collect { println(it) }
}

OK after 3 attempts

The predicate { e is IOException } helps you only retry with network errors. Business errors like IllegalArgumentException won’t be retried. Inside the predicate, you can call delay() to implement backoff.

7. StateFlow and SharedFlow: hot streams

flow { } is cold. But often you need a data source that exists independently of collectors, for example shopping cart state or system events. That’s when you need a hot stream.

7.1. StateFlow

StateFlow always holds a current value. A new collector will immediately receive that value:

import kotlinx.coroutines.cancel
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.launch

class CartService {
    private val _total = MutableStateFlow(0L)
    val total: StateFlow<Long> = _total.asStateFlow()

    fun add(price: Long) = _total.update { it + price }
}

fun main() = runBlocking {
    val cart = CartService()
    val job = launch {
        cart.total.collect { println("Cart total: $it") }
    }
    delay(50)
    cart.add(100_000)
    delay(50)
    cart.add(250_000)
    delay(50)
    job.cancel()
}

Cart total: 0
Cart total: 100000
Cart total: 350000

Things to note:

• The _total property is private and mutable. Outside, you only see the read-only StateFlow. This is the standard pattern for encapsulating state.
• update { } updates atomically, safe when multiple threads call add.
• StateFlow never completes. So you must cancel the job, otherwise runBlocking will wait forever.
• StateFlow skips values identical to the current value. It also conflates: a slow collector only sees the latest value, not intermediate steps.

7.2. SharedFlow

SharedFlow is suitable for events, for example “order has been paid”. Here each emission has meaning and there’s no concept of “current value”:

import kotlinx.coroutines.flow.MutableSharedFlow
import kotlinx.coroutines.flow.SharedFlow

class PaymentService {
    private val _events = MutableSharedFlow<String>()
    val events: SharedFlow<String> = _events

    suspend fun pay(orderId: Int) {
        _events.emit("Order #$orderId has been paid")
    }
}

fun main() = runBlocking {
    val payments = PaymentService()
    payments.pay(0)                       // no one listening yet -> lost

    val job = launch { payments.events.collect { println("Email: $it") } }
    delay(50)
    payments.pay(1)
    payments.pay(2)
    delay(50)
    job.cancel()
}

Email: Order #1 has been paid
Email: Order #2 has been paid

With the default configuration (replay = 0), events emitted when there are no subscribers will be lost. If a late collector needs to receive a few recent events, use MutableSharedFlow<String>(replay = 1).

You can also turn a cold Flow into a hot Flow using stateIn(scope, SharingStarted.WhileSubscribed(5_000), initialValue) or shareIn(...). Then multiple collectors share a single upstream run.

8. Testing Flow

The kotlinx-coroutines-test library provides runTest. This function skips delay using virtual time, so tests run almost instantly:

import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.toList
import kotlinx.coroutines.test.runTest
import org.assertj.core.api.Assertions.assertThat
import org.junit.jupiter.api.Test

class OrderFlowTest {

    @Test
    fun `keep only orders of 100k or more`() = runTest {
        val ids = orderStream()
            .filter { it.amount >= 100_000 }
            .map { it.id }
            .toList()

        assertThat(ids).containsExactly(2, 3)
    }

    @Test
    fun `payment error is converted to FAILED status`() = runTest {
        val states = paymentStatus()
            .catch { emit("FAILED") }
            .toList()

        assertThat(states).containsExactly("PENDING", "FAILED")
    }

    @Test
    fun `shopping cart accumulates values`() = runTest {
        val cart = CartService()
        cart.add(100_000)
        cart.add(250_000)

        assertThat(cart.total.value).isEqualTo(350_000L)
    }
}

For finite Flows, toList() is the simplest way to check all emitted values. For StateFlow, you read .value directly. Don’t call toList() on StateFlow, because this Flow never completes so the test will hang.

9. Common mistakes

• Code after collect doesn’t run. collect is a suspend function and only returns when the Flow completes. With StateFlow/SharedFlow, that never happens. Put each infinite collect in its own launch.
• Calling API multiple times unintentionally. Each collector of a cold Flow runs upstream again. Use shareIn/stateIn when you need to share.
• Using withContext around emit. Replace it with flowOn or channelFlow.
• Thinking catch catches all errors. This operator only catches upstream errors. If you want to catch errors in the handling part, move the logic into onEach { } placed before catch, then call collect() without arguments.
• Using StateFlow for events. Two identical consecutive events will be merged into one. Use SharedFlow for events.

Note: Don’t use Flow when you only have one result. A simple suspend fun findOrder(id: Int): Order is much simpler and more readable than a Flow<Order> that only emits one element.

10. Conclusion

In this article, you’ve seen that Flow is an asynchronous data stream that’s cold and lazy. You’ve learned how to transform data with map/filter, safely switch threads with flowOn, and handle errors with catch, retry, onCompletion. For sharing state and events, StateFlow and SharedFlow are the two main tools. runTest helps you test them quickly and reliably.

From here, you can try advanced operators like combine, flatMapLatest, debounce, or buffer to build more complex data pipelines.