Kotlin Channels

A channel is a communication pipe between coroutines. One coroutine sends data into the channel, and another coroutine receives data from it. Channels allow coroutines to exchange data safely without shared mutable state.

Channel as a Pipe

Sender Coroutine                   Receiver Coroutine
      │                                    │
      ▼                                    ▼
  channel.send(1)  →  [ 1 | 2 | 3 ]  →  channel.receive()
  channel.send(2)       (buffer)
  channel.send(3)

Basic Channel

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

fun main() = runBlocking {
    val channel = Channel()

    // Producer coroutine
    launch {
        for (i in 1..5) {
            println("Sending $i")
            channel.send(i)
        }
        channel.close()   // signals no more data
    }

    // Consumer coroutine
    for (value in channel) {
        println("Received $value")
    }
    println("Done")
}

Output:

Sending 1
Received 1
Sending 2
Received 2
Sending 3
Received 3
Sending 4
Received 4
Sending 5
Received 5
Done

produce Builder

The produce builder creates a producer coroutine and returns a ReceiveChannel directly:

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

fun CoroutineScope.generateNumbers(): ReceiveChannel = produce {
    for (i in 1..5) {
        send(i * i)   // send squares
    }
}

fun main() = runBlocking {
    val squares = generateNumbers()
    for (sq in squares) {
        println(sq)
    }
}

// Output: 1 4 9 16 25

Channel Types

Type                     │ Buffer     │ Behavior
─────────────────────────┼────────────┼────────────────────────────
Channel()                │ none       │ sender suspends until receiver is ready
Channel(capacity = 10)   │ 10 items   │ sender suspends only when buffer is full
Channel(UNLIMITED)       │ unlimited  │ sender never suspends
Channel(CONFLATED)       │ 1 item     │ new send overwrites old value

val buffered = Channel(capacity = 3)

// Can send 3 items without waiting for receiver
launch {
    buffered.send("A")
    buffered.send("B")
    buffered.send("C")
    // buffered.send("D")   ← suspends here — buffer full
}

Pipeline Pattern

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

fun CoroutineScope.numbers(): ReceiveChannel = produce {
    for (i in 1..10) send(i)
}

fun CoroutineScope.square(input: ReceiveChannel): ReceiveChannel = produce {
    for (n in input) send(n * n)
}

fun CoroutineScope.filter(input: ReceiveChannel): ReceiveChannel = produce {
    for (n in input) if (n > 20) send(n)
}

fun main() = runBlocking {
    val nums     = numbers()
    val squared  = square(nums)
    val filtered = filter(squared)

    for (n in filtered) print("$n ")
    // Output: 25 36 49 64 81 100
}

Fan-Out: Multiple Receivers

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

fun main() = runBlocking {
    val channel = Channel()

    // 1 producer
    launch {
        for (i in 1..6) { delay(100); channel.send(i) }
        channel.close()
    }

    // 2 consumers competing for items
    repeat(2) { id ->
        launch {
            for (item in channel) {
                println("Worker $id got item $item")
                delay(200)
            }
        }
    }
    delay(2000)
}

Practical Example: Order Processing Pipeline

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

data class Order(val id: Int, val item: String)
data class PackedOrder(val order: Order, val box: String)

fun main() = runBlocking {
    val orders = produce {
        listOf(Order(1,"Laptop"), Order(2,"Mouse"), Order(3,"Keyboard")).forEach { send(it) }
    }

    val packed = produce {
        for (order in orders) {
            delay(100)
            send(PackedOrder(order, "Box-${order.id}"))
        }
    }

    for (p in packed) {
        println("Shipped: ${p.order.item} in ${p.box}")
    }
}

Output:

Shipped: Laptop in Box-1
Shipped: Mouse in Box-2
Shipped: Keyboard in Box-3