12. Broadcast Channels

Written by Nishant Srivastava

A channel is all about transferring a stream of values. It is quite common to put a stream of items in the channel and then have receivers consume the items as they are emitted. It works when an item is sent in a basic channel and when emitted it is consumed by a receiver. Other receivers do not get the same item; instead, they wait for another item to consume from the channel.

Often times, you will encounter use cases in which you would like all the receivers to consume the same value. This is where a broadcast channel comes into the picture. This and much more related to broadcast channels are covered in this chapter.

Getting started with broadcast channels

With the channel, if you have many receivers waiting to receive items from the channel, the emitted item will be consumed by the first receiver and all other receivers will not get the item individually. In fact, in such a scenario wherein there are more than one receivers, there is the possibility of a race condition.

Take a look at this code snippet:

fun main() {
 
  // 1
  val fruitArray = arrayOf("Apple", "Banana", "Pear", "Grapes",
      "Strawberry")
  // 2
  val kotlinChannel = Channel<String>()

  // 3
  runBlocking {

    // 4 Producer
    GlobalScope.launch {
     // Send data in channel
     kotlinChannel.send(fruitArray[0])
   }

    // 5 Consumers
    GlobalScope.launch {
      kotlinChannel.consumeEach { value ->
        println("Consumer 1: $value")
      }
    }
    GlobalScope.launch {
      kotlinChannel.consumeEach { value ->
        println("Consumer 2: $value")
      }
    }

    // 6
    println("Press a key to exit...")
    readLine()

    // 7
    kotlinChannel.close()
  }
}

Here:

1. A string array of fruit names is created, named fruitArray.
2. A basic channel is created named kotlinChannel.
3. Next, a runBlocking section is defined to run coroutines in our main function.
4. Start producing items and send them in the channel, all inside a launch coroutine builder.
5. Start consuming items from the channel, all inside two different launch coroutine builders.
6. Wait for a keystroke to exit the program. readLine() basically waits for standard input, and it is used here to stop the program from exiting before finishing its async operations.
7. Close the channel so that the consumers on it are canceled, too.

The output of this code snippet when run will be:

Press a key to exit...
Consumer 1: Apple

Note: To finish the program, you need to press the Enter key.

Here, you can see that there is one channel to which some values are sent. Then there are two consumers — i.e., two consumeEach calls on the channel being executed to consume the values being emitted by the channel. Now, which of these two consumers gets the value is not obvious. In fact, if you run the same program many times you might see the below output, too:

Consumer 2: Apple
Press a key to exit...

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

Thus, as you can see, it is not obvious which consumer will get the value every time the program is executed. Based on which consumer receives the value first, the value is consumed by that consumer and the other consumer does not get the value.

To mitigate this, the Kotlin Standard Library provides another type of channel called the BroadcastChannel.

The BroadcastChannel is non-blocking by nature and maintains a stream of values between the sender and the many receivers that subscribe.

Note: This is an experimental API. It may be changed in future updates.

To do this BroadcastChannel uses the openSubscription function and subscribes to values being sent into the channel. It is important to understand here that only when the subscription is obtained will the consumer receive the values being sent into the channel. Anything sent before obtaining the subscription is not received by the subscribed consumers of the channel.

You will use a similar code snippet of the channel’s race condition when there are many receivers. But this time you will make a slight modification. Take a look:

fun main() {

  val fruitArray = arrayOf("Apple", "Banana", "Pear", "Grapes",
      "Strawberry")
  // 1
  val kotlinChannel = BroadcastChannel<String>(3)

  runBlocking {

    // 2
    kotlinChannel.apply {
      send(fruitArray[0])
      send(fruitArray[1])
      send(fruitArray[2])
    }

    //3  Consumers
    GlobalScope.launch {
      // 4
      kotlinChannel.openSubscription().let { channel ->
        // 5
        for (value in channel) {
          println("Consumer 1: $value")
        }
        // 6
      }
    }
    GlobalScope.launch {
      kotlinChannel.openSubscription().let { channel ->
        for (value in channel) {
          println("Consumer 2: $value")
        }
      }
    }

  // 7
    kotlinChannel.apply {
      send(fruitArray[3])
      send(fruitArray[4])
    }

  // 8
    println("Press a key to exit...")
    readLine()

    // 9
    kotlinChannel.close()
  }
}

