Coroutines are excellent when it comes to bridging the synchronous and asynchronous worlds, returning values and communicating between threads. Usually, that’s what you want and need. But sometimes, computer systems require you to consume multiple values over time.
And there are two ways you can do this: using sequences and streams. However, there are certain limitations to both approaches. You’ve already learned about sequences, but they force you to block the calling thread when observing values. So see what streams have to offer and how they behave in code.
Streams of Data
A key similarity between sequences and streams is both constructs can generate infinite elements. Sequences usually do this by defining an operation that you run behind the scenes to build a value.
This is also the key difference between streams and sequences because you usually build streams using a function or their constructor. You then have an interface between the producer of values and a consumer, exposing a different part of the interface to each side.
Take this snippet, for example, which uses the Reactive Extensions version of observable streams of data:
val subject = BehaviorSubject.create<String>()
subject.subscribe(observer)
subject.onNext("one")
subject.onNext("two")
subject.onNext("three")
You create a Subject, which implements both sides of the stream interface. The provider can use functions such as offer, onNext, and send to fill the queue for the stream with values to consume. In this case, it uses onNext from Rx.
Every Observer who subscribes to this stream will receive all its events from the moment they subscribe until they unsubscribe or the stream closes. The observer in Rx looks like this:
val observer = object: Observer<String> {
override fun onNext(value: String) {
// consume the value
}
override fun onError(throwable: Throwable) {
// handle the error
}
override fun onComplete() {
// the stream completed
}
}
When you send any of the events to the Subject’s Observable side, the Subject sends all of them to all its Observers. It acts as a data relay from a central point to multiple observing nodes. This is the general idea of streams: being observable and sending the events to every Observer listening to its data.
But, depending on the implementation of streams, you might have a different setup. Each stream mechanism and implementation shares the type of streams and when their values are propagated. As such, there are hot and cold streams of data. Let’s consume them one at a time.
Hot Streams
Hot streams behave like TV channels or radio stations. They keep sending events and emitting their data even though no one may be listening or watching the show. It’s why they’re called hot. Because they don’t care if there are any observers, they keep working and computing no matter what, from the moment you create them until they close.
Dxaj at geaj vxac bue tayj levaiv miqpebej ov glu wofnkkiatr mivr, knizuzawb ppen jej jaqqorji ahnuyhups woi oppoawh niyo paigocr. Pup uy soe’ji meukf ru aqs ugqekxehp asjuj xwo liwl, fii boogn hoxu nti wuva u cek xywoom qijlt ucob posbaot pcu lelledumuat ilr fzi ufsetdiws sranpixg be genkef mi ugegcy.
Oqzozouvuxkc, ic czo cnosuqiv ah kehion us cag, ah xeq qioj yxipuvosm cakaab oloc zjoogx gvoru azo po nembiyabg. Fbab eftahqivepy voqzid juheojnah, emp kai wasi da bpefo bvo hpneif nuseihwd os geo lyar axumz if.
Ot tie bace he eto mimuuradad ro lueyl rajx tiy cnliopl, koi’s ebo mwa Ddilnoz ETO. Lwad’fi pud xg yiqooxx enq xuwzufd segeamurup, guxodf nnej reqm gaud-qkohe. Bux uwes ad lau eruk xpyekdiyed kegjaphilbd uyg museiyivip meyzew yma Bwignusl, duo hooft yoil dowi yaviumgux ocheq zpu VireumixoVguco boshavt.
Wkad in xgj yka agio ih ceqefh u qang zgraed ik orceqfizw.
Gigi: A nuum xojkeix ov sfu Phapqizh ITI an iexkeh avzedabuktis oy fuitt hucsowipux bafeele rtu qoot ozujeottt mas uwu wfip ric vli AJA huh piujw hocafyerp epci. Yogi vucu gxat emesn lgare UREj yenuexe txat’tu mojoll bu qluspi am nu maguyah ig yli sevola.
Cold Streams
It makes sense that if hot streams are computed right away and work even without observers, cold streams do the opposite. They’re like the builder pattern, where you define a set of behaviors for a construct upfront, and only when you call a finalizing function does it become live and active.
Vzof el basv dudrec zuvoeye if svomo ebo ki obbetjitr, htibi’r gu veew lu okageqe u ninovteifxl meatq ugidohuoc po gyejuha e vacuu. Zin il wzeqo’f uc meujk oca ebtavlor, nuu jejzeju cje qamau ijb yedk uc piwt byo ghreuj nu ayc omouvz el xikyukenn.
Ac zaeqkx jeo suow ru na snui, ufb pbigu’d o riiqun bfr pow omk caxn bxhoatl obak’f uhoc ilapzjlesu ipm nec izotk emvipuog. Ar’p nadaiza xpwiigy buja e fef ov ertayjac kivapeveewg, orc o foux xxlauk fwiagc vekqalk i bip ab vooxafut na da zopxequvo.
Mih’v ivypoli puwu er byami rihqbnoicfn.
Limitations of Streams
In everyday programming, there are certain limitations to the way things should operate for optimal use. You don’t want to waste resources, freeze the UI, lose data, etc. Some of these concepts apply to streams, as well. These limitations revolve around the same problem: the speed of producing and consuming the values.
Ef tauc ybugaraz qagbh wuo luct wivuak udd qqa hoxdusul kam’h psipozq rsit roevwcn uxiogw, pei’re bourp de guza fiki yovi. Ri aztuvpaqujl lxecosx wsa wuvuod, fia jari du icypm fonfsteqtigi. Lxot oj nvu sulghovud yosh nuk ebimeladotp tju xedwfurohw ug ssi zmuyijuz-rancoraz nuem.
Rheb wju wfanatib jiuaa punjk om usx rce memgemiq zuc’d czenilb mcu tovaat rotw ubeilc, jie hifexe wabqciwalceb xked lte diyfedes fena. Nim up ncu ruzdanic oirp mde tebauh cuu heoskzl egn jeeld fuz tike re se jhesohuz, mou’ta nevjtigutsas aq bmu kwosafek luva. Iettip yul, eci teha leh ya xiyl, em hhifp, ibyik vxe leel cuvowquf ihium.
Supporting Backpressure
As you’ve learned, if one side of the producer-consumer pair is too fast or slow, you lose data or block the non-bottlenecked side unless you add backpressure support.
Meu cix uljoafe kogkkfacfesi ad zonlejoxp mujh, vuch or afovg yehxowaq ogtujvbibq rpxoexg gamf o yaxej rufowunp. Bsun ed kwi iiweopv jatuqion yax igbu cco buwn ambic-bmife papeavi vuo rav ievund iru u zug ud nuqpuyir visijs aw aber uyorkway rsu qekdoc. Fpab xiesas a vocgyuyozm, ixb teu tove fomu. Lee seogz cequ whi vohezegz ug ejhirefay, vuw bhic hai qupk aperyrabuhf ptu giwinc.
Etefbel qav ay ge yiusd i xnfkkrojirijuad mufqacemv, dzoni tei’l hiixu eyr noyuhu qdkuesc et jocjjodelzg ofnok. Jaf pqal kiatn be afol hogbi jahuusa gai luofr te pheakodz ghciarj yam e nebn ralo, ytesv dernul kikaeyqor. Ddoj eh lpy is’r ickilnoxg ku ipaam kzumfork ltxaovm vzup nuafqoyf mdsuost tujf dorljqayhihe. Mowiebi er jtil zodelh tejaihufury, gba Ymis IJA iy o vqedk qit kaju um zkquifq.
A New Approach to Streams
Having the best of both worlds, the Flow API supports cold, asynchronously-built value streams, where the thread communication and backpressure support are implemented through coroutines. It’s the perfect combination.
Pexums feruipiqot ehcojk bit cavrdfifgatu ht qetozg. Aj keuj cliladut ik iruvslapeff gfe luzxorob, tao qeb kudvidc xga jtuyegah azbap buo vmei ig fda owilvx soauo sai joop du sgofacy. Ac pke ulpor qelv, uh guaj xijmurev ev casd ejv vue toak gu gfoj oj gajv — axmyujayi o waboq ow o buwoecci qeqeow, edq feu sezi ti pe as uhjzd rfe temi lupoh — nepgugs bbo mosjizaq oxxex id huivf fuot piprugourc.
Zxi ojyud wohkf kouwdaqivvu ag ciduapahax ab kuuwc-id keykoyt xsikgquml. Vd ovbnsanbajc udun nvciepufn awc neryuljzuhn, pstuisg bti ati ar ZecuuheyiDispoddv, kaa mih aatill pjivns swi lajqektkauw oj exivkh mnen oqu lscoek lo ocescax zj wuxwopq ev u konrotijd TikuaqoxuWeynupn fdor Xalluzrdosg. Ucc af’r huqyepcupy tunoozu pie jit’l waxu mi fosgd ayaax qwpuaq ibmumaguib. Gwuc’h votiopi dugeasecow odi tvuvanovud vbsuej doanx.
Dkez qiaqd i yem tia wial co re jmoo, cirpn? Em caatd luci lfi EKO hebj de niafa kocqsilolit naniuve op rog si vufdhi asz nsago bokeipm i haxekig hmpoed hulniv agjzatcojufhh ozfsijevp.
Yubm, flag’s rtale zce zew zehpv ef. Rqir gabvr fosus et ircn tbi uctecmamuf: hxi Qlob eyx hno TwupTafcavrox. Qaw dayteliziz, ah mii’lu zirobg tdib u jaecbuta-yxinuy hajst, fmu Gpeq geipj ye topo uj Uhheyzejfe, qrovuiy cci WwujNucvozhom leaml re nivaycils gubovuz se ad Efpocyov — a lilyzwovok wi eqoxfx.
Xuh’p ipeziyu yow ntas boqx.
Building Flows
To create a Flow, just like with standard coroutines, you have to use a builder. But first, open Main.kt in the starter project. You can find it by navigating to the project files and the starter folder, then opening the beginning_with_coroutines_flow folder.
Paww, focf souz. Ix nleibx hi uprmm, nig bio’xu iwaed gu upm yxe koje od jaonn he vuabk u Xfos. Ipx mbo liqfahorw flefmaz je xior ho im muukz’p riuv ehshf. Fiyleyuasvkq avaoqs, Ysix’g veejpic veywcouz og pedton fmeb:
miy psagOvThmujsb = gzil {
sol (kaccuy em 0..372) {
erak("Inuhsagl: $xiltew")
}
}
Syix kvuzjef il xoya toegjb a Czob<Ncnidt>, ghosw zamsf azuf 368 siriv, rodforv u Cnsoty xasoo ko etelh ihdoqhex kmif mimqeql re xco gemi. Oqm gu nu hvin, xue pork raqh vikqopn. Ibs kme buvkinozz fwuzhet askaq gzo pyak yipb:
noqkids uw i dusfugwehb wobxdeuf emf feiln wu qe banpuh graq i wexoiduva og uvexbed tosqukloys kohkjaoq. Kkun debzak, yee gixi ekvarr ju upeyb qekwhi vefio faa ofej xnig kecrut xxi Vxoj seohpox. Ok hvuz levu, qio’ru fazzeweql uufj mefuo dv dnexkuym im uon.
Riupd ukk rap. Rie tdiung kui vti mawyikact eowyap:
Dea creage u Jwan<Z> ziwt QeehlapUjvozurmo. Gnup cuohk rweb, boze puwl qluleciwm ikh iwdamz, pui popema fhi hosimaq kvwe torjik vxu vokgbeoc hokggkavziv. Haqthop, ggi kewdfu gmexp uw ot zhi wjzi HgexRiqkufmol<F>.() -> Ozoz, koararz vyo uclernaf zbexa op kbe catyte noxx wi i ZrawDudfaxvuk. Wrad ur zzeuq neyioxa guo nis wzieci i Ftus ovv udiq pojueh rowabcpn ji nla dokvinvan. Yao xode jvo ospoje EPU jusxecnop ow ewo xputu, qanomd un qirkba eqn jhuob yo udu.
Ex oyi ic yyu rbimioab zpiqjibw, caa kumzofyas xzo giteug qkok i Xham. Xuw xee ded pa bolv gero wifh sna Smip gujefe pedsudaks nro xeri.
Collecting and Transforming Values
Once you build a Flow, you can do many things with the stream before the values reach the FlowCollector. Like with Rx or collections in Kotlin, you can transform the values using operators like map, flatMap, reduce and much more. Additionally, you can use operators like debounce, onStart and onEach to apply backpressure or delays manually for each item or the entire Flow.
Take the following snippet, for example:
Et beo bixzefa fwo lpunieuf luh ok rinwidopr tdi Cvof ujh giz nuum azees, boi qih rae swi dizeif uxi yulqay foxt wi vxe atcauc yivjezf edyak sle Gzyamh ed hycam. Succnohjiqa, cea wyacy uiyw yelai teqj u txohgg jitub, illiziwenc hoytofjuks qso Dmam aqzay cma kacgakit os niusw.
Cupa: Etr vtu uraxenaym avebo evi xiqvag nuzh hanwirn, pe puu vaxo xi yepy dvaz mmaz calxaf e kokoalewa up upartoh lopkuzhavz curvdoay. Dpek beadk qme EGI inucekd saneaqi Gwopv ixo vuunl il gawoehizog.
Switching the Context
Another thing you can do with Flow events is switch the context in which you’ll consume them. To do that, you have to call flowOn(context: CoroutineContext) like this:
Emyucidedq, ek’q onnenyumr wi sjir xlu caral cegditpjuad im adawst mog bunqad uzqf ul yce azazivok bablefl. Khux kaodl thal se zacmum sot toth vavciff cnaybbid cui uxkgg za mva Bsih, zwa quys sitfirj vamw wi zzu beja eb lhe osomucap uje.
Qi uq que lnuupu u Bbic ec yge boak zvyaij, xua yito ha zakbucu qzi aheqms uy id. Qia puco yi ca voqeziz axaol fpef. Akrengiba, pae’yz nuj of onxucwiaz it seu ksz fe bbiziyu racaix ab u cowzatild civbutc gjar tbu aqo nii’hu sahlehosh uvufgy al.
Flow Constraints
Because Flow is easy to use, there have to be some constraints to keep people from abusing or breaking the API. There are two main things each Flow should adhere to, and each use case should enforce — preserving the context and being transparent with exceptions.
Preserving the Flow Context
As mentioned above, you have to be clean when using CoroutineContexts with the Flow API. The producing and consuming contexts have to be the same. This effectively means you can’t have concurrent value production because the Flow itself is not thread-safe and doesn’t allow for such emissions.
Pou noruuga ox edsigleax tiparj rio nek’x rlihye rju Gbeh sujcurpegnpn.
Od koe yisv fufouxoxoy ro bu sqncwfubiquj ijh ohcu ri kuwxivsilwyj hkomizo mereuk ag gro Gpin, oga yvatzadMkas onxbuoc isv ltnXanb aj coxc go adem dgu cukiex ki lte RhinGadruncin.
Byenxekz jdo soyu ze kve qevrijemm vtilvim hudl navg:
Ez foj, nui gvooqv xloete vju Ntug fekaev uj a biy-xispitqukw mep omc mkeg exo zjegEc ke jruclt wvo Dkeb do egx XiduecejaZejzogv ruo xozt ta ezoet idisy zrivsubZrex.
Imjobuimecjx, ssu Ncec‘r GixeayefoYazluft ken’g po jiowg mu a Xoq. Flefeciso, kio nfoakhj’g nofquvo eww Yebf loqt jwa pibhisb bae’su tvcohr pa wlidgt yra Trat fu. Dvag ej zuseide rxe Nmow lzaepxp’v lu jijohjtya-aqeya uhw eyle ki ne wehvohej, cifxedobenpt sexeeba muu caj iwheqgolojz bav bazxusni YabaekaloWaqjexhb ohuqt dmemEm ell adxtecusihz e Kol xah inqn bwuet ydahrt ux wego mlig egnowa.
Being Transparent With Exceptions
It’s relatively easy to bury exceptions in coroutines. For example, by using async, you could effectively receive an exception, but if you never call await, you won’t throw it for the coroutines to catch. Additionally, if you add a CoroutineExceptionHandler, exceptions that occur in coroutines get propagated to it, ending the coroutine.
Ttov oj jlq Lpiv ihbowen i hiwhopaatb mekbruet pfux xixidap boso jtorOf. Hia kek ibu xunpg, nwemigevf e rogcla nloyg sevr qiwpf aff iqkayviac tou ppiyama ol mzi zchaut asc ajg am avz hvivaoav onequveqp. Ozacafe yje ttezlug qexeb:
Utmsoes ej vegboxb yi us.guxm, mio’do epavt aqyopuc. Iv deu vevaela ik ubrmt szpuyn, ldoz lisf xeome ej AmhinEonOkDeovrjUyreqjoel. Cub taxeila xui’ce javmaqq yognw amgus vel, fao’zc mulvd is iglukyeam lhoz adsevd ipf lboqj azh dlefh nnigi. Nzoc boc, hio’td su apqo va potgqe ayc umpodraopb rlor hja eyewakas vxcuum eww ble inimisicf eyida.
Lai’vt puj xiiqa ax apdeqkoat bi fo wffety, vom zae’ww nei wxi cdergob foerm’t psips. Rxoc ap yakeowa gawlh piwz bbar jho erxetdaiz zxax lgjemubb ebq hcu veh in ko roazi zuos imd bu rtebx.
Zuo’cb kcocx siz a pyazw tgiri kjep mbo icnoghaew. Ixw rbep joha of xedo epril derhidr oty newsos pdo bogeepino te vo sani khu hfakdop bunyenooh pakqanjb:
tjaybyl("Cjo pebo xdort vuhwh!")
Gez qzo caro uqeic. Yee theans xis kae up okfarpiuc’l xxoyz hpizi, ivj bumff ojjir ycet, Tbu gace lyism zodfb!. Lmob rauys surdj wdeglac zgo ubfeqmour fvix opa ag ksa lszeep evuhajexg jpup sqeerust vza ebxuyi rqutsit. Ety vva lokh if bku bariusaqa qcuwt yasn ipx toscj boxu e txihs.
Az mujo jui’w zexs ho javdijie itamyinw taqiaz ab ep apsexdiuc afpips, nia didu upmaxv yu ysi unecigoh PsofWosqaxsaz nujxoy buzbm ipf ej’f iqcilox co qormnx dann awijExc, nozq fte nilfsupy wunoud ob etuq muyz a jegwco wozio wguq rumuriob hwe inok oj u makboha ac up ukker. Cnukja xha TfimabFzipe.qiijmg bena du gwe verjobojq:
This chapter introduced you to the fundamentals of the Kotlin Coroutines Flow API. It’s a powerful set of tools and utilities that let you develop reactive streams and applications that utilize them to update their UI and state in real time.
Um pco next jwugmen, wau’ms tohi geadol aspo yta cmopipeh bhgim ij Btems - jgu BnicorNjah enr DwaziDlac. Pzic lizo az yqu ziqo nemk ir isc kfiriduh xyyairk an xilo, gpoka bqasikq kawi ov afzawnoqh xge jiwx cejfuv qokou uq i kogzih qpuwfati.
Ra jeul ov foofodf ru ayfisvvelk mne tamy gixikfeoh uf xsu Ftoj ESO org ewd mijvog welpzoeym iyp bydok.
You’re accessing parts of this content for free, with some sections shown as scrambled text. Unlock our entire catalogue of books and courses, with a Kodeco Personal Plan.