Coroutines & Async

Kotlin coroutines, suspend functions, and structured concurrency — async programming the Kotlin way

Kotlin coroutines provide a way to write asynchronous, non-blocking code sequentially. They are lightweight threads managed by the Kotlin runtime.

Adding the dependency

// build.gradle.kts
dependencies {
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.1")
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-android:1.8.1") // Android
}

Launching coroutines

import kotlinx.coroutines.*

// Launch in a scope — structured concurrency
fun main() = runBlocking {              // blocks until all children complete
    launch {                            // fire-and-forget coroutine
        delay(1000L)
        println("World!")
    }
    println("Hello,")
}

// With a dispatcher
launch(Dispatchers.Default) { /* CPU-intensive */ }
launch(Dispatchers.IO) { /* blocking I/O */ }
launch(Dispatchers.Main) { /* UI thread (Android) */ }

// Async — returns a result
val deferred: Deferred<Int> = async {
    delay(1000L)
    42
}
val result = deferred.await()           // suspends until result is ready

Suspend functions

// Mark functions that perform async work with suspend
suspend fun fetchData(): String {
    delay(1000L)                         // non-blocking wait
    return "data"
}

// Suspend functions can only be called from coroutines or other suspend functions
suspend fun loadData() {
    val data = fetchData()
    println(data)
}

Coroutine scopes

// Structured concurrency — coroutines are scoped
// When a scope is cancelled, all children are cancelled too

// Custom scope
val scope = CoroutineScope(Dispatchers.Default)

// In a class
class MyService(private val scope: CoroutineScope) {
    fun start() {
        scope.launch {
            // work
        }
    }
}

// coroutineScope — suspends until all children complete
suspend fun fetchAll() = coroutineScope {
    val a = async { fetchA() }
    val b = async { fetchB() }
    a.await() + b.await()               // both run in parallel
}

// supervisorScope — children fail independently
suspend fun fetchAllResilient() = supervisorScope {
    val a = async { fetchA() }
    val b = async { fetchB() }
    try { a.await() } catch (e: Exception) { null }
    try { b.await() } catch (e: Exception) { null }
}

Cancellation and timeouts

// Cancellation
val job = launch {
    repeat(1000) {
        ensureActive()                    // check for cancellation
        // or yield() to check and give up thread
        println("working $it")
    }
}
job.cancel()                              // cancel the coroutine
job.join()                                // wait for completion
job.cancelAndJoin()                       // cancel and wait

// Timeout
withTimeout(3000L) {
    // cancel if this takes more than 3 seconds
    val result = longRunningOperation()
}

withTimeoutOrNull(3000L) {
    longRunningOperation()
} ?: "default"                            // returns null on timeout

Flow

Flow is the cold asynchronous stream — analogous to RxJava’s Observable.

import kotlinx.coroutines.flow.*

// Creating a flow
val numbers = flow {
    for (i in 1..5) {
        emit(i)
        delay(100L)
    }
}

// Flow from a collection
val flow = listOf(1, 2, 3).asFlow()

// Consume
numbers.collect { println(it) }          // terminal operator — triggers emission

// Transform
numbers.map { it * 2 }
       .filter { it > 2 }
       .collect { println(it) }

// Transform with access to emitter
numbers.transform { value ->
    emit(value)
    emit(value * 10)
}.collect { println(it) }

// Take
numbers.take(3).collect { println(it) }  // first 3 values only

// Terminal operators
val list = numbers.toList()               // collect into a list
val first = numbers.first()               // first emitted value
val result = numbers.single()             // single emitted value
val count = numbers.count()              // count of values
val reduced = numbers.reduce { acc, v -> acc + v }

// Flatten
val flowOfFlows = (1..3).asFlow().map { flowOf(it, it * 10) }
flowOfFlows.flattenConcat().collect { println(it) }  // 1, 10, 2, 20, 3, 30 (sequential)
flowOfFlows.flattenMerge().collect { println(it) }   // concurrent

SharedFlow and StateFlow

Hot flows that share values across multiple collectors.

// StateFlow — holds a single updatable state
val state = MutableStateFlow(0)
state.value = 42                           // update state
state.collect { println(it) }             // collectors receive current + updates

// SharedFlow — broadcast events to multiple collectors
val events = MutableSharedFlow<String>()
events.tryEmit("hello")                    // non-suspend emit
events.emit("world")                       // suspend emit
events.collect { println(it) }             // each collector gets events

Error handling

// Try-catch in coroutines
try {
    withTimeout(1000L) {
        longRunningOperation()
    }
} catch (e: TimeoutCancellationException) {
    println("Timed out")
} catch (e: CancellationException) {
    throw e                                // never catch CancellationException!
}

// catch on Flow
numbers.catch { e -> emit(-1) }            // emit fallback on error
       .collect { println(it) }

// Result type for flows
numbers.map { if (it > 3) throw RuntimeException("too big") else it }
       .catch { e -> emit(-1) }
       .collect { println(it) }

Channels

Channels are for passing values between coroutines — like a blocking queue.

val channel = Channel<Int>()

launch {
    channel.send(1)
    channel.send(2)
    channel.close()                        // signal no more values
}

launch {
    for (value in channel) {               // receive until closed
        println(value)
    }
}

// Produce — returns a ReceiveChannel
val squares = produce {
    for (x in 1..5) send(x * x)
}
for (value in squares) println(value)