11. Channels

Written by Nishant Srivastava

From the previous chapters, you already learned how to deal with sending a request for receiving a single value. This approach works perfectly when you just need to get a value once and show it to the user, e.g., fetching a user profile or downloading an image. In this chapter, you will learn how to send and receive streams of values.

Streams are convenient when you need to continuously get updates of data or handle a potentially infinite sequence of items. Kotlin isn’t the first one to offer a solution to these problems. Observable from ReactiveX and Queue from Java solve them as well. How Channels compare with Observable and Queue, as well as their benefits and disadvantages, will be covered further in this book.

Note: A stream is a source or repository of data that can be read or written only sequentially while a thread is a unit of execution, lighter in weight than a process, generally expected to share memory and other resources with other threads executed concurrently.

Getting started with channels

Channels are conceptually similar to reactive streams. It is a simple abstraction that you can use to transfer a stream of values between coroutines. Consider a source that sends content to a destination that receives it; i.e., elements are sent into the channel by producer coroutines and are received by consumer coroutines. Essentially, channels are like blocking queues that send and operate on data asynchronously.

A fundamental property — and an important concept to understand — of a channel is its capacity, which defines the maximum number of elements that a channel can contain in a buffer. Suppose you have a channel with capacity N. A producer can send values into the channel but, when the channel reaches its N capacity, the producer suspends until a consumer starts to read data from the same channel. You can think of the capacity like the size of the buffer for a specific channel; it’s a way to optimize performances in the case producing and consuming are operations, which take different amounts of time.

You can change the default capacity of a channel by passing it as an argument to its factory method. Take a look at the following method signature:

public fun <E> Channel(capacity: Int = RENDEZVOUS): Channel<E>

You will notice that the default capacity is set to RENDEZVOUS, which corresponds to 0 as per the source code:

public const val RENDEZVOUS = 0

What does it mean in practice? It means that the producer channel won’t produce anything until there is a consumer channel that needs data; essentially, there is no buffer.

An element is transferred from producer to consumer only when the producer’s send and consumer’s receive invocations meet in time (rendezvous). Because of this, the send function suspends until another coroutine invokes receive and receive suspends until another coroutine invokes send. This is the reason for the RENDEZVOUS name.

Note: The same happens in Java with the SynchronousQueue class.

Creating a channel is pretty straightforward. Write the following:

val kotlinChannel = Channel<Int>()

Consuming its values can be done via the usual for loop:

for (x in kotlinChannel){
  println(x)
}

Channels implement the SendChannel and ReceiveChannel interfaces.

public interface SendChannel<in E> {
  @ExperimentalCoroutinesApi
  public val isClosedForSend: Boolean

  public suspend fun send(element: E)
  public fun offer(element: E)
  public fun close(cause: Throwable? = null): Boolean
  ...
}

public interface ReceiveChannel<out E> {
  @ExperimentalCoroutinesApi
  public val isClosedForReceive: Boolean

  public suspend fun receive(): E
  public fun cancel(): Unit
  ...
}

Notice that SendChannel exposes the operation close, which is used — surprise — for closing the channel. As soon as the sender calls close() on the channel, the value of isClosedForSend becomes true.

Note: close() is an idempotent operation; repeated invocations of this function have no effect and return false.

You can’t send any message into a closed channel. Closing a channel conceptually works by sending a special close token over it. You close a channel when you have a finite sequence of elements to be processed by consumers. You must then signal to the consumers that this sequence is over. The iteration stops as soon as this close token is received, so there is a guarantee that all previously sent elements before the close are received. You don’t have to close a channel otherwise.

On the other hand, ReceiveChannel exposes the cancel operation, which cancels the reception of remaining elements from the channel. Once finished, this function closes the channel and removes all messages in the buffer, if any. After cancel() completes, isClosedForReceive starts returning true. If the producer has already closed the channel invoking the close() function, then isClosedForReceive returns true only after all previously sent elements are received.

The isClosedForReceive property can be used along with channel.receive() to iterate and get items from a channel one at a time:

while (!kotlinChannel.isClosedForReceive) {
    val value = kotlinChannel.receive()
    println(value)
}

Channels are not tied to any native resource and they don’t have to be closed to release their memory; hence, simply dropping all the references to a channel is fine. When the garbage collector runs, it will clean out those references.

Other important methods are send and receive. You can send items to the channel with the method send(element: E) and receive from it with receive():E.

This is typical usage for a channel:

fun main() {

  // 1
  val fruitArray = arrayOf("Apple", "Banana", "Pear", "Grapes",
      "Strawberry")

  // 2
  val kotlinChannel = Channel<String>()

  runBlocking {

    // 3
    GlobalScope.launch {
      for (fruit in fruitArray) {
        // 4
        kotlinChannel.send(fruit)

        // 5
        if (fruit == "Pear") {
          // 6
          kotlinChannel.close()
        }
      }
    }

    // 7 
    for (fruit in kotlinChannel) {
      println(fruit)
    }

    // 8
    println("Done!")
  }
}

Output:

Apple
Banana
Pear
Done!

Breaking down each part of the above code snippet, which you can find the executable version of the above snippet of code in the starter project in the file called ChannelsIntro.kt:

1. An array of string items.
2. Create a channel with default — i.e., 0 capacity.
3. Set up the producer.
4. Send data in the channel.
5. Conditional check, if the current item is equal to value Pear.
6. Signal the closure of the channel via calling close() on the channel.
7. Set up the consumer that is printing the received values using for loop (until the channel is closed).
8. Print the final Done status.

In the previous example, you create a Channel of String objects. Then, into the body of the launch coroutine builder, you iterate over an Array<String> and put each element into the channel using the send function. While iterating, you check if the current value equals to Pear, in which case you close the channel invoking the close method. This is an example of a condition for the closing of the channel.

On the receiving side, you use a normal iteration with a for cycle in order to consume all the elements available in the channel. The for cycle is smart enough to understand when the channel is closed because it uses the underlying Iterator.

The for loop solution is excellent because it allows you to use channels in the normal pattern that you’d use for iterating over a normal collection. If you want more control over what you’re doing, you can consume the channel using code like this:

while (!kotlinChannel.isClosedForReceive) {
    val value = kotlinChannel.receive()
    println(value)
}

However, there is yet another way to iterate over the channel values, via using repeat() Kotlin construct:

// Another way to iterate over the channel values
// You use channel.receive() to
// get the messages one by one
repeat(3){
  val fruit = kotlinChannel.receive()
  println(fruit)
}

Here, you explicitly use the receive method but you have to know exactly how many elements you’re getting from the channel, which is not always possible. If you try to put 4 instead of 3 as argument of the repeat function, you’ll have a ClosedReceiveChannelException exception like this:

Apple
Banana
Pear
Exception in thread "main" kotlinx.coroutines.channels.ClosedReceiveChannelException: Channel was closed
	at kotlinx.coroutines.channels.Closed.getReceiveException(AbstractChannel.kt:1070)

It’s interesting to note that the exception is not thrown on the receive function but on the close one. This happens because the close function is a suspend function, which actually completes only when the receiver consumes all the items in the channel. If the receiver requests more items that the one available, the producer tries to provide some new data. But, in this, case the channel is closed and this is not possible. This is the reason for the ClosedReceiveChannelException. In the case that you put a value smaller than the number of available objects, on the other hand, you’re going to miss some data.

Understanding closed channels

In order to understand the state of the channel, you can use two handy properties: isClosedForReceive and isClosedForSend.

When you close a channel from the sender, like in the previous example, you implicitly put its isClosedForSend property to true, which means that you can’t send new data. It’s important to understand that this doesn’t imply that isClosedForReceive is also true. This is because there should still be some data in the channel. When the receiver consumes all the data, then the isClosedForReceive is also set to true. Because of this, you can consume the data using the following code:

while (!kotlinChannel.isClosedForReceive) {
  val fruit = kotlinChannel.receive()
  println(fruit)
}

Here, you’re receiving data until the isClosedForReceive property is true. Sadly, if you run this example, you might still get an exception on the close function. Why? Unfortunately, channel APIs are unstable and, in this case, there’s a race condition on the update of the isClosedForReceive property. In order to fix this, you could add a simple delay, which gives time to the channel for the update, but this is not deterministic and sometimes it won’t work:

while (!kotlinChannel.isClosedForReceive) {
  val fruit = kotlinChannel.receive()
  delay(10)
  println(fruit)
}

A more reliable way to produce the same results is to use the produce coroutine builder on the producer side and an extension function consumeEach, that replaces a for loop on the consumer side. Basically, it makes everything run smoothly, bringing order and a clean approach to consuming items in the channel:

@ObsoleteCoroutinesApi
@ExperimentalCoroutinesApi
fun main() {

  val fruitArray = arrayOf("Apple", "Banana", "Pear", "Grapes", "Strawberry")

  fun produceFruits() = GlobalScope.produce<String> {
    for (fruit in fruitArray) {
      send(fruit)

      // Conditional close
      if (fruit == "Pear") {
        // Signal that closure of channel
        close()
      }
    }
  }

  runBlocking {
    val fruits = produceFruits()
    fruits.consumeEach { println(it) }
    println("Done!")
  }
}

