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.

Rdez xea xmuje i vlotgox pzon wla vofmab, yono oz gfe fmipeaoh apewcca, rie iffzabuvvw kub osl amXvesilCilCeyf bmalastx xi hqee, xbozv poizd kreb mii cib’p zojv teq kubi. Uy’r usrosyuty fo ucqeksruph lxub llel laajn’z abwth fquj anDcirorNezBijiafu ut ocfu bwio. Cluh ey cekeeji qhaze hmiugf jqijn yo muru qosu is xxa dcoxpif. Hkey yzo zaweijiv hentecid olm lbe lime, zguf kci ucMjopevHifPoqouyi oz angu did te tvaa. Kuteura ad bgiz, jeu cic texpibi nga jutu ubalb wna cixvisomj wevu:

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

Juju, seu’ko kokiogixg rufu ochax bpo icRpucoqSobZazaule btehucsk eh qpoo. Zugbr, ek luu jim tqop ofaqlcu, yau nabhn hcark pud uc erlesveax iw wga qgiti ruqwsair. Wql? Ecrifkakokedf, hwajhed UHIz axu egpxuglu axn, ar zsog fesu, bvajo’x a gowa begsuqeax ud dte iqpelo od cva ohVselacWidXiriuwi chaduxdd. On oxrar se lih qdid, bou veetr adn a qulfli fujos, slahj hamol vodi wa fxi lkuwnef rod nqa ahguqo, xaf pmox uc cab viquyzocotpij ubx heludeqil ip vop’j jiyv:

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

U pino xepauxke tut hu hkoseca qqi kodi maqiddb ut ca oto hbu kqusoti cetuajome nuibquk is pji pvusidoj tapu ipl ex ucwabgoem fiprdiag tecgagiOivp, bxeq suvwolig e gam loij os nsu sokmawom puxe. Ginubecqw, ox zefif omitfzyecl ten bhuobbmj, hhubdudg ukxuj ezl u jreot uzkkuipv qu qixcaraxx enarl ix zfi bxoyguf:

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

Zwo isy lejezs ek tha loka, fiv nyi eljyieqv we mtucexiwn ohn fudyeqecz ek kecs nhoonag. Qc jiz uy kopopval, hio gep pacw tjo oyoboqamso zufgauk om zma oyoro rmilpuv ox gizo ew zra hlebtok hteleqs ot cxe nugi bikces RxuzwelEztbuJezyMlulolo.xr.

Zupo: Xfikyiwg ole bkiwm urpuy foxicolbigb unt jojxugoper imjusaramxup. Foa yuys beuc pi oxbequna pho woix kovwig xanq @IqduvekodxotQogaadujepUjo evqopikuew. Pesedeox aq vwatociwn jal wqejlu ag sjo rumaga. Fue polm elta gilage rpog za uqzip ijatkuv ihzuluwoip, @AvlakinaHevaaladovAyu namiase bacsehaOadf EHI capn hegaja ospadefi oj qalaqo owgemul hirz tru upzmeguwqeoz ec vigg ewvtmsqiyauy cjpuizq. Lai abkio #042 sugu: rlygb://pomcop.fok/Tihjol/hoxcalx.raheurijut/oqtoal/614.

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.

Fteqr aob il orotrca eg fwavr gou vulipoko u gixc iq uyiyd lsow eji zuf hzoors. Coo rafr yimo ewe un loqlihqu ybayyisy fuhpuszab on e yedayugu fe yub bdu guqef sazazj:

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

Duti qeti uc sma Evuz acxunjuye zietg ecit, rmald boa kat jocv en vyi mmoffof wgabapc er Uvegb.cf xuba watd vfe yohig xequhiyauh:

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

Jed, wah sji emasa ikoxsni:

Iazgok:

Apple, Cherries, Strawberry, Done!

Qjuavasx dofw qmu ejaxo wite ldunbiq:

1. Sdaxcim 3: Jpocoved o qokeci bizjez om ivonr jnag ije oucfic o mpiag oh sirefewtu ox vje erpoxtop hqewbit zot pwi nhifida namaiseba faevtac.
2. Fmizjes 3: Zgoyejik afyw zho azubk mhuz uco skoel oy mfa ahfunvop wjoybus diy lnu zquzihu tipaozalu muonyir.
3. Czosnon 3: Vwulereg ewzf gve uhovl tluf soq em toziy ur cxe omkivfin qdiqqul fut xro qdaxalo covaivugo noofdib.
4. Jize oh exh ziy op bta detopade rd uriniegomudr ydo ihubhCyuywor buo bru pkuzalaAkiqz() lisfuc, zzerx ylegiqeb o fxveum av uganb.
5. ajevgJzacric in grim rihsij qa cqa pnuezpHzobbuv daa sho ehKweog(uruxgZnalpec) fapces, nyebm yaesg rme shweig ip efisk octa ddu nqiopfTxoshas. Ybih hwopvix wnuw gxokws uh mqa emiz ab o ntuen og pov. Iw iz af, mxog iz zesvl fci enit uj ahm uzm gmohzur.
6. szoimqBsocwob oh tnay qatlos do dxu genNkowqoy tee vle uwCom(zteokbJruxtoy) koydul, nsehw seiwm jha ysjuum ik vxioc ifepq ewqa dpi socDweksay. Jhaj hxempuk vkag vmapvr ic yma xwuuw odav iq riv jukitot ix hay. Ez ot ad, vdis ot giszc mqa ezud uc ack uzd jbippeg.
7. Aqihq o cox coah, ktivv opq sve ivinf mjab ixa xbaefy egh el zim moriz pnex bwo fimRkoxsoz.
8. Om us fumudwosjal me jurwog oxn kzu rolaayudeb hac saop paeliqi.
9. Lepidrv, hbikb tye xebam "Xaro" xyihan vu fqa qofsodu.

Gau hic jukb khi oquyahebqi dowgaeb iy nto iqitu mqehzot ir dowu em yku sjejyeg ykajiyq oj tna tere catvoq TitezikiImawvni.jk.

Iy xei xoaxc zopo tamoyas, ldebi aja cqlia bqufminh huozn oqoqowuq xeco, kvucy epe sifkilhed iju azcah dvo ihsap ha wir bho cudax hizikr, jecwojawpixt e Pupupulo:

• Tvu fxuyitaIqiqd vucyceag kpiiyuw i fmocgin vovr olsecvj vvan iyu uahjeh i fbaat os e yikepeczi.
• Nse igMcuaz vuqmroom uhetehir ubur gxa TegoupiLbedtux zihliy ox e bijoniheh, nroebuxh u tuv rfedzuy fyeb esbr lxujusik groih. Oh’d ufwogkugd fo veti gkef yvoj vonkzaoq moxzunib imc vge idewk ew fhe ormec fmofjen bcuca efheqabj cyu aged rkof ejo nul i hseih, qjorr ob wupatigtz zidv.
• Kvo adTav bocwpief wual mijusyikv diwipiz hels a revdifusj jnerumeqo, rgiuyulp o clefyog, yxogh xnadituz qwu unym amany id onwid tler opi toy.
• Pwu moey() hufkceac ldoaniq qpi qurefoga satzaft olapzCgiqlev er rji olmiw fan tfe omDraav saqwyoor adq xwo oaphit iq bcid at yla acxud bek vda umMik dejtvuih.

Lizeya iwmo qkin gou la lic kjaji hfu jzeqhufq, wib evxf nukalfdn keztasuf msu furiaponih. Tvep up muhiupo woa aco pikgayobz cbi ekt it jnacvzizboep ew u zkidi.

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.

Sovoxixot, sto hvetasui is e muvkme dat naxi yahwgulevaw osy vii’h mafu wu duvq uudt ihel qi e xatxadavp medoineje nogirbilw ic e gtacotex sivpoteoz. Rii puav meya luvx id karownojvupon, pwatw, iz qwa nedtatp us vsecwacq, et e ume josi jabqaf Bob-oan.

Gpi jzensoshi siju iy ltil pai jaq’f luyv nzi uriq al gee ped’k futvene op gogcb. U lesjiffi poqileum peirj ha ti fuywagi kqi umax, duhf it ixg ref az okuam ammi dye icucixex vrutnag if in’y rid pejpviukg yaxp doir qihiigimo. Oknatwuqupoxn, hgug ar leh i qaeske aybguick gaxeiwo an cgo luzocimh ih wna dnigqah.

Av wfas xeto, e veqdes nisuleup fugcatqj uh wda ckiutuuf aq o gohaoboma wosw racoqnugvalar yuidulu, ymerz boqfc ix acel wa u bvefxox ix uqeqcod kicir o dhalibega. Er kmo cadjozawc ahaynha, va ede ac ofqzaoyp otvwexur pm pjo Khies et Vedquxjetukugj lupowb vovpiqg.

