Kotlin Coroutines

Kotlin Coroutines & Flow

Basic Coroutine Builders

// launch — fire and forget
viewModelScope.launch {
    repository.refresh()
}

// async — concurrent work with result
val deferred: Deferred<User> = async { userApi.getUser(id) }
val user = deferred.await()

// Parallel decomposition
coroutineScope {
    val user = async { userRepo.findById(id) }
    val orders = async { orderRepo.findByUserId(id) }
    UserWithOrders(user.await(), orders.await())
}

// withContext — switch dispatcher
val result = withContext(Dispatchers.IO) {
    File("data.txt").readText()
}

Structured Concurrency

class UserService(private val scope: CoroutineScope) {

    // Child coroutines are cancelled when scope is cancelled
    fun startSync() = scope.launch {
        while (isActive) {
            try {
                syncData()
                delay(60_000)
            } catch (e: CancellationException) {
                throw e  // always rethrow CancellationException
            } catch (e: Exception) {
                log.warn("Sync failed", e)
            }
        }
    }

    // supervisorScope — child failures don't cancel siblings
    suspend fun fetchBatch(ids: List<Long>): List<Result<User>> =
        supervisorScope {
            ids.map { id ->
                async {
                    runCatching { userApi.getUser(id) }
                }
            }.awaitAll()
        }
}

Flow Basics

// Cold flow — executed per collector
fun userFlow(id: Long): Flow<User> = flow {
    while (true) {
        emit(userRepo.findById(id))
        delay(5_000)
    }
}

// Operators
userFlow(id)
    .filter { it.isActive }
    .map { it.toDto() }
    .catch { e -> emit(User.empty()) }
    .onEach { log.debug("Got user: $it") }
    .collect { render(it) }

// flowOn — upstream runs on given dispatcher
repository.observeAll()
    .map { transform(it) }
    .flowOn(Dispatchers.Default)
    .collect { updateUi(it) }

StateFlow & SharedFlow

class SearchViewModel : ViewModel() {

    // StateFlow — always has value, replays last to new collectors
    private val _query = MutableStateFlow("")
    val query: StateFlow<String> = _query.asStateFlow()

    // SharedFlow — hot, no initial value, configurable replay
    private val _events = MutableSharedFlow<UiEvent>(extraBufferCapacity = 16)
    val events: SharedFlow<UiEvent> = _events.asSharedFlow()

    val results: StateFlow<List<SearchResult>> = query
        .debounce(300)
        .distinctUntilChanged()
        .filter { it.length >= 2 }
        .flatMapLatest { searchRepo.search(it) }
        .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), emptyList())

    fun onQueryChange(q: String) { _query.value = q }

    fun onItemClick(item: SearchResult) {
        viewModelScope.launch {
            _events.emit(UiEvent.Navigate(item.id))
        }
    }
}

Channel

// Channel for producer-consumer
val channel = Channel<Work>(capacity = 64)

// Producer
launch {
    workItems.forEach { channel.send(it) }
    channel.close()
}

// Consumer
launch {
    for (work in channel) {
        process(work)
    }
}

// callbackFlow — bridge callback APIs to Flow
fun locationUpdates(): Flow<Location> = callbackFlow {
    val callback = object : LocationCallback() {
        override fun onLocationResult(result: LocationResult) {
            trySend(result.lastLocation ?: return)
        }
    }
    locationClient.requestUpdates(callback)
    awaitClose { locationClient.removeUpdates(callback) }
}

Exception Handling

// CoroutineExceptionHandler — only for top-level launch
val handler = CoroutineExceptionHandler { _, exception ->
    log.error("Unhandled coroutine exception", exception)
}
scope.launch(handler) { riskyWork() }

// try/catch inside suspend functions
suspend fun safeLoad(id: Long): User? = try {
    api.getUser(id)
} catch (e: HttpException) {
    null
}

// runCatching — functional error handling
val result = runCatching { api.getUser(id) }
    .getOrElse { User.guest() }

Testing Coroutines

class UserViewModelTest {
    @get:Rule val mainDispatcherRule = MainDispatcherRule()

    @Test
    fun `search updates results`() = runTest {
        val vm = SearchViewModel(fakeRepo)
        vm.onQueryChange("kotlin")
        advanceTimeBy(400) // past debounce
        assertEquals(expectedResults, vm.results.value)
    }
}

// MainDispatcherRule
class MainDispatcherRule : TestWatcher() {
    val dispatcher = StandardTestDispatcher()
    override fun starting(d: Description?) = Dispatchers.setMain(dispatcher)
    override fun finished(d: Description?) = Dispatchers.resetMain()
}

Key Rules

• Always rethrow CancellationException — catching it silently breaks structured concurrency
• Use viewModelScope/lifecycleScope in Android; create CoroutineScope(SupervisorJob() + Dispatchers.Default) in plain Kotlin
• StateFlow for UI state; SharedFlow for one-time events; Channel for work queues
• stateIn(WhileSubscribed(5000)) prevents restarting upstream on config change
• Never do blocking I/O on Dispatchers.Main — always withContext(Dispatchers.IO)