The end result is the same, but the approach to producing and consuming is much cleaner. By way of reminder, you can find the executable version of the above snippet of code in the starter project in the file called ChannelIntroWithProduce.kt.

Note: Channels are still under development and considered experimental. You will need to annotate the main method with @ExperimentalCoroutinesApi annotation. Behavior of producers may change in the future. You will also notice that we added another annotation, @ObsoleteCoroutinesApi because consumeEach API will become obsolete in future updates with the introduction of lazy asynchronous streams. See issue #254 here: https://github.com/Kotlin/kotlinx.coroutines/issues/254.

Pipelines

With channels, you always have a producer and a consumer. Sometimes, a consumer receives the data from a channel, applies some transformations and becomes the producer of a new channel. When a consumer of a channel becomes the producer of another channel, you create a Pipelines. The source channel might be infinite and the pipeline might contain different steps.

Check out an example in which you generate a list of items that are red fruits. You will make use of multiple channels connected as a pipeline to get the final result:

data class Fruit(override val name: String, override val color: String) : Item
data class Vegetable(override val name: String, override val color: String) : Item

@ExperimentalCoroutinesApi
fun main() {


  // ------------ Helper Methods ------------
  fun isFruit(item: Item): Boolean = item is Fruit

  fun isRed(item: Item): Boolean = (item.color == "Red")


  // ------------ Pipeline ------------
  // 1
  fun produceItems() = GlobalScope.produce {
    val itemsArray = ArrayList<Item>()
    itemsArray.add(Fruit("Apple", "Red"))
    itemsArray.add(Vegetable("Zucchini", "Green"))
    itemsArray.add(Fruit("Grapes", "Green"))
    itemsArray.add(Vegetable("Radishes", "Red"))
    itemsArray.add(Fruit("Banana", "Yellow"))
    itemsArray.add(Fruit("Cherries", "Red"))
    itemsArray.add(Vegetable("Broccoli ", "Green"))
    itemsArray.add(Fruit("Strawberry", "Red"))

    // Send each item in the channel
    itemsArray.forEach {
      send(it)
    }
  }

  // 2
  fun isFruit(items: ReceiveChannel<Item>) = GlobalScope.produce {
    for (item in items) {
      // Send each item in the channel only if it is a fruit
      if (isFruit(item)) {
        send(item)
      }
    }
  }

  // 3
  fun isRed(items: ReceiveChannel<Item>) = GlobalScope.produce {
    for (item in items) {
      // Send each item in the channel only if it is red in color
      if (isRed(item)) {
        send(item)
      }
    }
  }

  runBlocking {
    // 4
    val itemsChannel = produceItems()
    // 5
    val fruitsChannel = isFruit(itemsChannel)
    // 6
    val redChannel = isRed(fruitsChannel)

    // 7
    for (item in redChannel) {
      print("${item.name}, ")
    }

    // 8
    redChannel.cancel()
    fruitsChannel.cancel()
    itemsChannel.cancel()

    // 9
    println("Done!")
  }
}

Take note of the Item interface being used, which you can find in the starter project in Items.kt file with the below definition:

interface Item {
  val name: String
  val color: String
}

Now, run the above example:

Output:

Apple, Cherries, Strawberry, Done!

Breaking down the above code snippet:

1. Channel 1: Produces a finite number of items that are either a fruit or vegetable on the internal channel for the produce coroutine builder.
2. Channel 2: Produces only the items that are fruit on the internal channel for the produce coroutine builder.
3. Channel 3: Produces only the items that red in color on the internal channel for the produce coroutine builder.
4. Wire up and set up the pipeline by initializing the itemsChannel via the produceItems() method, which produces a stream of items.
5. itemsChannel is then passed to the fruitsChannel via the isFruit(itemsChannel) method, which feeds the stream of items into the fruitsChannel. This channel then checks if the item is a fruit or not. If it is, then it sends the item in its own channel.
6. fruitsChannel is then passed to the redChannel via the isRed(fruitsChannel) method, which feeds the stream of fruit items into the redChannel. This channel then checks if the fruit item is red colored or not. If it is, then it sends the item in its own channel.
7. Using a for loop, print all the items that are fruits and of red color from the redChannel.
8. It is recommended to cancel all the coroutines for good measure.
9. Finally, print the final "Done" status to the console.