Here’s what’s going on above:

1. A BroadcastChannel with a capacity of three is created named kotlinChannel.
2. Start producing items and send them in the channel, all inside a launch coroutine builder. The first three items from fruitArray have already been sent in kotlinChannel.
3. Start consuming items from the channel.
4. Here, a subscription is opened on the kotlinChannel using the openSubscription() function — i.e., start listening to values being sent in the kotlinChannel.
5. Iterate over all the values in the channel and print them out.
6. When finished iterating over the values in the channel, the subscription is closed — i.e., stop listening to values being sent in the kotlinChannel.
7. Now that the subscription has been obtained on the kotlinChannel, send two more values in the kotlinChannel.
8. Wait for a keystroke to exit the program.
9. Close the channel so that the consumers on it are canceled, too.

The output of executing this code snippet will be:

Press a key to exit...
Consumer 2: Grapes
Consumer 1: Grapes
Consumer 2: Strawberry
Consumer 1: Strawberry

Note: To finish the program, you need to press the Enter key.

As you can see, both the consumers that had opened subscription on the BroadcastChannel received both the values sent into the channel — i.e., Grapes and Strawberry. That is how the broadcast channel simplifies the whole process of broadcasting the values in the channel to all receivers.

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

Yet, there is one optimization that you can still do. Similar to how there is a consumeEach helper DSL defined for a channel, there is one defined for BroadcastChannel, which subscribes and performs the specified operation for each received item.

Take a look at the implementation of the consumeEach method:

public suspend inline fun <E> BroadcastChannel<E>.consumeEach(action: (E) -> Unit) =
    consume {
        for (element in this) action(element)
    }

Diving deeper into the source code:

public inline fun <E, R> BroadcastChannel<E>.consume(block: ReceiveChannel<E>.() -> R): R {
    val channel = openSubscription()
    try {
        return channel.block()
    } finally {
        channel.cancel()
    }
}

You will notice that consumeEach will call the openSubscription() method. That means we can replace openSubscription() calls with just the consumeEach DSL straight up.

In the last code snippet, simply replace the following lines of code:

kotlinChannel.openSubscription().let { channel ->
        for (value in channel) {
          println("Consumer 1: $value")
        }
        // subscription will be closed
}

With:

kotlinChannel.consumeEach { value ->
   println("Consumer 1: $value")
}

Note: The replacement code snippet shown is only for Consumer 1, you will need to do the same replacement for Consumer 2

Now run the code snippet again. You will see that the results are the same. This is just a more concise and idiomatic Kotlin way of consuming values on a channel.

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

Often times, one of the common use cases is to be able to get, at least, the most recently emitted value on subscription. This is where ConflatedBroadcast channel comes into play and is explained in the next section.

ConflatedBroadcast channel

Like a BroadcastChannel, ConflatedBroadcastChannel enables many subscribed receivers to consume items sent in the channel but it differs in one aspect: a ConflatedBroadcastChannel only emits the most recently sent item while the older items are lost. Also, any future subscribers to this channel will receive the item that was most recently emitted.

To understand how it works in practice, take a look at the following code snippet:

fun main() {

  val fruitArray = arrayOf("Apple", "Banana", "Pear", "Grapes",
      "Strawberry")
 
  // 1
  val kotlinChannel = ConflatedBroadcastChannel<String>()

  runBlocking {

    // 2 
    kotlinChannel.apply {
      send(fruitArray[0])
      send(fruitArray[1])
      send(fruitArray[2])
    }

    // 3
    GlobalScope.launch {
      kotlinChannel.consumeEach { value ->
        println("Consumer 1: $value")
      }
    }
    GlobalScope.launch {
      kotlinChannel.consumeEach { value ->
        println("Consumer 2: $value")
      }
    }

  // 4
    kotlinChannel.apply {
      send(fruitArray[3])
      send(fruitArray[4])
    }

    // 5
    println("Press a key to exit...")
    readLine()

    // 6
    kotlinChannel.close()
  }
}

