Kotlin Channels Split Send and Receive Into Separate Interfaces — Here’s the Iterator That Powers the Loop

1. Core interface division of labour

SendChannel: Acts as the sending end of a channel, exposing only data-writing capability. Its core method is the suspending function send(element: E), which submits data to the channel. When the buffer is full, the current coroutine suspends automatically instead of blocking a thread; it also supports trySend for non-blocking send attempts and operations such as close() to shut the channel.

ReceiveChannel: Acts as the receiving end of a channel, exposing only data-reading capability. Its core method is the suspending function receive(): E, which takes data from the channel. When the channel is empty, the current coroutine suspends automatically and waits for new data to arrive; it also supports tryReceive for non-blocking receive attempts, and you can directly use a for-in loop to iterate over all elements in the channel until the channel is closed.

2. Usage example

fun main() = runBlocking {
    // Create a default unbuffered channel, holding both the send and receive ends
    val channel = Channel<Int>()
    // Extract the SendChannel side, usable only for sending data
    val sender: SendChannel<Int> = channel
    // Extract the ReceiveChannel side, usable only for receiving data
    val receiver: ReceiveChannel<Int> = channel

    // Producer coroutine: sends data via SendChannel
    launch {
        repeat(5) {
            sender.send(it)
            println("Sent: $it")
        }
        sender.close() // Close the channel after sending is complete
    }

    // Consumer coroutine: receives data via ReceiveChannel
    launch {
        // The for loop automatically iterates; the loop ends automatically after the channel closes
        for (element in receiver) {
            println("Received: $element")
        }
        println("Communication ended")
    }

    delay(1000)
}

3. Compiler-generated iteration logic

The logic actually generated by the compiler is equivalent to:

val iterator = receiver.iterator() // Obtain the iterator
while (iterator.hasNext()) {        // Check whether there is a next element
    val element = iterator.next()   // Get the next element
    println(element)
}

4. Channel implementation classes

public fun <E> Channel(
    capacity: Int = RENDEZVOUS,
    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND,
    onUndeliveredElement: ((E) -> Unit)? = null
): Channel<E> {
    // The when expression here selects the concrete implementation class
    return when (capacity) {
        RENDEZVOUS -> RendezvousChannel(onUndeliveredElement) // Unbuffered, capacity 0
        CONFLATED -> ConflatedChannel(onUndeliveredElement)   // Conflated, capacity -1
        UNLIMITED -> LinkedListChannel(onUndeliveredElement)  // Unlimited buffer, uses a linked list
        else -> ArrayChannel(capacity, onBufferOverflow, onUndeliveredElement) // Bounded buffer, uses an array
    }
}

The above implementation classes all inherit from a common base class, AbstractChannel.

// AbstractChannel.kt simplified source logic
abstract class AbstractChannel<E>(
    private val onUndeliveredElement: ((E) -> Unit)?
) : Channel<E>, SendChannel<E>, ReceiveChannel<E> {

    // 1. Provides the iterator entry point
    public final override fun iterator(): ChannelIterator<E> = Itr(this)

    // ... other send/receive logic
}

5. Iterator implementation principle

private class Itr<E>(@JvmField val channel: AbstractChannel<E>) : ChannelIterator<E> {
    var result: Any? = POLL_FAILED // E | POLL_FAILED | Closed

    override suspend fun hasNext(): Boolean {
        // check for repeated hasNext
        if (result !== POLL_FAILED) return hasNextResult(result)
        // fast path -- try poll non-blocking
        result = channel.pollInternal()
        if (result !== POLL_FAILED) return hasNextResult(result)
        // slow-path does suspend
        return hasNextSuspend()
    }

    private fun hasNextResult(result: Any?): Boolean {
        if (result is Closed<*>) {
            if (result.closeCause != null) throw recoverStackTrace(result.receiveException)
            return false
        }
        return true
    }

    private suspend fun hasNextSuspend(): Boolean = suspendCancellableCoroutineReusable sc@ { cont ->
        val receive = ReceiveHasNext(this, cont)
        while (true) {
            if (channel.enqueueReceive(receive)) {
                channel.removeReceiveOnCancel(cont, receive)
                return@sc
            }
            // hm... something is not right. try to poll
            val result = channel.pollInternal()
            this.result = result
            if (result is Closed<*>) {
                if (result.closeCause == null)
                    cont.resume(false)
                else
                    cont.resumeWithException(result.receiveException)
                return@sc
            }
            if (result !== POLL_FAILED) {
                @Suppress("UNCHECKED_CAST")
                cont.resume(true, channel.onUndeliveredElement?.bindCancellationFun(result as E, cont.context))
                return@sc
            }
        }
    }

    @Suppress("UNCHECKED_CAST")
    override fun next(): E {
        val result = this.result
        if (result is Closed<*>) throw recoverStackTrace(result.receiveException)
        if (result !== POLL_FAILED) {
            this.result = POLL_FAILED
            return result as E
        }

        throw IllegalStateException("'hasNext' should be called prior to 'next' invocation")
    }
}

Inside pollInternal, this.result is assigned the following result:

val result = channel.pollInternal()
this.result = result

6. The pollInternal function

protected open fun pollInternal(): Any? {
    while (true) {
        val send = takeFirstSendOrPeekClosed() ?: return POLL_FAILED
        val token = send.tryResumeSend(null)
        if (token != null) {
            assert { token === RESUME_TOKEN }
            send.completeResumeSend()
            return send.pollResult
        }
        // too late, already cancelled, but we removed it from the queue and need to notify on undelivered element
        send.undeliveredElement()
    }
}

protected fun takeFirstSendOrPeekClosed(): Send? =
    queue.removeFirstIfIsInstanceOfOrPeekIf<Send> { it is Closed<*> }

7. Internal queue mechanism

The design of Channel is inspired by Java's BlockingQueue, but it is purpose-built for non-blocking coroutine suspension.

Producer (SendChannel): When send(element) is called, the data element is actually wrapped into a node and placed at the tail of this internal queue. If the queue is full (for a bounded Channel), the sending coroutine is suspended until space becomes available.

Consumer (ReceiveChannel): When receive() is called, a data element is actually taken from the head of this internal queue. If the queue is empty, the receiving coroutine is suspended until new data enters the queue or the channel is closed.