You can find the executable version of the above snippet of code in the starter project in the file called PipelineExample.kt.

As you would have noticed, there are three channels being utilized here, which are connected one after the other to get the final result, representing a Pipeline:

• The produceItems function creates a channel with objects that are either a fruit or a vegetable.
• The isFruit function iterates over the ReceiveChannel passed as a parameter, creating a new channel that only produces fruit. It’s important to note that this function consumes all the items in the input channel while ignoring the ones that are not a fruit, which is basically lost.
• The isRed function does something similar with a different predicate, creating a channel, which produces the only items in input that are red.
• The main() function creates the pipeline setting itemsChannel as the input for the isFruit function and the output of this as the input for the isRed function.

Notice also that you do not close the channels, but only directly canceled the coroutines. That is because you are signaling the end of transmission as a whole.

Fan out

In the previous example, you created a pipeline as a sequence of channels, each one with a single producer and a single consumer. Coroutines were consuming the data from a channel and testing if that data satisfied certain conditions. In the case of success, the items were put into the new channel; otherwise, they were discarded.

Sometimes, the scenario is a little bit more complicated and you’d like to send each item to a different coroutine depending on a specific condition. You need some kind of demultiplexer, which, in the context of channels, is a use case called Fan-out.

The challenge here is that you can’t test the item if you don’t consume it first. A possible solution would be to consume the item, test it and put it again into the original channel if it’s not compliant with your coroutine. Unfortunately, this is not a doable approach because of the laziness of the channel.

In this case, a better solution consists in the creation of a coroutine with demultiplexer feature, which sends an item to a channel or another given a predicate. In the following example, we use an approach inspired by the Chain of Responsibility design pattern.

You can find the code for this example in the starter project in the file called FanOut.kt.

In this case, you need some initial abstractions:

typealias Predicate<E> = (E) -> Boolean

typealias Rule<E> = Pair<Channel<E>, Predicate<E>>

A Predicate is any function with a parameter of generic type E which can return either true or false. A Rule is a name for a Pair of a Channel and a Predicate. The idea is to allow a coroutine to send a value to a specific channel only if its predicate returns true if evaluated for the value itself.

You can encapsulate the demultiplexing logic into a class like below:

class Demultiplexer<E>(vararg val rules: Rule<E>) {

  suspend fun consume(recv: ReceiveChannel<E>) {
    for (item in recv) {
      // 1
      for (rule in rules) {
        // 2            
        if (rule.second(item)) {
          // 3
          rule.first.send(item)
        }
      }
    }
    // 4
    closeAll()
  }

  // Closes all the demultiplexed channels
  private fun closeAll() {
    rules.forEach { it.first.close() }
  }
}

1. You iterate over all the values of the channel to consume.
2. Iterate over the possible destination channels into the rules passed as varargs parameters.
3. If the predicate for the current value evaluates to true, you invoke the send function on the corresponding channel. If the predicate is false, the value is skipped.
4. When you exit the for loop it means that the source channel is closed, and so you close all the destination channels. A cancelation like the one in the previous example could be another option.

Finally, you can refer to the following example, which generates a list of items that are either a fruit or a vegetable, and it dispatches them to two different channels depending on their type:

@ExperimentalCoroutinesApi
fun main() {

  data class Fruit(override val name: String, override val color: String) : Item
  data class Vegetable(override val name: String, override val color: String) : Item

  // ------------ Helper Methods ------------
  fun isFruit(item: Item) = item is Fruit

  fun isVegetable(item: Item) = item is Vegetable

  // 1
  fun produceItems(): ArrayList<Item> {
    val itemsArray = ArrayList<Item>()
    itemsArray.add(Fruit("Apple", "Red"))
    itemsArray.add(Vegetable("Zucchini", "Green"))
    itemsArray.add(Fruit("Grapes", "Green"))
    itemsArray.add(Vegetable("Radishes", "Red"))
    itemsArray.add(Fruit("Banana", "Yellow"))
    itemsArray.add(Fruit("Cherries", "Red"))
    itemsArray.add(Vegetable("Broccoli", "Green"))
    itemsArray.add(Fruit("Strawberry", "Red"))
    itemsArray.add(Vegetable("Red bell pepper", "Red"))
    return itemsArray
  }

  runBlocking {

    // 2
    val kotlinChannel = Channel<Item>()

    // 3
    val fruitsChannel = Channel<Item>()
    val vegetablesChannel = Channel<Item>()

    // 4
    launch {
      produceItems().forEach {
        kotlinChannel.send(it)
      }
      // 5
      kotlinChannel.close()
    }

    // 6
    val typeDemultiplexer = Demultiplexer(
        fruitsChannel to { item: Item -> isFruit(item) },
        vegetablesChannel to { item: Item -> isVegetable(item) }
    )

    // 7
    launch {
      typeDemultiplexer.consume(kotlinChannel)
    }

    // 8
    launch {
      for (item in fruitsChannel) {
        // Consume fruitsChannel
        println("${item.name} is a fruit")
      }
    }

    // 9
    launch {
      for (item in vegetablesChannel) {
        // Consume vegetablesChannel
        println("${item.name}  is a vegetable")
      }
    }
  }
}