Going through this step-by-step:

1. A ConflatedBroadcast channel is created named kotlinChannel.
2. Start producing items and send them in the channel, all inside a launch coroutine builder. The first three items from fruitArray are already sent in kotlinChannel.
3. Start consuming items from the channel. Here a subscription is opened on the kotlinChannel and each emitted value is consumed and acted upon using the consumeEach DSL block — i.e., start listening to values being sent in the kotlinChannel and act on them as defined inside the DSL block.
4. Now that the subscription has been possibly obtained on the kotlinChannel, send two more values in the kotlinChannel.
5. Wait for a keystroke to exit the program.
6. Close the channel so that the consumers on it are canceled, too.

The output of executing this code snippet will be:

Press a key to exit...
Consumer 2: Strawberry
Consumer 1: Strawberry

Note: To finish the program, you need to press the Enter key.

From the output, you can see that only the last emitted item in the channel, which is “Strawberry” — i.e., value at fruitArray[4], is only consumed by the receivers. Initially, there were three items sent in the channel, but no subscriptions were made on the channel. Then two receivers subscribed to the channel. Next, two more items were sent in the channel namely “Grapes” and “Strawberry” in order. Since ConflatedBroadcast channel only sends the most recently emitted item, “Strawberry” was sent to both the subscribed receivers.

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

Notice that this example is exactly the same as the one for BroadcastChannel. The only difference is the kind of channel initialized in both. This was done specifically to show the difference in how BroadcastChannel and ConflatedBroadcastChannel works, as well as their output.

Another thing to note about how ConflatedBroadcastChannel differs from BroadcastChannel is their suspend behavior of the send operation. The send operation on a BroadcastChannel does not suspend if there are no receivers, but the send operation on a ConflatedBroadcastChannel never suspends at all.

ReactiveX vs. BroadcastChannel

Reactive programming uses a similar kind of approach to handle streams of data as the Kotlin coroutine channel. Like channels, reactive programming has observables with data sources that emit items and should be observed. Then you also have the observer, which is basically a consumer of items emitted by the observables. To track the flow of data, observers subscribe to observables, which emit data items and those are then consumed by observers. There can be many observers observing an observable.

Notice that the data flow here is unidirectional. Observables emit data items and observers consume the emitted data.

Note: Reactive programming is a complete topic in itself on which a book can be written. You will be focusing on the reactive approach as it resembles the behavior of a broadcast channel.

To implement this approach in various languages, there are corresponding libraries. For JVM-based languages, you have RxJava, which simplifies and enables the reactive programming approach when writing code for the Java platform.

To add RxJava to your projects, you simply need to add the dependency in your build.gradle file with:

implementation "io.reactivex.rxjava2:rxjava:2.2.6"

Note: As of this writing, RxJava is at version 2.2.6.

Once you sync your project, you will have access to RxJava classes.

RxJava consists of a class that enables broadcasting capabilities, called a subject. When you create a subject and send an element, all subscribers will get the same object at the same time.

Take a look at the code snippet below:

fun main() {

  val fruitArray = arrayOf("Apple", "Banana", "Pear", "Grapes",
      "Strawberry")
  // 1
  val subject = PublishSubject.create<String>()

  // 2
  subject.apply {
    onNext(fruitArray[0])
    onNext(fruitArray[1])
    onNext(fruitArray[2])
  }

  // 3
  subject.subscribe {
    println("Consumer 1: $it")
  }
  subject.subscribe {
    println("Consumer 2: $it")
  }

  // 4
  subject.apply {
    onNext(fruitArray[3])
    onNext(fruitArray[4])
  }

  // 5
  println("Press a key to exit...")
  readLine()
 
  // 6
  subject.onComplete()
}

Here:

1. An instance of type PublishSubject is created and called subject.
2. Start producing items — i.e., publish them using the subject.onNext(item) function. Here, you will see that the first three items from the fruitArray are published to the subject.
3. Subscribe to items published from the subject using subject.subscribe{} — i.e., start listening to items subject is publishing. When a value is published then they are printed out to the standard console.
4. Next, publish two more values from the subject.
5. Wait for a keystroke to exit the program.
6. Signal completion of the subscription.

