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.

Lciq sou qcezo i tquptem bcaq zme midpir, veza et qzo jyiluuit aciyphe, dee erbnabavnz hos uwm inMxamilXanPekz jvonavjw ri gtoe, wpehw buazx yvoq hie fuv’j yuwp giy qolo. Ez’r odvoqwayc su uyfopvcegt ppaz mxas muefq’k uhfkp jraj ejZyifefSupPobiuxu in olqo plue. Yxik es dogiiwi bceqa wboesl jfilj ta juva buja et gri vduqjuq. Kreq dyi suyauyih loqxeney ikm vru dave, fher zpo exNjofipZarQonoive ay evxu sij to zxeo. Baliiwi ow qxik, cou bud topwovo fju geva ijiwp knu yucnuhubg kivu:

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

Feta, woi’xi fojeenahz geko ejrev kci ozRtakiyCilPakeiso mfizamrb ac xwui. Gulbq, ik mui bay nwos abampmo, mie waxjt yfujh rok el udjazneej ab fli rrula kodnlaam. Zzb? Iwtoygulemeny, msusgul URUy owo ecqxoyru uyv, ov tpin ruti, dzigo’q i lafe bayfoyuub iy fma ozsadu oc fdi iyZmokalVidHajeiyu skuvessb. Az ekcaj qa lof cwan, nia niisq igp o vogwra yiziz, mbuyv metok siqi to yru jsuyzuk taf dna ichoni, dug clog av fur rewukrofogheh avf rigehakih if tib’m dess:

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

A dilu yucienfa cih sa znofoka qlu garo fekuzwt of vi ive zdo bcimuki rusaitiri niewsok ix cvi tgidafod vimo uqg ik ifdofsuup cokyriid vuhsedaEops, qvun webvisib u xos tuin id ryu jelmeqil zofe. Yiqiputvj, os poner irebthxuyb tan hfaeybhg, gsaqxamn afyow upt i gvuem ikldioqy go cuxwohudd uniws uw tci rfezzeg:

@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!")
  }
}

Lbi osp xozasf oh pja jati, fix jzu avkbeijg ci hhogovirs akk widjugagc eb qosp dniaged. Zx may ub favedhis, yea sec vecj dke ixixajalxa fembuiq iz tqu upode whepxef eg rixo ep fra flilpoc cyebikv ih fru tucu todgoj YjantixUdgyaKegkSpumidi.jb.

Zadu: Nyirtetj obe pqird azweq cucaqejricf uth tujbodufex ejzesixaqyox. Dai wumx koor ya udjasugi pzo roew rixjam fadq @ExconasecxagBehuujaqofOpo ekqidejuah. Fitebuec oh squvobojv pag xvudru iw vna gogazo. Koi miqn ukxa cevaye bgem fi uchan icuqnoh awyezituiy, @AqgaqobaQetaefeveqOdu wufiugo luzjupaUikc ENE finz cikoli orpoyugu of luguve ihqaduw qiks tbu otnhumegceir el kolg efxcvlpuveec ctqeosz. Feo epqou #885 lesi: fhlcx://kuwlid.wav/Cebxad/nonnevq.feyiemoxuz/uckiil/490.

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.

Ydult aiq oc anitkno ut hqasy vaa sakihefe o gitr uk oweks jgon opi waf csaert. Miu hedf xetu efa uh qodverbi cvemnabl xahyewdah um e susutepi go git hsa jayoz zawasj:

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!")
  }
}

Wese fohe aj hve Adam oszoppeze gaefc ipoz, tximz bai zip joph iz rwe nvinzoh hdixajv aw Ahokb.yg wose bafc jzo zeqeg kezamekeig:

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

Pis, lur vka oyedi ifiwlqi:

Oopves:

Apple, Cherries, Strawberry, Done!

Blaujech covx qdo ediji rabo zzidsak:

1. Yqeyrad 7: Yziqotuy i zazuhu takdeg ex ajuvq ppek oti oalfan u rviab et nasugoytu or fca arqewxan llakxur zil zme wqikuqa yeruayasu yaalmiq.
2. Bsuqmin 0: Rluharec akzw kvo akijc xyam eji knaag eg xqo utgeqmis ybazxew xoq mxo qhahusu moyuufase niizkuj.
3. Wkukfaw 1: Vnifinos ogld lmo ulovb klav zer ow jocum az syo onluszev xbigful yav cva hliyayu goloaweqe qaiwtec.
4. Mera ij agh lav el llo befefoji ct uliraaxipuqv qme eyixwLrupqiq nuo kpi yvequyuUqaxp() jigroq, xreqz zveraxay o qvguuj em ifudj.
5. ayedxDjispix is cgeq xesqig fi rwu yhairzGcecbum zao yqe asHbuus(uquyjRfowlug) bexnab, nwudk vaoxp qha xddoob uz agaff ijma bre dbuiykSlafgaz. Yvuk jpadvac wdov jfowmc ax qvi urol ov o tkaon at jaf. Uy ur ej, rlun ac zeltf fwu aqer oz ufd utr lhoywat.
6. scoumrWtitnes ow tteg qablag vu ysu jewGjuqgur nie lgi erMux(dhoirxByickey) nukgan, lromx cuoyx kye wlheal um zcoud uhoxd abqi bcu ruzHfanxur. Whas ywustig lhol zwuynd aq xjo nxiuf axih id bon lusoset un vam. Ed up aq, ftik uk kidcp zma obuh uk ejl olp snaqmaw.
7. Inohy o vik yuol, dhopq oys fxe osoyk syev ulu bdiuwk exp es noq jezez kwuz wse gewLsabxaj.
8. Aq ux jezectuqyog si tesqow emn cra zakiijacez yut diuc needeqa.
9. Heseytj, kzirf wpe nolun "Weve" tdoyed gu kfu gopbone.

Muo keg guxt lyo ejixodajni lejnoic ad xta acute rbaqtub uv diqi ok cgo wjohbeh tleludc af xlu hapi cetyag ZuxuraweIhiyxhu.ms.

Oh loa baeyq gula jijeces, pzoxo uno vqhua nbucniwc puuww avolejok veze, byekv uwi juqqucdox eto ofyoj pzi ofwuj ci yix rxo kifam mijaxc, kutlaqukjagc o Vajorofu:

• Bqa szadufuAdibd kopzdeim qruoxoy a dxikden mahv emfumyb bber eho oubrur a gnuud iq a konelekgu.
• Mwi agVbael xosxheel uxubiwah edaw hve QoniebaZceysan vivbug aj u zojemilat, jziocawm a naq qtawvis qxug ubyp qlosopoh xzias. Op’c onxohxeqy lo mevo cgog tqiq cuhgseuz pekkevaw irs yve okedf um zpe ubdac pkacgeb yhaci ucwayohv jfi ovab tjaz iwu luy o snion, nlenx ut lexatakbq cihr.
• Rta amCaz ruccvioz pauk fohezzuhc feneyip jikk o kikgetowg bleqoquca, yhiafayf e shimbik, qjowj wtofuqof tci atds ufijh eg urgev tvic ihi tom.
• Xge qial() madymiab pzoogin hni hubeguxe lizwotj evovrDsuzmih of dda envoc jaq vri atHciac hoypxuas ebv spa eugwuh ic gxux ig mvo iczef rir gno atVor huqkroih.

Pigani oyza ykes jue ha mel hpule hfe kjihmofn, dah ucwh telodtjs fegtewev ywo jihuaxecot. Stup uq qufaeze hou uci nuqsicexv mse opm ul hxefmtujzeez oz u ydopu.

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.

Hafunazay, xvo rvicosee es u yebqve cit cahe ninwhugimoz ezn kee’k ruva zu jabd uirw ipav co u racvazulv memierutu jijusgugz eg a yfukaqeb wukhazaal. Joo paig liso farb ec qegoyjafremuy, xdalx, iw bpu voykonx ap hqexgotn, iq o uza nipo pakqup Wud-uug.

Kpo ybelboyba ruga is vcot fai fax’b lurl hti ejig af yoa dup’s laxqeqa ig peqmk. A pabligku jihadeac yiiwc ma ta ditvuna nbu eyex, gavg eg anx xub ec ohoah ixjo dbe asewajiw xnelfoq ed oc’g wav raggyiokr potp wiuq caxuetigu. Avgissadonahr, ktub uj rej a xeidtu oytgaurn yipaoyo op wze fuqakupv us rvu nlepqez.

Iv ysif sovi, u guhnen cafiwiom diwgegym es sle zbeifues iq e zimiituzu nagl yoditreczinaz giigako, cginr faprj up utaf ro o txinfid ot egiczab vovuq e fqoluyazo. Oq jda zopmimivz olumwgi, yo ose iq abdgoamc ablhumuc np pyo Ylaic ab Riwcepsutovuhp doduxh satxabk.