Here, in the above code snippet, you:

1. Create a produceItems function for producing a finite number of items, which are either a fruit or vegetable.

2. Create a channel for Item.

3. Create the fruitsChannel channel for items that are fruits and a vegetablesChannel channel for items that are vegetables.

4. Launch a coroutine for sending all items generated by the produceItems function.

5. When completed, close the channel.

6. Create a Demultiplexer instance, which maps items that are fruit to the fruitsChannel and items that are vegetables to the vegetablesChannel.

7. The Demultiplexer has a consume method, which is suspending, and it needs a coroutine that you launch.

8. You consume the fruitsChannel channel, printing its values.

9. You consume the vegetablesChannel channel, printing its values.

As you can see, the output will be:

Apple is a fruit
Zucchini  is a vegetable
Grapes is a fruit
Radishes  is a vegetable
Banana is a fruit
Cherries is a fruit

Broccoli  is a vegetable
Strawberry is a fruit
Red bell pepper  is a vegetable

The order of the item type evaluation is now different. This is obvious because now each channel can be consumed independently.

Fan in

In the previous example, you created a coroutine that was able to demultiplex the items into different channels based on certain criteria. That was a way to simulate the case in which you have one producer and many consumers.

A different case happens when you have multiple producers and one consumer: This is called Fan-in, and it’s a simpler situation compared to the previous.

As an example, you can implement the following code:

@ExperimentalCoroutinesApi
fun main() {

  data class Fruit(override val name: String, override val color: String) : Item
  data class Vegetable(override val name: String, override val color: String) : Item

  // ------------ Helper Methods ------------
  fun isFruit(item: Item) = item is Fruit

  fun isVegetable(item: Item) = item is Vegetable


  // 1
  fun produceItems(): ArrayList<Item> {
    val itemsArray = ArrayList<Item>()
    itemsArray.add(Fruit("Apple", "Red"))
    itemsArray.add(Vegetable("Zucchini", "Green"))
    itemsArray.add(Fruit("Grapes", "Green"))
    itemsArray.add(Vegetable("Radishes", "Red"))
    itemsArray.add(Fruit("Banana", "Yellow"))
    itemsArray.add(Fruit("Cherries", "Red"))
    itemsArray.add(Vegetable("Broccoli", "Green"))
    itemsArray.add(Fruit("Strawberry", "Red"))
    itemsArray.add(Vegetable("Red bell pepper", "Red"))
    return itemsArray
  }


  runBlocking {

    // 2
    val destinationChannel = Channel<Item>()

    // 3
    val fruitsChannel = Channel<Item>()
    val vegetablesChannel = Channel<Item>()

    // 4
    launch {
      produceItems().forEach {
        if (isFruit(it)) {
          fruitsChannel.send(it)
        }
      }
    }

    // 5
    launch {
      produceItems().forEach {
        if (isVegetable(it)) {
          vegetablesChannel.send(it)
        }
      }
    }


    // 6
    launch {
      for (item in fruitsChannel) {
        destinationChannel.send(item)
      }
    }

    // 7
    launch {
      for (item in vegetablesChannel) {
        destinationChannel.send(item)
      }
    }

    // 8
    destinationChannel.consumeEach {
      if (isFruit(it)) {
        println("${it.name} is a fruit")
      } else if (isVegetable(it)) {
        println("${it.name} is a vegetable")
      }
    }


    // 9
    coroutineContext.cancelChildren()
  }
}

You can find the code for this example in the starter project in the file called FanIn.kt.

In the above:

1. Create a produceItems function for producing a finite number of items, which are either a fruit or vegetable.
2. Initialize the destination channel.
3. Create the fruitsChannel channel for items that are fruits and vegetablesChannel channel for items that are vegetables.
4. Launch the coroutine that inserts the items that are fruits into the fruitsChannel channel.
5. Launch the coroutine that inserts the items that are vegetables into the vegetablesChannel channel.
6. Here is where the multiplexing is happening for items that are fruits that are sent into the destination channel.
7. Here is where the multiplexing is happening for items that are vegetables that are sent into the destination channel.
8. You consume the destination channel and print a label depending on the type of item.
9. You cancel all the coroutines when there’s nothing more to consume.

