Channels, shared state and advanced concurrency

Channel and produce, select, Mutex and atomics, SharingStarted strategies, exception handling, and deterministic coroutine tests.

Channels and pipelines

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*

fun CoroutineScope.numbers(): ReceiveChannel<Int> = produce {
    for (i in 1..5) send(i)
}   // the channel closes automatically when the block completes

suspend fun main() = coroutineScope {
    val input = numbers()
    val doubled = produce {
        for (value in input) send(value * 2)
    }
    for (result in doubled) println(result)

    // a rendezvous channel: send suspends until a receiver takes the value
    val handoff = Channel<String>(Channel.RENDEZVOUS)
    launch { handoff.send("ready") }
    println(handoff.receive())

    // buffered capacity trades memory for throughput
    val queue = Channel<Int>(capacity = 64)
    queue.close()                  // close lets receivers finish normally
}

• produce is a scope-bound coroutine that owns a channel and cancels it if the scope dies.
• Channel.CONFLATED keeps only the newest value — ideal for progress updates where intermediate values do not matter.
• Iterating a channel ends when it is closed. Forgetting to close leaves consumers suspended forever.
• Channels are for hand-off between coroutines; Flow is for a cold stream that can be collected many times.

Mutex, atomics and state flows

import kotlinx.coroutines.flow.*
import kotlinx.coroutines.sync.withLock
import java.util.concurrent.atomic.AtomicLong

class Counter {
    private val mutex = Mutex()
    private var total = 0L

    suspend fun add(amount: Long) = mutex.withLock {
        total += amount
        total
    }

    // for a single value, an atomic is enough and never suspends
    private val hits = AtomicLong()
    fun hit() = hits.incrementAndGet()
}

class Ticker(private val repository: Repository) {
    private val _state = MutableStateFlow(State())
    val state: StateFlow<State> = _state.asStateFlow()

    val live: StateFlow<State> = _state
        .stateIn(
            scope = scope,
            started = SharingStarted.WhileSubscribed(5_000),
            initialValue = State()
        )

    suspend fun refresh() {
        _state.update { it.copy(loading = true) }
        val result = runCatching { repository.load() }
        _state.update { current ->
            result.fold(
                onSuccess = { current.copy(loading = false, items = it) },
                onFailure = { current.copy(loading = false, error = it.message) }
            )
        }
    }
}

Exceptions and deterministic tests

import kotlinx.coroutines.test.*

class TickerTest {
    @Test
    fun refreshExposesTheError() = runTest {
        val repository = FakeRepository(failure = IllegalStateException("offline"))
        val ticker = Ticker(repository, backgroundScope)

        ticker.refresh()
        advanceUntilIdle()

        assertEquals("offline", ticker.state.value.error)
    }
}

// a handler on the scope catches what a child failed to handle
val scope = CoroutineScope(
    SupervisorJob() + Dispatchers.Default + CoroutineExceptionHandler { _, cause ->
        System.err.println("unhandled: " + cause.message)
    }
)

FAQ

Related

Functional Kotlin: lambdas, inline functions and DSLs Coroutines and collections

Last refreshed 2026-09-18.