Reactive Programming in Kotlin: Flow and Coroutines |…

Reactive Programming in Kotlin: The Power of kotlinx.coroutines and Flow

Reactive programming is a paradigm focused on asynchronous processing of data streams and propagation of changes. In the Kotlin ecosystem, the primary tool for implementing reactive patterns is the kotlinx.coroutines library, specifically its Flow component. Unlike RxJava, Flow is tightly integrated with coroutines, making the code more concise and safe.

Core Concepts of Flow

Flow is a cold asynchronous data stream that emits values sequentially. It supports:

Cold streams — data generation starts only when there is a subscriber (terminal operator).
Execution context — the stream can switch between different coroutine dispatchers.
Cancellation via coroutines — cancellation of the parent coroutine automatically terminates the Flow.

Setup and Configuration

To use Flow, add the dependency in build.gradle.kts:

dependencies {    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.1")}

Creating and Consuming Flow

Let's create a simple stream of numbers with a delay:

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

fun simpleFlow(): Flow<Int> = flow {    for (i in 1..5) {        delay(100) // simulating asynchronous work        emit(i)    // emit the value    }}

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

In this example, flow { ... } creates a cold stream, and collect is a terminal operator that starts execution.

Data Transformation with Operators

Flow provides a rich set of operators, similar to RxJava:

map — transforms each value.
filter — filters values based on a condition.
flatMapConcat — sequentially merges streams.
catch — handles exceptions within the stream.

Example with operators:

fun transformFlow(): Flow<String> = flow {    emit("Kotlin")    emit("Java")    emit("Python")}.map { it.lowercase() } .filter { it.startsWith("k") }

fun main() = runBlocking {    transformFlow().collect { println(it) }    // Output: kotlin}

Error Handling and Completion

Flow allows elegant exception handling with catch and performing completion actions via onCompletion:

fun errorProneFlow(): Flow<Int> = flow {    emit(1)    emit(2)    throw RuntimeException("Error!")    emit(3) // will not execute}

fun main() = runBlocking {    errorProneFlow()        .catch { e -> println("Caught error: ${e.message}") }        .onCompletion { println("Flow completed") }        .collect { println(it) }}// Output: 1, 2, Caught error: Error!, Flow completed

Hot Streams: StateFlow and SharedFlow

Unlike cold Flow, StateFlow and SharedFlow are hot streams that hold state and can have multiple subscribers.

StateFlow

Used for storing and observing state (e.g., in ViewModel):

class CounterViewModel {    private val _counter = MutableStateFlow(0)    val counter: StateFlow<Int> = _counter.asStateFlow()

    fun increment() {        _counter.value++    }}

// In Activity or FragmentlifecycleScope.launch {    viewModel.counter.collect { value ->        textView.text = "Count: $value"    }}