Now, the output will be something like this:

Apple is a fruit
Zucchini is a vegetable
Grapes is a fruit
Banana is a fruit
Radishes is a vegetable
Cherries is a fruit
Broccoli is a vegetable
Strawberry is a fruit
Red bell pepper is a vegetable

Buffered channel

As you might have noticed above, the channel examples demonstrated previously used a default value for the capacity, called RENDEZVOUS. These kinds of channels are called unbuffered channels because the producer produces only if there’s a consumer ready to consume.

However, this behavior can be overcome easily by specifying the buffer capacity of the channel as a parameter in the factory method. In this way, your channel won’t suspend on a send operation when there is a free space in the buffer. You can create buffered channels that will allow senders to send multiple elements before suspending:

// Channel of capacity 2
val kotlinBufferedChannel = Channel<String>(2) 

Check out a working example:

fun main() {

  val fruitArray = arrayOf("Apple", "Banana", "Pear", "Grapes", "Strawberry")

  val kotlinBufferedChannel = Channel<String>(2)

  runBlocking {
    launch {
      for (fruit in fruitArray) {
        kotlinBufferedChannel.send(fruit)
        println("Produced: $fruit")        
      }
      kotlinBufferedChannel.close()
    }

    launch {
      for (fruit in kotlinBufferedChannel) {
        println("Consumed: $fruit")
        delay(1000)
      }
    }
  }
}

Output:

Produced: Apple
Produced: Banana
Consumed: Apple
Produced: Pear
Consumed: Banana
Produced: Grapes
Consumed: Pear
Produced: Strawberry
Consumed: Grapes
Consumed: Strawberry

You can find the executable version of the above snippet of code in the starter project in the file called BufferedChannelExample.kt.

The output is a perfect description of what is happening: You create a channel with capacity two and then you start a producer.

In the output, you can see that the producer sends two items and fills the channel. At this point, the producer suspends waiting for a consumer to consume the item and this is what is happening with the Apple. Then, a new place is available and the producer sends a Pear. The consumer consumes the Banana and frees another place in the buffer and so on. In the end, the producer stops and the consumer can consume all the remaining items in the channel.

As mentioned earlier, the capacity of a channel depends on the performance requirement of your app. A typical example is when you have a pipeline that dispatches items from a channel into multiple channels. If the throughput of the producer and consumer is different, using a buffered channel is usually a good solution.

Comparing send and offer

In the previous examples, you sent values into a channel using the send function. Depending on the channel’s capacity, send is a function that can suspend. This is happening when the channel’s buffer is full or, in case of RENDEZVOUS, when there’s not receiver ready to consume.

In the case in which you don’t want to suspend, the Channel abstraction provides the offer(element: E) function whose signature is:

abstract fun offer(element: E): Boolean

Since this method is not a suspending function, it doesn’t need to be XXX into a coroutine. If there’s enough capacity, the item goes into the channel and it returns true. If there’s not enough capacity, the function does nothing and returns false. In both cases, it doesn’t suspend.

You can try it with the following code:

fun main() {

  val fruitArray = arrayOf("Apple", "Banana", "Pear", "Grapes", "Strawberry")

  val kotlinChannel = Channel<String>()

  runBlocking {
    launch {
      for (fruit in fruitArray) {
        val wasSent = kotlinChannel.offer(fruit)
        if (wasSent) {
          println("Sent: $fruit")
        } else {
          println("$fruit wasn’t sent")
        }
      }
      kotlinChannel.close()
    }

    for (fruit in kotlinChannel) {
      println("Received: $fruit")
    }
    println("Done!")
  }
}

Output:

Sent: Apple
Banana wasn’t sent
Pear wasn’t sent
Grapes wasn’t sent
Strawberry wasn’t sent
Received: Apple

Here, you will notice a few things:

1. The capacity of the channel is 0 (RENDEZVOUS).
2. Using offer() is similar to send().
3. As soon as the first value("Apple") is sent, the channel is full.
4. Once the channel is full, calls to offer() doesn’t send anything. Instead, it returns false, which is denoted by print statements Banana wasn’t sent and similar statements.
5. Once all the calls to offer() have been made, only one item was actually added to the channel. Thus, when the consumer receives the values it is just that value.

