Kotlin Coroutines and Flow

Kotlin Coroutines provide structured concurrency for Android and backend development. Combined with Flow for reactive streams, they offer a powerful async programming model.

Suspend Functions

// Suspend functions can pause without blocking the thread
suspend fun fetchUser(id: String): User {
    return withContext(Dispatchers.IO) {
        api.getUser(id)  // Network call
    }
}

// Call from coroutine
viewModelScope.launch {
    val user = fetchUser("123") // Non-blocking suspend
    updateUI(user)
}

Coroutine Scopes and Builders

// launch: fire-and-forget
val job = scope.launch {
    delay(1000)
    println("Done after 1 second")
}

// async: returns a Deferred (like Future/Promise)
val deferredUser = scope.async {
    fetchUser("123")
}
val user = deferredUser.await()

// Concurrent execution with async
suspend fun loadDashboard(): Dashboard {
    coroutineScope {
        val user = async { fetchUser(userId) }
        val orders = async { fetchOrders(userId) }
        val recommendations = async { fetchRecommendations(userId) }
        
        // Await all concurrently
        Dashboard(
            user = user.await(),
            orders = orders.await(),
            recommendations = recommendations.await()
        )
    }
}

Dispatchers

// IO dispatcher for network/disk operations
withContext(Dispatchers.IO) {
    database.query("SELECT * FROM users")
}

// Default dispatcher for CPU-intensive work
withContext(Dispatchers.Default) {
    processLargeDataset(data)
}

// Main dispatcher for UI operations (Android)
withContext(Dispatchers.Main) {
    binding.textView.text = result
}

// Custom dispatcher with thread pool
val customDispatcher = Executors.newFixedThreadPool(4).asCoroutineDispatcher()

Structured Concurrency

// Parent scope cancels all children
class UserRepository(private val scope: CoroutineScope) {
    
    fun loadUserData(userId: String) = scope.launch {
        try {
            val user = fetchUser(userId)
            val orders = fetchOrders(userId)
            processUserData(user, orders)
        } catch (e: CancellationException) {
            // Clean up resources
            throw e // Always rethrow CancellationException
        } catch (e: Exception) {
            handleError(e)
        }
    }
}

// SupervisorScope: child failures don't cancel siblings
supervisorScope {
    val result1 = async { riskyOperation1() }
    val result2 = async { riskyOperation2() }
    
    // Even if result1 fails, result2 still runs
    val r1 = try { result1.await() } catch (e: Exception) { null }
    val r2 = try { result2.await() } catch (e: Exception) { null }
}

Flow for Reactive Streams

// Cold flow: executes on collection
fun getNumbers(): Flow<Int> = flow {
    for (i in 1..10) {
        delay(100)
        emit(i)
    }
}

// Collect the flow
viewModelScope.launch {
    getNumbers()
        .filter { it % 2 == 0 }
        .map { it * it }
        .collect { value ->
            println("Received: $value")
        }
}

// Flow from database (Room)
@Dao
interface UserDao {
    @Query("SELECT * FROM users WHERE id = :id")
    fun observeUser(id: String): Flow<User>  // Emits on every DB change
}

// Collect in ViewModel
viewModelScope.launch {
    userDao.observeUser(userId)
        .catch { e -> emit(User.empty()) } // Handle errors
        .collect { user ->
            _uiState.value = UiState.Success(user)
        }
}

StateFlow and SharedFlow

// StateFlow: holds current state, replays to new collectors
class UserViewModel : ViewModel() {
    private val _uiState = MutableStateFlow<UiState>(UiState.Loading)
    val uiState: StateFlow<UiState> = _uiState.asStateFlow()
    
    init {
        viewModelScope.launch {
            try {
                val user = fetchUser(userId)
                _uiState.value = UiState.Success(user)
            } catch (e: Exception) {
                _uiState.value = UiState.Error(e.message)
            }
        }
    }
}

// Collect in Activity
lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        viewModel.uiState.collect { state ->
            when (state) {
                is UiState.Loading -> showLoading()
                is UiState.Success -> showUser(state.user)
                is UiState.Error -> showError(state.message)
            }
        }
    }
}

// SharedFlow: for events (no replay by default)
class EventBus {
    private val _events = MutableSharedFlow<AppEvent>()
    val events = _events.asSharedFlow()
    
    suspend fun publish(event: AppEvent) {
        _events.emit(event)
    }
}

Channels

// Channel: for communication between coroutines
val channel = Channel<Int>(capacity = Channel.BUFFERED)

// Producer
launch {
    for (i in 1..5) {
        channel.send(i)
    }
    channel.close()
}

// Consumer
launch {
    for (value in channel) {
        println("Received: $value")
    }
}

// Producer coroutine builder
fun produceNumbers() = produce<Int> {
    for (i in 1..5) {
        send(i)
    }
}

Error Handling

// CoroutineExceptionHandler for uncaught exceptions
val handler = CoroutineExceptionHandler { _, exception ->
    println("Caught exception: ${exception.message}")
    // Log to crash reporting
}

val scope = CoroutineScope(SupervisorJob() + Dispatchers.Main + handler)

scope.launch {
    throw RuntimeException("Something went wrong")
    // Caught by handler, scope continues
}

// Flow error handling
userFlow
    .retry(3) { cause -> cause is NetworkException }
    .catch { e -> emit(defaultValue) }
    .onEach { value -> processValue(value) }
    .launchIn(viewModelScope)

Ktor: Coroutines in Backend

// Ktor server with coroutines
fun Application.configureRouting() {
    routing {
        get("/users/{id}") {
            val id = call.parameters["id"]!!
            
            // Suspend function in route handler
            val user = userRepository.findById(id)
                ?: return@get call.respond(HttpStatusCode.NotFound)
            
            call.respond(user)
        }
        
        // WebSocket with Flow
        webSocket("/stream") {
            userRepository.observeAll()
                .collect { user ->
                    send(Json.encodeToString(user))
                }
        }
    }
}

// Parallel database queries
suspend fun getUserDashboard(userId: String): Dashboard = coroutineScope {
    val user = async { userRepository.findById(userId) }
    val orders = async { orderRepository.findByUserId(userId) }
    val stats = async { analyticsRepository.getUserStats(userId) }
    
    Dashboard(user.await()!!, orders.await(), stats.await())
}

Testing Coroutines

@Test
fun testFetchUser() = runTest {
    // runTest replaces runBlocking for tests
    val mockApi = mockk<UserApi>()
    coEvery { mockApi.getUser("123") } returns User(id = "123", name = "Alice")
    
    val repository = UserRepository(mockApi)
    val user = repository.getUser("123")
    
    assertEquals("Alice", user.name)
}

@Test
fun testFlow() = runTest {
    val flow = flowOf(1, 2, 3)
    val results = flow.toList()
    assertEquals(listOf(1, 2, 3), results)
}

Summary

Kotlin Coroutines and Flow provide:

• Suspend functions: Non-blocking async code that reads like synchronous
• Structured concurrency: Parent cancels children, preventing leaks
• Dispatchers: Route work to appropriate thread pools
• Flow: Reactive streams with rich operators
• StateFlow/SharedFlow: Hot flows for UI state and events
• Channels: Safe communication between coroutines

The combination makes async Android and backend code maintainable and testable.