Channels

Channels in Kotlin are a way to send data between coroutines. They provide a safe, concurrent way to pass data between different parts of your program without the need for complex synchronization. Think of channels as a pipe: one coroutine can send data into one end (the producer), and another can receive that data from the other end (the consumer).

Understanding channels

At a more technical level, channels in Kotlin are non-blocking primitives used for communication between coroutines. They are similar to BlockingQueues but with a key difference: rather than blocking a thread, their operations suspend. A coroutine is suspended if the necessary conditions aren't met (for example, sending to a full channel), freeing up the underlying thread to do other work.

To simplify, envision a channel as a data pipe connecting parts of your program that run concurrently. This pipe ensures safe data transfer, so you don't need to worry about the complexities of manual synchronization.

The Producer: Sending data to a Channel

The coroutine sending data is the producer. We use the send() function to place an element into the channel. If the channel doesn't have the capacity to accept the element immediately (which we'll explore later), the send function will suspend until it does.

When the producer has finished sending all its data, it should call close() to indicate that no more elements are coming. After a channel is closed, any further attempts to send will throw a ClosedSendChannelException, but any data already in the channel remains available for consumers to receive.

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

// ...

val channel = Channel<Int>()

// A producer sending 3 elements and then closing the channel
launch {
    for (num in 1..3) {
        channel.send(num)
    }
    channel.close() // Signal that the producer is done sending
}

The consumer: Receiving data from a Channel

The coroutine that receives data is the consumer. There are two primary ways to get data from a channel.

• Using receive():

  The receive() function retrieves a single element. If the channel is empty, receive() will suspend until an element becomes available. However, if you call receive() on a channel that is both empty and has been closed, it will throw a ClosedReceiveChannelException.

  We can receive the elements from our previous producer as follows:

  // A consumer receiving 3 elements
  launch {
      repeat(3) {
          val num = channel.receive()
          print("$num ") // 1 2 3
      }
  }
  

  Note that calling receive a fourth time would result in an exception since we have received all 3 sent elements and the channel was closed after.

• Using a for loop:

  A convenient way to consume from a channel is with a for loop. This loop will suspend when the channel is empty, waiting for the next element. When the channel is closed by the producer and all remaining elements have been received, the loop exits. Conceptually, close() is like sending a special "end" token to the channel causing iterations to stop when the token is received.

  Let's combine our producer and consumer:

  val channel = Channel<Int>()
  
  // Producer coroutine
  launch {
      for (num in 1..3) {
          channel.send(num)
      }
      channel.close() // Essential for the consumer's for loop to terminate
  }
  
  // Consumer coroutine
  launch {
      // The loop suspends for new elements and exits when the channel is closed and empty
      for (num in channel) {
          print("$num ") // 1 2 3
      }
  }
  

Producer-consumer pattern helpers

The producer-consumer pattern is so common that Kotlin provides convenient helper functions for it.

The produce coroutine builder is a handy way to create a producer. It returns a ReceiveChannel (a channel that only can be received from) and, most importantly, automatically closes the channel when its code block finishes.

On the consumer side, consumeEach is a terminal operation that processes elements from a channel. However, it has one critical difference from a for loop: when consumeEach finishes (either by processing all elements, returning early, or an exception), it cancels the channel. This is an abrupt termination, meaning any elements still in the channel's buffer are lost (we'll explore buffers in the next section). For this reason, consumeEach is best used when a single exclusive consumer has the authority to shut the entire channel.

import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.runBlocking

// This function creates a producer coroutine that automatically closes its channel
fun CoroutineScope.produceNumbers(): ReceiveChannel<Int> = produce {
    for (x in 1..5) send(x)
}

fun main(): Unit = runBlocking {
    val numbers = produceNumbers()
    run {
        numbers.consumeEach { num ->
            print("$num ")
            // If we returned early here,
            // for example: `if (num == 3) return@run`,
            // the channel would be cancelled,
            // and any remaining numbers (e.g., 4, 5) would be lost
        }
    }
}

Types of channels

So far, all the channels we've used have been Rendezvous channels. This is the default type, but Kotlin offers others based on their capacity, which affects when send() will suspend.

Rendezvous Channels:

Have no buffer capacity. When a coroutine sends data to a rendezvous channel, it suspends until another coroutine is ready to receive that data. Similarly, if a coroutine tries to receive data from a rendezvous channel, it suspends until another coroutine is ready to send data. They must "meet" for the data transfer to happen.

Consider a small elevator that can only accommodate one person at a time. If a person is in the elevator, no one else can enter until that person exits. Similarly, a Rendezvous Channel can only hold one piece of data at a time. It suspends the send operation until its data is duly received, clearing up the channel again.

val channel = Channel<Int>() // Same as Channel(0)

Buffered Channels:

Have a specified buffer capacity. If the buffer is full, a sending coroutine will suspend until space becomes available in the buffer. If the buffer is empty, a receiving coroutine will suspend until there is data in the buffer.

Consider a small storage room with a limited capacity. This room can hold only a certain quantity of items. If the room is full, you must wait until more space becomes available before you can bring in additional items. In this scenario, the storage room represents the buffered channel, and the items symbolize the data. The buffered channel suspends the sending operation when it's filled, and resumes it as space becomes available.

val channel = Channel<Int>(5) // Buffer capacity of 5

Unlimited Channels:

Can hold a theoretically infinite number of elements (limited only by system memory). When a coroutine sends data to an Unlimited Channel, it will not suspend because the channel can always accept new data. Conversely, if a coroutine tries to receive data from an empty unlimited channel, it will suspend until there is data to be received.

To illustrate this, imagine a massive warehouse capable of storing an unlimited number of items. No matter how many items you deliver, the warehouse always has room for more. In this analogy, the warehouse represents the unlimited channel, and the goods represent the data. An unlimited channel never suspends the sending operation because it can always accept additional data.

val channel = Channel<Int>(Channel.UNLIMITED)

Conflated Channels:

Retains only the most recent element sent. If a new element is dispatched to a conflated channel, it replaces the previous one. While a sending coroutine will never suspend, a receiving coroutine will suspend if there is no data to receive.

Imagine a digital billboard displaying a message. When a new message arrives, it immediately replaces the previous one. In this analogy, the billboard represents the conflated channel, and the messages symbolize the data. The conflated channel always holds the most recently sent data, discarding any previous data if it hasn't been received yet.

val channel = Channel<Int>(Channel.CONFLATED)

Conclusion

In conclusion, Kotlin's channels offer a safe and concurrent way to pass data between coroutines. They come in four types, namely: unlimited, buffered, rendezvous, and conflated, each with distinctive characteristics. The primary operations are send() and receive(), and you can close a channel when sending data is no longer needed. Understanding these concepts will help you manage concurrent data flows effectively within your Kotlin applications.