You can find the executable version of the above snippet of code in the starter project in the file called OfferExample.kt

Note: The caveat with the offer() is that it doesn’t guarantee that the element will be added to the channel. It won’t be added if the channel is full.

Comparing receive and poll

In the previous section, you’ve seen that a producer can use offer as a not suspending version of the send function. What about the consumer? In this case, the version of receive without suspending is the poll function whose signature is:

abstract fun poll(): E?

Since this method is not a suspending function, there is no need to invoke it inside a coroutine. It retrieves and removes the element from the channel and returns null if the channel is empty. If the channel was closed for receive, it throws the close cause exception:

fun main() {

  val fruitArray = arrayOf("Apple", "Banana", "Pear", "Grapes", "Strawberry")

  val kotlinChannel = Channel<String>()

  runBlocking {

    launch {
      for (fruit in fruitArray) {
        if (fruit == "Pear") {
          break
        }
        kotlinChannel.send(fruit)
        println("Sent: $fruit")
      }
    }

    launch {
      repeat(fruitArray.size) {
        val fruit = kotlinChannel.poll()
        if (fruit != null) {
          println("Received: $fruit")
        } else {
          println("Channel is empty")
        }

        delay(500)
      }

      println("Done!")
    }
  }
}

Output:

Received: Apple
Sent: Apple
Received: Banana
Sent: Banana
Channel is empty
Channel is empty
Channel is empty
Done!

Here, you will notice a few things:

1. The capacity of the channel is 0 (RENDEZVOUS).
2. Using poll() is similar to receive().
3. As soon as the first value("Apple") is sent, the channel is full. The consumer then receives the value, after which time the channel is empty again.
4. Another cycle of the above process runs with the second value "Banana".
5. For the third value, "Pear", because of the if check in the for loop, no more items are sent in the channel, i.e., the channel is empty.
6. Once the channel is empty, calls to poll() returns null, which is denoted by print statements "Channel is empty".

You can find the executable version of the above snippet of code in the starter project in the file called PollExample.kt

Error handling

As you have seen in the previous examples, exceptions play an important role in the way you can use a channel. It’s crucial to understand what the main exceptions are and what you should do when they happen. You have to consider two main use cases, depending on if you’re on the producer side or on the consumer side of the channel.

You’ve already seen that, when you consume all elements from a closed channel, its isClosedForReceive property returns true. If you consume the channel using a for loop, everything works in a transparent way. If you attempt to consume a new value, you get a ClosedReceiveChannelException.

When this happens, the channel is considered a failed channel. A failed channel re-throws the original close clause exception on received attempts.

Here’s an example:

fun main() {

  val fruitArray = arrayOf("Apple", "Banana", "Pear", "Grapes", "Strawberry")

  val kotlinChannel = Channel<String>()

  runBlocking {
    launch {
      for (fruit in fruitArray) {
        // Conditional close
        if (fruit == "Grapes") {
          // Signal that closure of channel
          kotlinChannel.close()
        }

        kotlinChannel.send(fruit)
      }
    }

    repeat(fruitArray.size) {
      try {
        val fruit = kotlinChannel.receive()
        println(fruit)
      } catch (e: Exception) {
        println("Exception raised: ${e.javaClass.simpleName}")
      }
    }
    println("Done!")
  }
}

Output:

Apple
Banana
Pear
Exception raised: ClosedReceiveChannelException
Exception raised: ClosedReceiveChannelException
Done!

Here, you will notice a few things:

1. The capacity of the channel is 0 (default).
2. Once close is called, all values retrieved after that raise the ClosedReceiveChannelException.

Note: You can find the executable version of the above snippet of code in the starter project in the file called ClosedReceiveChannelExceptionExample.kt

However, this is what is happening for a receive operation. When you close a channel on the producer side, its isClosedForSend property becomes true.

If you attempt to send another value, you’ll get a ClosedSendChannelException. Also in this case, when this happens, the channel is a failed channel. Any further attempts to send an element to a failed channel throws the original close cause exception.

Here is a functional example:

fun main() {

  val fruitArray = arrayOf("Apple", "Banana", "Pear", "Grapes", "Strawberry")

  val kotlinChannel = Channel<String>()

  runBlocking {
    launch {
      for (fruit in fruitArray) {
        try {
          kotlinChannel.send(fruit)
        } catch (e: Exception) {
          println("Exception raised: ${e.javaClass.simpleName}")
        }
      }

      println("Done!")
    }

    repeat(fruitArray.size - 1) {
      val fruit = kotlinChannel.receive()
      // Conditional close
      if (fruit == "Grapes") {
        // Signal that closure of channel
        kotlinChannel.close()
      }
      println(fruit)
    }
  }
}