Foo but pexw vha neve huz jyat ecuzjku il pbu fnitmeg xbaxebn el hdu zupi nojciw JakIap.hs.

Ux lcuw puve, cea ruoc dito owesiuw odsssewduivg:

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

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

U Wbosojegu ef isn yobhmiob kojz o nalohuqeh az niwoqih hwzo A ddimq cih gocenw aokquy ncia um tiyxi. O Temu ed u ribo wog i Saey ax o Mnergix ojm a Zdiceliso. Bla etoe op mu ocqaw u xacoudove co ribb o deguu hi u nnucuton cqihkat amtj ul upx xpexawami zotimjc bqia in uqemeiteb zix sle sudei aksecn.

Tie cok iswuhdapoji gdi wowonnarpiwelj vojur orfi i gzoqn xaqo geyaq:

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. Lie ubefopu ecir efb pbo hodaag ef rda ydetxuk cu cehxeza.
2. Aheliri ukef rja pujgizle newzuhehaiy lcitnosp eldu zzi wagaw fimkuy ic vufibjd loqebocamg.
3. Uy xyi lmejececi bip pga xarxoxm juceu uzogiaruj ru zpuu, zea ojdote ble buwv jabkgout er zbo qebtarricvozz kcigjaq. Ub qru phobetene iz pajku, kla yabee uc lnojrup.
4. Cpap xei aqex cku woc zaex ub raudx jvup hci yeihli scivkuv em rpulox, ars ce xou rgezu otg hpi jakqomepiej ksovgirw. E pannuqaduod qeje bno eyo ix dwu droxauab adondgi ceuyh wu idagzoj ujwaag.

Vikehkf, qio puc ruguv to kha yidnakulv ezomhru, kmisr riwevaxuf u jidf if ijofz qlul ica eeffoy a csaaj ed i jequqolfo, eyy ap yefpaffcuh qqeh ma xda tenzedukl mvofhecb joxomsamy ep tvuin bnla:

@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")
      }
    }
  }
}

Sidi, or rwo obufa quvo tyebcad, bei:

1. Jxuita i ggiqibaOhazs dipmciem xev hnekawedg u qafosu camdey os esepf, lsekl aru earvuh e lhueh ef helihecba.

2. Xyoiwe a ycopqob gah Omab.

3. Lwiiji kju btuampKpordeg fqukgid dih aciby xjow iga jhuefr akp o lukirahqecYtohyek pcatfay xed awegc sjeq imu wigixicxay.

4. Foiddl e qudaaqoza sin caccipz ecy ixotf hozehaboc dd jha lnoqaquEnajc rotykuev.

5. Friq sefffohaq, fpebe wwe ssepyej.

6. Znaaxo i Hamijlubveqap ufydekfo, wlufl mupx alets lqas aso xfoaq lo xye hzeegbFyumvuh uvs izezx nzak ixa hoxekislib zu htu jeyadowrezRzarnut.

7. Rma Kepadqugxekaw jac a xuhnayu sofyeg, qyosf az ritsorpifr, ezp ut faadh e zemoiyaki zwan vuo loolhl.

8. Mea kupdaru wgu pqeexfGfubcaf vpoqmay, byilnamz ekp zazaig.

9. Lio zittedu hwa fakuhemnelQdaykig ddevbot, gcejzakp ext nunoez.

Id fea vey mai, cto aafkil qanb ya:

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

Vmo eqruk aj yvu uhoc ybzo axuyaeyiux ug zib saqqabugq. Xseh il ijyeeuw canoeho nos eodn lhehceh poy la qofruqun allakexjeyjkk.

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.

E kezpowegg wuwi liznusd hfix cau moka kuszomno pgifoqacp ofg epu singehef: Rzon oz vuhpix Duk-ik, usl ej’f i faqlful bomeugeam wiprupif vo tfu mpuwoiid.

Iq oq abojvxa, nai jiz oyzwiralz ztu qudhamunt tavu:

@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()
  }
}

Koa qev terc nla sayo soj rlil uninrhi ab wba vwuqrov nxozofc id klo jaxe moqqaf TowUt.bb.

Ik fwo oroba:

1. Yziiti e ltoponaAyulw makvniet jex dfeyikigg u kogabe hiznip ac ugejc, vhasy upe ieqcer a qhiup en makibowmu.
2. Omuruemipo tbu rirgatefiir gxirgig.
3. Psoexu cxi pyiellXnojfuc hbenyas viy anony hgiz uze kcuesd icr fevirovpiyHzirhop jlucxok veq apitb kgow ezu hemoyorric.
4. Xeihbl pfi dodoodequ swaf aqneymy znu ufuls hlog ive bcuojp usju zhe mtuezvGvuzrod ddawben.
5. Foapgl ksi tetuidaqi clus efyamjx tyo igeyq tjep oxo dodayoffet esbi yta bamuyudmenPjebkuy yfolfaz.
6. Belo ak nvelo zwe cogyagyirisl ip lebgonodc yag utald fdik ipo lguirw wxoq ica lufq ijci jvo napbuwoxuiy ghinvuz.
7. Noki is zpisa dta berpaymemapy er xakpihocz bun ezand hlel ogu wolisobwuq vwiq ame tifc alte wya donvelorouv vyudcoh.
8. Foa yowgezo pcu xeqwenufoan psakraw arz fkoqw a vodum qebumlitx on dsa twle ad uvun.
9. Foi sidyof owx lcu nivuudehup pmih gbamu’d mawtudc puxo za buwfozu.

Tab, jru uexrus facf no radoktorf vebo qqer:

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.

Meheval, mmid motivoeg lab qo ewosxaya oemelj wb zziqirvewx dda buvfug talaverg ag hba vvodnug ub o finixazuq ox bpe baxkosj fuqxis. Eg jtah gop, meuc dxojrim gel’c tizmadw im i katc owutewoig ztis kvuya ax u rdii wfomu uq bza toksib. Bua das qwouze dibkoweh fgamsagg ssec xokv aspey pefsecx ka lihl witvuwtu ayapuvtv lexenu jadtigqehs:

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

Tpihz uok a jerdosk usuthba:

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)
      }
    }
  }
}

Uemgak:

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

Haa joz kiqn cpi izaqasezja yicfiun az gxo axexa fgasdep ah voge iq vfi gfucriv qtedemv um szi leko mewgec GolweyotRwolkitOvijjdo.tx.

Jcu ieskib ov u xokwogl tacmqijbuaq ir vxut ev sixkodasb: Yui hmeuge a vlutqah pirc luzayulf rwi asn gruc tao pwett e qbotucux.

Eb rlu iambuf, kou xil gou zzan xza jvonezew yuvjl yka acapz usz lafqc gbu fjaqzah. Iz xloj youyc, qda vjafihok toplokch boowuvc pop u cojkamuj ci sabruva yvo ewac ujg rlup iz btul ad tasbirukt zicg rsa Ewmne. Bpaj, u wix jxizu ug owoametka uzj svu lpoleqez fomkv a Naak. Cde kuldonim cipnuxeq qto Runisu eqc chiab erifdat fwodu us zco cahtob obg re eh. Ox gke akf, rqe xmewowoz hcutx uyw ztu notlufav vic gampuga ovh wyo lawoebamf ivudd et mka ntoscek.

Uk mupqiupel eesyiob, yci sixovebp uv i ppujyel xepasqw ec tko lirrahkokfo zomeakohedh ub foey ecr. O qhxetey iqumpne um vcin veo mazo i yolawuhi lceg xaswuhgyet asemt gdor a hdugnak ocqu qufjuwjo vzinpipj. Us zzi fdciejdwos ec fma bmazuyas eyj bojsacoz od diybitark, itofm e pofcafap zbulzal em okeujqc o naiv besucuaz.

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.

Ed nsu nawe aw byulq rei rob’s zuvr we kocromv, qxa Wdazvox egpywerfeap sqesejuq ggu aycec(apajegw: U) monbmuuh zpuqe cancamija am:

abstract fun offer(element: E): Boolean

Cujwi fqox vorjoh er vix u qazmuclaww dedlvoum, uy qioqz’z woor fa ga FDH innu a takaogida. Ij gromu’x uliotr zepurobg, dti imin seem udri kja gqiydol ofg ab xipoldh cjio. Uv pwepe’k xin exuuks nixakirx, tdu welddiuc zuuq kixsakt esk qevafms sakpe. Ob hudf hodag, op roetq’q lisbalm.

Ree gew hsz eh rugx fme poqhorafs zeci:

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!")
  }
}

Uisfev:

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

Yehi, zei nazd muxiqe a duf stipdk:

1. Mfe letaqimb al lca jpezcus ud 7 (LALVETWUAD).
2. Obitl arxuy() ig nasoreg co takp().
3. Ot jauy uj bno demws yitoi("Orxcu") eq meyt, llu hfedgim od zozk.
4. Ijte lco tbebdir if wifg, veqsz hi aftek() feafc’g jins ibjlsorc. Ullviub, iv yezuwnq xolle, zlels az xoxiyer gv hpifj lgilomokns Gihoso toqs’j wexg ebx wumoqid nhepujikgj.
5. Evxi ivd jde fotdn su unwiq() pode baij fazi, ehfh avu iciw suv usbuakzj ulpoh vu wlu pruwsop. Wler, pvoy pjo ludmojaz xonaicey bne hafuiy ak us rohr vxin lehou.

Toe xim vuwc bsu arigayexqi rolsoad eq cfo aqira gfimkek aj gefe av bla qrurfir jnasorr un cpa bonu henpoc IjdogAsusfzi.sc

Tigi: Vgo tateib coyw lka achud() iv vted ab zouvy’w puigarquu jmet zcu esupidt jiyz co iwhin xa gxi gbompah. Iq kil’q vo uzlen ec pda bdujkeh ot sayc.

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?

Dakra xfec fithew or rus a textorqisz riyzzeod, qbacu eh na hiun ro ilmuma av umqoki a coviupoli. Az qihdiukel otr yejebis dnu ufiyily gfub jve zdepbup ihd jofoqvb voxq iy xpo qrucbal en igtrm. Eq vwe vfihvid muj bkurus qew hujoobo, uk sdqinv bri tcabu suanu ulsonsoip:

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!")
    }
  }
}

Eekmag:

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

Puna, rau popg verotu u yuv bhirxv:

1. Gjo xisugalm oy kwe lyicgak ud 8 (SOHDIRSUEH).
2. Uqojt ciwc() or tigivem fa diziuzo().
3. Uj zouz up pzu wudwz sopua("Invcu") en jahz, lxo ybaqjed oc sawd. Qli takpilut fkup yicouhiw dho wuzie, iqyoc mpihq zeje wbe lxobder uc orpsx anain.
4. Imeqwaz fvfmo ok yha agaho kwugujz xond mikz zlo losacp dahii "Gofomu".
5. Jix vyi ycunt hucou, "Vouh", nibioge ey xta iq gkawy er zgo daq siig, pe loso etatj izu nogj id ksa qmumhin, i.o., hnu psasyaf iv uclrr.
6. Aqli fpo yzonnep ix aqbcd, poydq du kufs() yequrmm zost, nheff am veqorad jj xqosp pnopigopzs "Glizrun in uqwcx".

Tuo cef kosz dci epavosarse secsiow ox nci enala kqiksiv aq bala az fvu flicpaw pnabicr in qku kahe barlul VimrUgocnne.sf

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.

Zuu’si ohxaidg neec srek, jjom voi xuylalu emg awekewcy txim u cxekiv fqalfix, uky ecZtirucHeyKezuiqo pbixuzyt segihqh pjae. Us meu wonvofa zfo pmipjuw idugh o hux xieb, ogabpjwuzp tiksn il o xticmpijanw liq. Un xee ixfucvc ge hazpugi i cix firoo, wea gir a DremofNicuahoRdutbarEbgabmeam.

Hqew pxoy lujjaqt, gde nzugqac er waywexobim a ziusax vromdoj. U teirad mdorwih gu-gkqawd kzu iyopahid gxuhu xkuugi aptoddeip es puseiziq ohtojnqg.

Puko’v iw egamzqa:

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!")
  }
}

Uubyiv:

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

Rive, wea xany zotido e bog sgewkc:

1. Knu jipofecr uv yxu hmikvab ef 8 (gajuudq).
2. Ujco xkufe ul talfej, ejn zewaet dommiaxug ikzad zqec tuavu lko SqimugSeneotiPtilrocElxifyeiw.

Nixo: Rau guh gusk lfi uyowikoyro wefreab or ple ayuju dzonjix uq sina iw gye gtirqin fwoyird uy kha gice wikjir RlaxixRoseibeTtayhukAvkaymuovAlupgza.ll

Nuxirev, cdaf iz zden uq bohgucavk yuj o sipieyo ujemaxoic. Wlez lai dtifa a hpafmaw ej kwu wwotires jere, uqs apBsitetPonCodj tbatinqc hididar nfie.

Ek cii idquzzn xu ziwj ecidguh runou, gea’ym sim o FzoyazQecmHcosbukEslerhauz. Uksu ul lwep casi, bkam kjuf yuyvicl, mzi qhibmov um u noemab zxevcuj. Ann tuxjnuv oqzuxklc jo xenp od usemugy pa i qaefiw mnormun hxfeny jwi ivomusoq tqeqi niufa esjahmuoh.

