/* Ċ Ġ* ĠCopyright Ġ 2 0 2 0 ĠgRPC Ġauthors . Ċ Ġ* Ċ Ġ* ĠLicensed Ġunder Ġthe ĠApache ĠLicense , ĠVersion Ġ 2 . 0 Ġ( the Ġ" License "); Ċ Ġ* Ġyou Ġmay Ġnot Ġuse Ġthis Ġfile Ġexcept Ġin Ġcompliance Ġwith Ġthe ĠLicense . Ċ Ġ* ĠYou Ġmay Ġobtain Ġa Ġcopy Ġof Ġthe ĠLicense Ġat Ċ Ġ* Ċ Ġ* Ġhttp :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ Ġ* Ċ Ġ* ĠUnless Ġrequired Ġby Ġapplicable Ġlaw Ġor Ġagreed Ġto Ġin Ġwriting , Ġsoftware Ċ Ġ* Ġdistributed Ġunder Ġthe ĠLicense Ġis Ġdistributed Ġon Ġan Ġ" AS ĠIS " ĠBASIS , Ċ Ġ* ĠWITHOUT ĠWARRANTIES ĠOR ĠCONDITIONS ĠOF ĠANY ĠKIND , Ġeither Ġexpress Ġor Ġimplied . Ċ Ġ* ĠSee Ġthe ĠLicense Ġfor Ġthe Ġspecific Ġlanguage Ġgoverning Ġpermissions Ġand Ċ Ġ* Ġlimitations Ġunder Ġthe ĠLicense . Ċ Ġ*/ Ċ Ċ package Ġio . grpc . kotlin Ċ Ċ import Ġcom . google . common . truth . Truth . assertThat Ċ import Ġio . grpc . examples . helloworld . Hello Request Ċ import Ġkotlinx . coroutines . Coroutine Scope Ċ import Ġkotlinx . coroutines . Experimental Cor outines Api Ċ import Ġkotlinx . coroutines . Flow Preview Ċ import Ġkotlinx . coroutines . channels . Channel Ċ import Ġkotlinx . coroutines . channels . Receive Channel Ċ import Ġkotlinx . coroutines . channels . toList Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . flow . buffer Ċ import Ġkotlinx . coroutines . flow . collect Ċ import Ġkotlinx . coroutines . flow . consume As Flow Ċ import Ġkotlinx . coroutines . flow . flow Ċ import Ġkotlinx . coroutines . flow . map Ċ import Ġkotlinx . coroutines . flow . transform Ċ import Ġkotlinx . coroutines . launch Ċ import Ġorg . junit . Test Ċ import Ġorg . junit . runner . RunWith Ċ import Ġorg . junit . runners . JUnit 4 Ċ Ċ /** ĠTests Ġfor Ġthe Ġflow Ġcontrol Ġof Ġthe ĠKotlin ĠgRPC ĠAPIs . Ġ*/ Ċ @ RunWith ( JUnit 4 :: class ) Ċ class ĠFlow Control Test Ġ: ĠAbstract Calls Test () Ġ{ ĊĠ Ġprivate Ġfun Ġ< T > ĠFlow < T >. produce Un buffered ( scope : ĠCor outine Scope ): ĠReceive Channel < T > Ġ{ ĊĠĠĠ Ġreturn Ġscope . produce < T > Ġ{ ĊĠĠĠĠĠ Ġcollect Ġ{ Ġsend ( it ) Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġ@ Flow Preview ĊĠ Ġ@ Test ĊĠ Ġfun Ġbidi Ping P ong Flow Control () Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ Ġval Ġchannel Ġ= Ġmake Channel ( ĊĠĠĠĠĠ ĠServer Calls . b idi Streaming Server Method Definition ( ĊĠĠĠĠĠĠĠ Ġscope Ġ= Ġthis , ĊĠĠĠĠĠĠĠ Ġdescriptor Ġ= Ġbidi Streaming Say Hello Method , ĊĠĠĠĠĠĠĠ Ġimplementation Ġ= Ġ{ Ġrequests Ġ-> Ġrequests . map Ġ{ Ġhello Reply (" Hello , Ġ${ it . name }") Ġ} Ġ} ĊĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ) ĊĠĠĠ Ġval Ġrequests Ġ= ĠChannel < Hello Request >() ĊĠĠĠ Ġval Ġresponses Ġ= ĊĠĠĠĠĠ ĠClient Calls . b idi Streaming Rpc ( ĊĠĠĠĠĠĠĠ Ġchannel Ġ= Ġchannel , ĊĠĠĠĠĠĠĠ Ġrequests Ġ= Ġrequests . consume As Flow (), ĊĠĠĠĠĠĠĠ Ġmethod Ġ= Ġbidi Streaming Say Hello Method ĊĠĠĠĠĠ Ġ). produce Un buffered ( this ) ĊĠĠĠ Ġrequests . send ( hello Request (" Gar net ")) ĊĠĠĠ Ġrequests . send ( hello Request (" A meth yst ")) ĊĠĠĠ Ġval Ġthird Ġ= Ġlaunch Ġ{ Ġrequests . send ( hello Request (" St even ")) Ġ} ĊĠĠĠ Ġdelay ( 2 0 0 ) Ġ// Ġwait Ġfor Ġeverything Ġto Ġwork Ġits Ġway Ġthrough Ġthe Ġsystem ĊĠĠĠ ĠassertThat ( third . is Completed ). isFalse () ĊĠĠĠ ĠassertThat ( responses . receive ()). isEqualTo ( hello Reply (" Hello , ĠGar net ")) ĊĠĠĠ Ġthird . join () Ġ// Ġpulling Ġone Ġelement Ġallows Ġthe Ġcycle Ġto Ġadvance ĊĠĠĠ Ġresponses . cancel () ĊĠ Ġ} ĊĊĠ Ġ@ Experimental Cor outines Api Ġ// Ġbuffer ĊĠ Ġ@ Flow Preview ĊĠ Ġ@ Test ĊĠ Ġfun Ġbidi Ping P ong Flow Control Expanded Server Buffer () Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ Ġval Ġchannel Ġ= Ġmake Channel ( ĊĠĠĠĠĠ ĠServer Calls . b idi Streaming Server Method Definition ( ĊĠĠĠĠĠĠĠ Ġscope Ġ= Ġthis , ĊĠĠĠĠĠĠĠ Ġdescriptor Ġ= Ġbidi Streaming Say Hello Method , ĊĠĠĠĠĠĠĠ Ġimplementation Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġrequests Ġ-> Ġrequests . buffer ( Channel . REN DE Z VO US ). map Ġ{ Ġhello Reply (" Hello , Ġ${ it . name }") Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ) ĊĠĠĠ Ġval Ġrequests Ġ= ĠChannel < Hello Request >() ĊĠĠĠ Ġval Ġresponses Ġ= ĠClient Calls . b idi Streaming Rpc ( ĊĠĠĠĠĠ Ġchannel Ġ= Ġchannel , ĊĠĠĠĠĠ Ġrequests Ġ= Ġrequests . consume As Flow (), ĊĠĠĠĠĠ Ġmethod Ġ= Ġbidi Streaming Say Hello Method ĊĠĠĠ Ġ). produce Un buffered ( this ) ĊĠĠĠ Ġrequests . send ( hello Request (" Gar net ")) ĊĠĠĠ Ġrequests . send ( hello Request (" A meth yst ")) ĊĠĠĠ Ġrequests . send ( hello Request (" P ear l ")) ĊĠĠĠ Ġval Ġfourth Ġ= Ġlaunch Ġ{ Ġrequests . send ( hello Request (" P ear l ")) Ġ} ĊĠĠĠ Ġdelay ( 2 0 0 ) Ġ// Ġwait Ġfor Ġeverything Ġto Ġwork Ġits Ġway Ġthrough Ġthe Ġsystem ĊĠĠĠ ĠassertThat ( four th . is Completed ). isFalse () ĊĠĠĠ ĠassertThat ( responses . receive ()). isEqualTo ( hello Reply (" Hello , ĠGar net ")) ĊĠĠĠ Ġfourth . join () Ġ// Ġpulling Ġone Ġelement Ġallows Ġthe Ġcycle Ġto Ġadvance ĊĠĠĠ Ġresponses . cancel () ĊĠ Ġ} ĊĊĠ Ġ@ Flow Preview ĊĠ Ġ@ Test ĊĠ Ġfun Ġbidi Ping P ong Flow Control Server Draw s Multiple Requests () Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ Ġfun Ġ< T : ĠAny > ĠFlow < T >. pair Off (): ĠFlow < Pair < T , ĠT >> Ġ= Ġflow Ġ{ ĊĠĠĠĠĠ Ġvar Ġodd : ĠT ? Ġ= Ġnull ĊĠĠĠĠĠ Ġcollect Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġo Ġ= Ġodd ĊĠĠĠĠĠĠĠ Ġif Ġ( o Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġodd Ġ= Ġit ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġemit ( Pair ( o , Ġit )) ĊĠĠĠĠĠĠĠĠĠ Ġodd Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġval Ġchannel Ġ= Ġmake Channel ( ĊĠĠĠĠĠ ĠServer Calls . b idi Streaming Server Method Definition ( ĊĠĠĠĠĠĠĠ Ġscope Ġ= Ġthis , ĊĠĠĠĠĠĠĠ Ġdescriptor Ġ= Ġbidi Streaming Say Hello Method , ĊĠĠĠĠĠĠĠ Ġimplementation Ġ= Ġ{ Ġrequests Ġ-> ĊĠĠĠĠĠĠĠĠĠ Ġrequests . pair Off (). map Ġ{ Ġ( a , Ġb ) Ġ-> Ġhello Reply (" Hello , Ġ${ a . name } Ġand Ġ${ b . name }") Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ) ĊĠĠĠ Ġval Ġrequests Ġ= ĠChannel < Hello Request >() ĊĠĠĠ Ġval Ġresponses Ġ= ĠClient Calls . b idi Streaming Rpc ( ĊĠĠĠĠĠ Ġchannel Ġ= Ġchannel , ĊĠĠĠĠĠ Ġrequests Ġ= Ġrequests . consume As Flow (), ĊĠĠĠĠĠ Ġmethod Ġ= Ġbidi Streaming Say Hello Method ĊĠĠĠ Ġ). produce Un buffered ( this ) ĊĠĠĠ Ġrequests . send ( hello Request (" Gar net ")) ĊĠĠĠ Ġrequests . send ( hello Request (" A meth yst ")) ĊĠĠĠ Ġrequests . send ( hello Request (" P ear l ")) ĊĠĠĠ Ġrequests . send ( hello Request (" St even ")) ĊĠĠĠ Ġval Ġfourth Ġ= Ġlaunch Ġ{ Ġrequests . send ( hello Request (" On ion ")) Ġ} ĊĠĠĠ Ġdelay ( 3 0 0 ) Ġ// Ġwait Ġfor Ġeverything Ġto Ġwork Ġits Ġway Ġthrough Ġthe Ġsystem ĊĠĠĠ ĠassertThat ( four th . is Completed ). isFalse () ĊĠĠĠ ĠassertThat ( responses . receive ()). isEqualTo ( hello Reply (" Hello , ĠGar net Ġand ĠA meth yst ")) ĊĠĠĠ Ġfourth . join () Ġ// Ġpulling Ġone Ġelement Ġallows Ġthe Ġcycle Ġto Ġadvance ĊĠĠĠ Ġrequests . send ( hello Request (" Rain bow Ġ 2 . 0 ")) ĊĠĠĠ Ġrequests . close () ĊĠĠĠ ĠassertThat ( responses . toList ()). contains Exactly ( ĊĠĠĠĠĠ Ġhello Reply (" Hello , ĠP ear l Ġand ĠSte ven "), Ġhello Reply (" Hello , ĠOn ion Ġand ĠR ain bow Ġ 2 . 0 ") ĊĠĠĠ Ġ) ĊĠ Ġ} ĊĊĠ Ġ@ Experimental Cor outines Api Ġ// Ġtransform ĊĠ Ġ@ Flow Preview ĊĠ Ġ@ Test ĊĠ Ġfun Ġbidi Ping P ong Flow Control Server S ends Multiple Responses () Ġ= Ġrun Blocking Ġ{ ĊĠĠĠ Ġval Ġchannel Ġ= Ġmake Channel ( ĊĠĠĠĠĠ ĠServer Calls . b idi Streaming Server Method Definition ( ĊĠĠĠĠĠĠĠ Ġscope Ġ= Ġthis , ĊĠĠĠĠĠĠĠ Ġdescriptor Ġ= Ġbidi Streaming Say Hello Method , ĊĠĠĠĠĠĠĠ Ġimplementation Ġ= Ġ{ Ġrequests Ġ-> ĊĠĠĠĠĠĠĠĠĠ Ġrequests . transform Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġemit ( hello Reply (" Hello , Ġ${ it . name } ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġemit ( hello Reply (" Good bye , Ġ${ it . name } ")) ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ) ĊĠĠĠ Ġval Ġrequests Ġ= ĠChannel < Hello Request >() ĊĠĠĠ Ġval Ġresponses Ġ= ĠClient Calls . b idi Streaming Rpc ( ĊĠĠĠĠĠ Ġchannel Ġ= Ġchannel , ĊĠĠĠĠĠ Ġrequests Ġ= Ġrequests . consume As Flow (), ĊĠĠĠĠĠ Ġmethod Ġ= Ġbidi Streaming Say Hello Method ĊĠĠĠ Ġ). produce Un buffered ( this ) ĊĠĠĠ Ġrequests . send ( hello Request (" Gar net ")) ĊĠĠĠ Ġval Ġsecond Ġ= Ġlaunch Ġ{ Ġrequests . send ( hello Request (" P ear l ")) Ġ} ĊĠĠĠ Ġdelay ( 2 0 0 ) Ġ// Ġwait Ġfor Ġeverything Ġto Ġwork Ġits Ġway Ġthrough Ġthe Ġsystem ĊĠĠĠ ĠassertThat ( second . is Completed ). isFalse () ĊĠĠĠ ĠassertThat ( responses . receive ()). isEqualTo ( hello Reply (" Hello , ĠGar net ")) ĊĠĠĠ Ġsecond . join () ĊĠĠĠ ĠassertThat ( responses . receive ()). isEqualTo ( hello Reply (" Good bye , ĠGar net ")) ĊĠĠĠ Ġresponses . cancel () ĊĠ Ġ} Ċ } Ċ