Output:

Apple
Banana
Pear
Grapes
Exception raised: ClosedSendChannelException
Done!

Here you will notice a few things:

1. The capacity of the channel is 0 (default).
2. Once close is called, all values sent after that raise the ClosedSendChannelException.

You can find the executable version of the above snippet of code in the starter project in the file called ClosedSendChannelExceptionExample.kt.

Comparing Channels to Java Queues

As mentioned, Java offers a similar solution for handling streams, called Queue<E>, which is an interface and has several implementations. Take a look at an implementation of the BlockingQueue<E> interface, as it supports a similar behavior as Channel of waiting until a queue has space before inserting an element.

public class BlockingQueueExample {

  public static void main(String[] args) {

    BlockingQueue<String> queue = new LinkedBlockingQueue<>();

    System.out.println("Beginning:");
    try {
      System.out.println("Let’s put in basket: Apple");
      queue.put("Apple");
      System.out.println("Let’s put in basket: Banana");
      queue.put("Banana");
    } catch (InterruptedException e) {
      e.printStackTrace();
    }

    System.out.println("Done!");
  }
}

Output:

Beginning:
Let’s put in basket: Apple
Let’s put in basket: Banana
Done!

You can find the executable version of the above snippet of code in the starter project in the file called BlockingQueueExample.java.

In the code snippet above, you create an instance of LinkedBlockingQueue<String> and put a couple of values in the queue. Pay attention to the differences between Java Queue and Kotlin Channel:

1. If the queue has no space left, the current thread would be blocked, until another thread takes an item from the queue, instead of just suspending the coroutine, which is not a good option considering the resources necessary for thread handling.

2. BlockingQueue has a blocking put operation, while Channel has a suspending send. Moreover, instead of a suspending receive operation on Channel, it has a blocking take operation.

3. As the current thread potentially could be interrupted, it’s necessary to use try-catch block to handle the possible exception.

4. There is no way to stop queues from accepting more values, whereas a Channel can be turned off to indicate that no more elements will enter the Channel.

Therefore, the common recommendation for the usage of Java Queues is to use non-blocking method for inserting and retrieving items (offer(E item) and poll()) to avoid blocking a thread and spending extra resources.

It’s possible to use a BlockingQueue instead of a Channel for a typical producer/consumer scenario like the one in this Kotlin code:

fun main(args: Array<String>) {
  // 1
  val queue = LinkedBlockingQueue<Int>()
  runBlocking {

    // 2
    launch {
      (1..5).forEach {
        queue.put(it)
        yield()
        println("Produced ${it}")
      }
    }

    // 3
    launch {
      while (true) {
        println("Consumed ${queue.take()}")
        yield()
      }
    }

    println("Done!")
  }
}

1. You create a LinkedBlockingQueue as an implementation of the BlockingQueue interface.
2. You launch a coroutine that inserts 10 numbers into the queue: the producer.
3. This is the consumer that uses the blocking take function in order to consume.

In general, the yield method for the Thread class is a way for asking the system to suspend the current thread in order to allow other threads to proceed. It’s important to note that this is not guaranteed and the scheduler could simply ignore it. Anyway, in your case, this is a Kotlin suspending function and the output is proof that it actually works:

Consumed 1
Produced 1
Consumed 2
Produced 2
Consumed 3
Produced 3
Consumed 4
Produced 4
Consumed 5
Produced 5

As you can see, the yield suspending function knows what coroutines are running and can then suspend one in favor of another.

You can find the executable version of the above snippet of code in the starter project in the file called BlockingQueue.kt

Key points

1. Channels provide the functionality for sending and receiving streams of values.
2. Channel implements both SendChannel and ReceiveChannel interfaces; therefore, it could be used for sending and receiving streams of values.
3. A Channel can be closed. When that happens, you can’t send or receive an element from it.
4. The send() method either adds the value to a channel or suspends the coroutine until there is space in the channel.
5. The receive() method returns a value from a channel if it is available, or it suspends the coroutine until some value is available otherwise.
6. The offer() method can be used as an alternative to send(). Unlike the send() method, offer() doesn’t suspend the coroutine, it returns false instead. It returns true in case of a successful operation.
7. poll() similarly to offer() doesn’t suspend the running, but returns null if a channel is empty.
8. Java BlockingQueue has a similar to Kotlin Channel behavior, the main difference is that the current thread gets blocked if the operation of inserting or retrieving is unavailable at the moment.