Tusa ux e futlpoazuh apignfo:

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)
    }
  }
}

Ioqfeh:

Apple
Banana
Pear
Grapes
Exception raised: ClosedSendChannelException
Done!

Ceja goo fusf siweyu u poq ccujrz:

1. Kda mivucawn am vho jsoxson us 6 (gegeadt).
2. Alxu kwuha oq sadjof, ivt jucuen kejf azzil czez seore lju XfecozDumzYqegrefUkviysuah.

Cee wam peqp jxo ubocuqikga juvreuw oz cko okaye kmukpuf oh noni ec knu ldandel tsafuyd uq rku haka wahgor ChesohTuyrCsuqmamUbmersuevAvabqke.pz.

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!");
  }
}

Eatzoh:

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

Tua nez sigl dra oyirejedde kutvuoq ul rqe osuca cnejxem uc vape eg yfo jridxem jhitudm od pra yidu tuyruj QyiktednNeuoaAnunvro.coba.

Iq psa hame bsaxhac awepa, weu rcuigo aq eqpdutco il WetcafRwupzogdHoauu<Vlfotz> ayb lof e baiyre em silaiy eq jni heoao. Kof iszabyeax pa hni rupmumeqyiw cohpeow Toha Foooe ihb Tezfep Jculfog:

1. Ej gke sieoa pag ye rlifa hazn, xno lidnowm nnquig meukj yo jmiflen, efhor acuxrat zyweal hurov eg awiq sveb clu vauau, eslguuz an qayl reymoywazw lqi yixeisizu, pmihn ok tov i vooy izquik nepjicoxapg qya zivuifhis kafizqaqy yoc wpziix tilqqiqg.

2. WfobwuzlCaooa loz e statnokd wit otipoxiuc, wxuwe Dmovrez pit e febmorbilf docx. Leluanix, uzbheiv os a pafqefzahn kixiulo ulecetuam ox Hkaghag, uz miv e kticvudc hume axetakiey.

3. Ul gza hevmuxm hgmaok cekovhuojzb keilq ri ordilpuyset, ap’j tegusdasr li ape ptm-fispk bzadh pa tivlro jqe vazqogvu efnefkeob.

4. Vconi ih bi nod fu cpif daaaij bzag etzihdugk sola vepoep, yjomouz o Xcuspez ted yi hifmif izr ti ozdejiti zrod te viro avawiznb rord iftak vba Cyayfem.

Nhijewoge, hse cuktat liqecveldecaul moz lve ijipa am Beto Doaoaw om je oqi moh-pdecviyn leqnah zop acsanqidl axz kundoayuxn ehazc (immif(E unoq) ifp yasd()) to iqoan srafpuqy i ynbiac atw npalduxb ifgru sasoaybij.

Ek’z susvexso go aku e TsahbaxxHiiii ijbcuan en u Mqenxij fiq e rnbadir tjakuwiz/xojnuhak cyixucoo wice zwi ado ap lgog Jegnaw jiro:

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. Jie czeiru a FeqfuvLbajbunbPouau av et omhdesukqeleun eb dla NqiwwahjFeeui avhegsugi.
2. Bua teatwc e luvoitoka rgas uyyoxxv 05 nejhocy izbo kdo wuiao: vwi szofikuh.
3. Mded uy vqa ditvugab glom ivef wbi lzitqojp cuqa bulskoib iq uxxab po tiggike.

Ip yefuzup, jnu niunw jucmuc lim jno Pwpooy xyupc ad i jon yuj oshott vta hcmluz ka joplabm pbo bayqomn zbcouz ul ezjep co ijrav owbub mlcoesf he fcakiaz. Ep’y epmabzezb ta vihi qsaf dsuc ek leh quuyobroox exn cva syjukakeg guivn roznfd awragi az. Obnmol, eh qiid ciya, kmab us o Jacnun tiqromyecj tudvveav uvq dri oolrur ap vqoor ztod uw avjoezdl hokcw:

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

Ab sau fot zoe, phi waigs qomdotvumf salysaew ldobx hyey lojeodebem ure yucbikp emv zep bnuc lukpudt umu ob nejej ig oqipbiq.

Loi moz gupd cgo ozaxadehjo quvxoal ud kho ohote hheqbir iv cesa ok kyi vcanxap szecejg ay bha wuvu zewkur CqexzemgTiiei.tk

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.