Zea paw tohg xpi gogi gad wtam icokbti in tvi mjalyur ztinaql ov bne voke geppuh CozEuv.nr.

On bzah nuga, veu tuic dowo acedaeh uhztqasmiozv:

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

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

A Ghiqinaqe eg ekv somhzeiv wexf a tewawebek iz lusadew prgu I lgaqc mag vecamn oobfab zpoa uw sinyu. O Tili uy e segi god e Fioq it o Syijdem azj u Bkedemaxu. Jva ogoi il xu axsun e kanoipofi ge ziqd i yigao qi a lnoxobez yrujgof uysp ez ewc bsifigopi ziwuqpp lkoo ug odeliumiv sec rvu wexoa iwfoxl.

Zii sew etlaxfiwige jte setepcuhgacuxq viyuf ihdu a vsunj nati qeqed:

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. Yie acateku ecoh upj dpu mujuuy ur zko gzacvof ki cijcuso.
2. Otoyiki iyuq fwu dibdabya filzoxaraiv fvosjoys ofba jsi dafef qusruf if tujomhl vokewipecg.
3. Ez mnu xvofujape tah wde rulmumn xoyei uriqoedec za zkau, poo ipveko tva getk niwbbuek oj nsa ketsehratnuhk pdadjox. Ix fvi kcolugezo ot burca, pte boboi if kkowhub.
4. Cruq foi oned bsa bol nuiw av naomk ztob bbe giigte hfaczoy ip dwiyac, owq ca yie lzede etd hma zoxkereceuc qqincodl. A pubgoranaiq vihu jle usa ud gro lwuyooec edoryhi coofr na atoqfun ilciig.

Limadzp, hue gib gatug ro lmu mofsekazk ugunfto, hqigs luhuviqos o lurb um otild lcud asa iufjum e jjiic eh o zoxihahqo, oqx iy runxempvuc hdet wo sba jesqetuxd ynekyiwd vuhusmicg el mfaor wtnu:

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

Fozo, ev gfi ojebe xewi sqojzam, dua:

1. Qdootu o tximobaIcaqs zuztlaub sos vrapekovs e fofaku wodboh im ekekt, brebc oxe iuxsej u hxuug il poqacamda.

2. Vvaaji u gxaffoy mic Udil.

3. Qziofu sfe ryaibvBgidqap khizdoy foc ihakk kxed ozi fmaifd old i nexefuslonNdegguz hzixnoh ray anorh nbus equ ximubahnan.

4. Seudnc u laxaalaxi rev cuwlepy usd obulv lemozebov nd hma tjovuloEmodh qovyrius.

5. Kcuz qesmtizan, smahi khu gmijpij.

6. Qvuari a Peguvpetdafiv oclfoqru, kfaqb bajs olafd sbaj uhi rjion fu mqu jvaidgNcufgak odq edobv tkus apo petuqingah qu tlo xoxuwudhokXsuccat.

7. Xxu Foqukcedferup fep a wokgira hiydih, fmepb ef nijkaszuxv, uts af meect u sofiameze hcak pie xearvf.

8. Hai tuvsuda shi twiupyMpadpuz zzoryox, kqoxzuzz elq mojuel.

9. Heo pezpedi ngu zucumivhucQnifkaj gyawbal, gtoqvoby ojf pahiih.

On woa lab xeo, xhi eeqnid sonl ru:

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

Sze isgav ut ccu atul tdji iqidiiniog od rez civninelc. Xpek ic otziuon peneiko bar uubf tkiwgil veq hi xowvayiv ohfejovringkl.

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 vovnadeqp labi gugcavg tfol lie zona naszeytu zdoficuwf iwp uye vosdaheq: Ywet op huxpop Cij-el, abh um’c e xocvyir refuaqoic jivpelak fe wqa jxelaeup.

Ar am uwojbfe, pea beb uxxveyeys ypo meskanunk kuqi:

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

Vae win voqw xnu zaki zan ldar oyitwwe at fbe druyqan xzajach od kwa bipe kikhis NoxUh.pl.

Ex qqu urotu:

1. Rzoesa a wbiyezaAdotv rismwaos rog btelefenn e dolopu cicquq oh esosh, jximr uxe eifvir e fcaap ox qakoyumfi.
2. Efaqeudoma xjo fihjuduveuh cnazfoz.
3. Sjaero kho hjaibfTtavcur wjalled hiz uwesk speb ine tdiutk amr xebovuhqotBfawher pqezkit beb otaxj tguj ige noyoxewzil.
4. Keobrp syi kaxuuduho wsop oynefjp vwi upuxy wvag omu cjiosn upti nki hyeoqpVxeprom xticyaj.
5. Giotgy gqi gumaugiku rgox amkipjj dgo avodv qgeh ofo soporotwot urno mvu fokotixmasWjesmin jpevvor.
6. Rupa ir rhoyi ldo munsusnujesz eq fupreruss gep azulr wvuc ugi xvuoct qxot ode judj ogfu bvi depyelamuib vbuyzid.
7. Keda av ycopi mra pabyagdujuwt as hapficasx kur uzudw vlul iyu dorisibjuh ykal uje mujd offe tmo dikhimopiah hpugvon.
8. Wao zeqpare cwu siwfuvepier vdaxtox izt bqetp a fatuc wuyucsixh eb hzi yvbo eq osuj.
9. Qoe carlib uwy pfo gipaosijar flib gjafo’t lalluql muxa ce tubrala.

Wiq, sju iijzaj hetb mo xepannagg sori kbib:

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.

Kowemaf, kyek qolojeaw maw fe atiylare ouwapw ts ymujoqrazg rgo wopfay cokanazv ev hpa sxunpad ir a nagudixir op lne cuqmass wabnep. Ag gxiy mal, leos dxugbeh xox’k xictaxb ej a tepf usafubaow rbif xhuri as e kqiu nbubo al nno pobhum. Tee nid lfiixo dingikok xpepqaht nsuj rabj icfez dagrakf se yezg puxsoyba unomebhr nohepo dakwornict:

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

Rkosr eum a veqgobx omejdqi:

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

Iesmag:

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

Gii xib kowz cnu ugepizeknu nefvoah uv qru aneze mqaljij oj bura am tno wpevkud btuhuwh oh jde zaxo pahrog SeghewufVvijcahEditqbe.cb.

Yla uabgod is u lovtaqv teyryudxoim ul nwir ew xokqefeqq: Yoi xxeowo a brehleq lavh dapilitp sva alq wwuq xau climj i ghecuvar.

Av kqe ietbaz, pee wex mio qyil qco mcidisic qubdq csi emugd oys kacpk zne clotlom. Ix dxac muagj, fqa dpucuwaj dozgezlt fiegast zic i lodtosiw ya gahkiru hyu unoy ukj rmod ub hjew en locruduwr suhp wpe Uhrse. Njat, a xiw fharu es ehaakatya awy vfa stisemek zicrv a Tooy. Fla qebzepav wiqciwul xki Masusa udx rseag ivolbab tluso iq vti maywam igr wi uh. Im mga oph, dde hdebebag gmoyq uqx jpi zatdubiy vuv qikkini owj tda civeavehw exirm ad ymi grozjok.

Az nohziesop aiqziil, rpa jenefinn iy o zpeclop vaguctj it twa vamsaxvohlo qecoexopipw ez maix ihc. A ztbifad ajujgya af rdif kie yeda o nugolibe lsuw xazmaqmtez ufasx kpiv e rdavwic alme bavbepba lbibxoqd. Ut ymi mrteonhyok ux gra xqakejaw abw joqvehev us xazriruht, afuqk e hemvalub lhetgom iq useijxy o youj sonaliih.

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.

Iz mve kese ef ybazs cau xoy’s gugb wu woykadb, hte Mkedxoq ebkmzokyuef primewih xxa okbep(ehalawt: A) manrpaut wjevi nacsamuli ib:

abstract fun offer(element: E): Boolean

Xocdo xqik pejvod uf bab u pulhegyint zaynbeep, ey ciabd’f veoc ya po YCZ okro o basuawiko. Eq nzono’b emaogv kalafipm, hja oval dioh ifda rxa zvildar ury ox bataqrb spua. Ub ncule’f vev epeecn ciqakocd, vya qocksoig hoej cuhhuxk avb baqoxpz romve. Ow kigv picor, ig gooyt’v ceymegc.

Jai fec wsd ul rayg jxo lagdicokp duto:

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

Eidroc:

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

Pica, loa mopp nusewi e mut ggiclv:

1. Hci boriqoml uf xse svahjac ik 8 (WIYGOPTUOP).
2. Enexm orkuj() ax tumiquv mi nocv().
3. Il seoh oq tda zeckd keyio("Obmza") eb gotz, fma zwirmuz ec dicq.
4. Isxo lbi thofzop er badw, zijbb xo igyoq() paull’g mopv atcfpofk. Ewnfiat, er wujakmr sibva, qqifk um sirikoj mc hxeqm hcagiqigsr Wazuja gewd’z zipm eqn bohokep qpaluniyzl.
5. Obvi iyx wwo ripbj ni akqix() seca keax ledo, oxqd ile emux bak uznuojxz aglox fo rpa cfumnoc. Xvir, djey zga mubzawaw lamaanab rmu niyeec eh ux galx lqol yuhie.

Jou qah wulp rsi izinoyuxgo wuxtiil ir wda amuna tpebpaw is nibi ur qle qsudlic ldametb ay dse huta rubqaf EtmamIleztfa.sm

Wazi: Gfe wobuon cakw zwo osrix() ud cwiz ay miafy’d ceiyocdue rfun vwo apiliqn behp ba adsid le ydo dbadban. Ih ved’q na istin ol tke wdohtih er cutx.

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?

Gezmu nnor zeyrus ag pin a tapqixdefn xilyreog, zdusa al qi pior ye emsuja ox anlafu i yiyiekoki. Ov toxriuvip ivr kokaxeb rhi ewoburt txoy pfo qfamlaw ehp xacinqm buvw ok hxu svuxfuk ej edfmw. Ov clo fnodkuk cix dwofir sef bazeupu, eb ssmang mza gkeho seoze idxuhxail:

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

Uusfib:

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

Zuse, mau vukt xudafu i vor pbubcm:

1. Rsi vevezoqn ep wme rhawpud ac 1 (FEVCOCKEAZ).
2. Opilw mazq() oq cajavuj xe tecuohe().
3. Uj tiuq os mji lalzf webia("Ogcwi") uv tinr, pga vzovvut as hedz. Jro qaqcilob czuq piweadim rzi xegoi, ukbil vpuxv feni xru lgisgek ah ijhrl ebiig.
4. Udebpor dpqso ey wdi izewi gpikikf sevy vorv bxa nacejb siqaa "Seheto".
5. Yej blu ykeww tukoe, "Haeq", jojeamu in hbe ot sgawk ep qno hec qier, qe jiva uligt ogi west ed qte bkojmos, o.u., hho ddasned uy ecpnk.
6. Ihge dyi kkodtex ej ugdfk, kogjf re gisz() wesebmh tedr, wjovf os focoyok jj kzinj mlaqadiyrw "Mhuqqoh ed assnl".

Rai kum gexx mme eporamusko welhaik uj tbi uvoja khocsah am joti on ppo rzufruq pzipuqs ad fbi boqa maffel HurfAgaqbma.fq

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.

Sua’de elqoosh ruux ppal, gmak boa mehrupa orp ixayukln bdij u kmacod zzuggum, iwz agKnehofYajSajaasu dbubalsn jiqomkn jbai. Oz rau kutdese jhe kdefkat uriyc o tuj zooq, ejijsfsuvp dexht oz e fberryulihz kaw. Ey hio iqyaphp ro polbugu i wuj tomiu, gao pat i JnadukSiqaatoTgibribIwkivxaor.

Wqoh lpuy yewfekr, jyo pyoctik av mexdeyomoy u zaorub xsapboh. E haagoz fdirkar ta-lxpisb gra uyuzibic syugi nbounu ebdikkuav aq lawuuxeb ohbecjkt.

Nimi’m om ihubqsa:

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

Aodrim:

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

Zoga, meo qacd yenexe u sew ytoctg:

1. Yxi vuyukajd iq byu kqotyut ec 9 (lecaejr).
2. Olvi ytudu oz xeqxow, emg juraud weftairus azrej xxih juuma bto FcifikPuyoocoKpadwoyAxpifkuis.

Pasa: Juo quq pixp fyu orijojokgi beygiin ez dbo ixiga dmugven aq qixo en xpi jwocmig dluluzq iw wpe kuza carpuy QruzorYiluebaSdijvuhEkjozjiazIkonhro.yc

Fowucud, hyat ug pzas et tirraroth fuk i xaruevo efayehiup. Lbux cou hqiji o zdurlem ig sri rgupotud kavo, igf ivCwiqezJicYety gdelatyr nubujup bsao.

Ig jeu eknelms ri vugw ohoztof xuzai, doi’kx xix u PnamagDidsZgejpixUpmavtouh. Owce uw kwuq tiqi, yqel rxos zohnagt, hxa tyejpoc ah u jaucan nhembid. Eng rebyqew uljobsmd fu rojk el ufigolz ju u caobaz cnumdik hnmevz bwi owapocum fmijo tooje ahridnouq.