The output of executing this code snippet will be:

Consumer 1: Grapes
Consumer 2: Grapes
Consumer 1: Strawberry
Consumer 2: Strawberry
Press a key to exit...

Note: To finish the program, you need to press the Enter key.

Here, you will notice three things:

1. This code snippet is very similar to BroadcastChannel’s code snippet. The only thing that changed was that the channel was replaced by a subject, and the corresponding methods for sending/consuming were used when it came to using subjects. There are two observers/consumers subscribed to the subject and, when a value is published from the subject, the observers all receive the value.
2. The output is the same, although it is ordered for the subject.
3. Before the subscription is made, values published by the subject are ignored. Once the subscription is made, the two values published by the subject are received by both the observers.

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

Similarly to how the Subject is the dual of the BroadcastChannel in behavior, ConflatedBroadcastChannel has a dual in RxJava called BehaviorSubject.

It means an observer will receive all the elements that the source of information emits after its subscription, but also the last item emitted by the source of information before the observer subscribed to the BehaviorSubject.

Take a look at the code snippet below to get an idea of how it works:

fun main() {

  val fruitArray = arrayOf("Apple", "Banana", "Pear", "Grapes",
      "Strawberry")
  // 1
  val subject = BehaviorSubject.create<String>()

  // 2
  subject.apply {
    onNext(fruitArray[0])
    onNext(fruitArray[1])
    onNext(fruitArray[2])
  }

  // 3
  subject.subscribe {
    println("Consumer 1: $it")
  }

  subject.subscribe {
    println("Consumer 2: $it")
  }

  // 4
  subject.apply {
    onNext(fruitArray[3])
    onNext(fruitArray[4])
  }

  // 5
  println("Press a key to exit...")
  readLine()

  // 6
  subject.onComplete()
}

Taking each part in turn:

1. An instance of type BehaviorSubject is created, named as subject.
2. Start producing items — i.e., publish them using subject.onNext(item) function. Here, you will see the first three items from the fruitArray are published to the subject.
3. Subscribe to items published from the subject using subject.subscribe{} — i.e., start listening to items subject is publishing. When a value is published then they are printed out to the standard console.
4. Next, publish two more values from the subject.
5. Wait for a keystroke to exit the program.
6. Signal completion of the subscription.

The output of executing this code snippet will be:

Consumer 1: Pear
Consumer 2: Pear
Consumer 1: Grapes
Consumer 2: Grapes
Consumer 1: Strawberry
Consumer 2: Strawberry
Press a key to exit...

Note: To finish the program, you need to press the Enter key.

Here, when you execute this code snippet, the output is very interesting. First, the BehaviorSubject publishes three values. Now, two observers subscribe to the subject and as soon as the subscription completes the most recent value published, which is fruitArray[2] — i.e., “Pear” is received by both the observers. Next, when the subject publishes two more values, both the subscribed observers receive the values and consume it — i.e., print it to the standard console.

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

You will notice it is a bit different in behavior than ConflatedBroadcastChannel because it receives the last published value when observers subscribe and then waits until the observers are unsubscribed for the values published by the subject to be received by the observers. But for ConflatedBroadcastChannel only the last published value is received.

Key points

• With channels, if you have many receivers waiting to receive items from the channel, the emitted item will be consumed by the first receiver and all other receivers will not get the item individually.

• BroadcastChannel enables many subscribed receivers to consume all items sent in the channel.

• ConflatedBroadcastChannel enables many subscribed receivers to consume the most recently sent item provided the receiver consumes items slower.

• Subject from RxJava is the dual of the BroadcastChannel in behavior.

• BehaviorSubject from RxJava is the dual of ConflatedBroadcastChannel in behavior.

Where to go from here?

Kotlin channels introduce a very simplified approach to handling a stream of data with well set-up constructs to enable better and faster development. This is not all the information about channels because there are operators, which enable various operations on the consumption of results. You will read about those in detail in the next chapter.