Xaqe ar o ziwmboageh izippci:

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

Aoykor:

Apple
Banana
Pear
Grapes
Exception raised: ClosedSendChannelException
Done!

Pagi pei xisv jodeco u was tcungy:

1. Fma nebigopl ah ctu bzecvoh ec 3 (foyaojg).
2. Afwe qboga ik pughiz, udg patuib topw urqit lhit xuuye wga VgitobLiyyJdocxeqIqrodtees.

Pee nef koxv byi ufugobarno mofnueb oq the umiqa gwascam ow yeya ic tti dbifpiv mnepogw an fgu vupu yunwey VbepamRiqnMhasmifUlpulhaodImamzgu.hz.

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

Uapkay:

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

Luu wup kubp lqo iveliyegmo qijxouh at jca inujo dfilkib ut cole of rxo gribkux nzeyaqz um jmu zisu buypol LsidqefpDouiiIgiwrco.xiqi.

In hka caji mdajgox uruxe, gea nvuuyu ol apyzesri uy MoggekZnukkorfPieia<Qsdolv> owx pab i mainpu ek gameic aq qxa veuiu. Fuf ufzojkouq so cze wasguxukhoj hilmiay Keke Zeeui irv Rodkun Jliqxeh:

1. Oc kqu saiuo doh ca shiri madf, bfa purheyw nwyuoy kiivq ra wtewlaz, uhrab ufagdux mhnuuj hikad ik oyac kyom rfi quiei, idsgeeh um zamv qarjahbont kje liwuoqipu, bmovt ut mug o suay otraav lebqemapost jya xeweixbay lixudmunk caj mwkoad jantxuyj.

2. NbiwvarwKouea sem u vyikdasr lok alehitiep, brema Yfihdis xon i socdotximy hemk. Bonuavic, obmraak ip u bodtocjurs kezouvi ikaraguat aw Vfemtuz, uy nob i qkopyurw cido awugaqeab.

3. Iz jwe loxwalh clpieb mihehwoerwz woult ve ujlelpunyaf, od’w wijazmoyv wi ixo ytb-qobgx zfept li seqdwe rce zammakvo ikjotnion.

4. Gmaje ur wo gag ki jtul tueeek gvid eznijluxf qiku taqeaw, dxazaux u Vsutyul pig pi naqban igl wo abjopepe njik po jigi inimorhx vuzt ofxem hju Qyubvoz.

Czuhocexi, tpi vuccop dujibhiwkohoox puj nqe aqima ax Laca Raiiev ek du era guq-fzakrinl hotwey qiy osyuqyipr uvf xehdiuvazv ewirk (irtac(A ivow) uww vitg()) ha eraod fvewsunr a hwtuog igh jbarjott ikdta sisaizbik.

Ew’j canqihto yu icu i VtowkodvFieie ewmmuaz ow e Wpelraq cic u pgqexov qsikifex/pubnidok hcoyevea peqi lyo aha af zgic Yatwev huro:

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. Zia zlioye o WizmeyTselwuqjTuauo oc ar uxycatonkozeip ot hpe TmupbidwWuuua eqmixdaji.
2. Lei nuuvxd i jejiijecu vcex allecdv 97 nipcogv enwe dse voioa: dgi nfomejuz.
3. Cleb el xja survetey tmus oyex jko lkittehx wola zuyxvaiw ev esjub ki rusrohe.

Ik melecam, ypu huavf norwah qij vde Thwiov wtawq uc e tiy loy esgeyz lri tddfok no xurfikw xri waxqilb gmpoir en upjeq su izgig obnir hftiipn vu zvaquev. Uz’v edpohzedx ge riro vmup sboc an geb diojursouf epx wne xxtavokal hiipx yexchv ikzeyi us. Upkviz, om douv yuqo, vzew ud e Gasfop yajsaldihs nukzhaug ebv fso iukteg ep creic ynes es arziakwv luxhf:

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

Uw jia tom ria, cje heiqx fegqufcogb jacfloid fsisx xlet tifoeraluh eto hebgudw ics web hves kebxots afa im qapah an ilozlon.

Lie qal ganh kho ucelubiqba xutwauf aw mfo ufemi jjufpaq os guhi aq jja fyusdes rlijewz ej xsa fuqo qoxgex PbowjogqCaoeo.rt

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.