/* Ċ Ġ* ĠCopyright Ġ 2 0 1 9 ĠRed ĠHat , ĠInc . Ċ Ġ* Ċ Ġ* ĠAll Ġrights Ġreserved . ĠThis Ġprogram Ġand Ġthe Ġaccompanying Ġmaterials Ċ Ġ* Ġare Ġmade Ġavailable Ġunder Ġthe Ġterms Ġof Ġthe ĠEclipse ĠPublic ĠLicense Ġv 1 . 0 Ċ Ġ* Ġand ĠApache ĠLicense Ġv 2 . 0 Ġwhich Ġaccomp anies Ġthis Ġdistribution . Ċ Ġ* Ċ Ġ* ĠThe ĠEclipse ĠPublic ĠLicense Ġis Ġavailable Ġat Ċ Ġ* Ġhttp :// www . eclipse . org / legal / e pl - v 1 0 . html Ċ Ġ* Ċ Ġ* ĠThe ĠApache ĠLicense Ġv 2 . 0 Ġis Ġavailable Ġat Ċ Ġ* Ġhttp :// www . opensource . org / licenses / apache 2 . 0 . php Ċ Ġ* Ċ Ġ* ĠYou Ġmay Ġelect Ġto Ġredistribute Ġthis Ġcode Ġunder Ġeither Ġof Ġthese Ġlicenses . Ċ Ġ*/ Ċ package Ġio . vertx . kotlin . coroutines Ċ Ċ import Ġio . vertx . core . Context Ċ import Ġio . vertx . core . Handler Ċ import Ġio . vertx . core . Vert x Ċ import Ġio . vertx . core . streams . Read Stream Ċ import Ġio . vertx . core . streams . Write Stream Ċ import Ġkotlinx . coroutines .* Ċ import Ġkotlinx . coroutines . channels .* Ċ import Ġkotlinx . coroutines . select s . Select Clause 1 Ċ import Ġkotlin . coroutines . Coroutine Context Ċ Ċ /** Ċ Ġ* Ġ@ author Ġ< a Ġhref =" http :// www . stream is . me "> Stream ĠLi u </ a > Ċ Ġ* Ġ@ author Ġ< a Ġhref =" mailto : jul ien @ j uli env iet . com "> Jul ien ĠV iet </ a > Ċ Ġ* Ġ@ author Ġ[ Jul ien ĠP onge ]( https :// jul ien . pon ge . org /) Ċ Ġ* Ġ@ author Ġ[ Guid o ĠP io ĠM ari ot ti ]( https :// github . com / gm ari ot ti ) Ċ Ġ*/ Ċ Ċ /** Ċ Ġ* ĠAn Ġadapter Ġthat Ġconverts Ġa Ġstream Ġof Ġevents Ġfrom Ġthe Ġ[ Handler ] Ġinto Ġa Ġ[ Receive Channel ] Ġwhich Ġallows Ġthe Ġevents Ċ Ġ* Ġto Ġbe Ġreceived Ġsynchronous ly . Ċ Ġ*/ Ċ class ĠReceive Channel Handler < T >( context : ĠContext ) Ġ: ĠReceive Channel < T >, ĠHandler < T >, ĠCor outine Scope Ġ{ ĊĊĠ Ġconstructor ( vertx : ĠVert x ) Ġ: Ġthis ( vertx . get OrCreate Context ()) ĊĊĠ Ġoverride Ġval Ġcoroutine Context : ĠCor outine Context Ġ= Ġcontext . dispatcher () ĊĠ Ġprivate Ġval Ġchannel : ĠChannel < T > Ġ= ĠChannel ( DEFAULT _ CAP ACITY ) ĊĊĠ Ġ@ Experimental Cor outines Api ĊĠ Ġoverride Ġval Ġis Closed For Receive : ĠBoolean ĊĠĠĠ Ġget () Ġ= Ġchannel . is Closed For Receive ĊĊĠ Ġ@ Experimental Cor outines Api ĊĠ Ġoverride Ġval ĠisEmpty : ĠBoolean ĊĠĠĠ Ġget () Ġ= Ġchannel . isEmpty ĊĊĠ Ġoverride Ġfun Ġiterator (): ĠChannel Iterator < T > Ġ{ ĊĠĠĠ Ġreturn Ġchannel . iterator () ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġpoll (): ĠT ? Ġ{ ĊĠĠĠ Ġreturn Ġchannel . poll () ĊĠ Ġ} ĊĊĠ Ġoverride Ġsuspend Ġfun Ġreceive (): ĠT Ġ{ ĊĠĠĠ Ġreturn Ġchannel . receive () ĊĠ Ġ} ĊĊĠ Ġoverride Ġval Ġon Receive : ĠSelect Clause 1 < T > ĊĠĠĠ Ġget () Ġ= Ġchannel . on Receive ĊĊĠ Ġoverride Ġval Ġon Receive Catch ing : ĠSelect Clause 1 < Channel Result < T >> ĊĠĠĠ Ġget () Ġ= Ġchannel . on Receive Catch ing ĊĊĠ Ġoverride Ġfun Ġhandle ( event : ĠT ) Ġ{ ĊĠĠĠ Ġlaunch Ġ{ Ġchannel . send ( event ) Ġ} ĊĠ Ġ} ĊĊĠ Ġoverride Ġsuspend Ġfun Ġreceive Catch ing (): ĠChannel Result < T > Ġ{ ĊĠĠĠ Ġreturn Ġchannel . receive Catch ing () ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġtry Receive (): ĠChannel Result < T > Ġ{ ĊĠĠĠ Ġreturn Ġchannel . try Receive () ĊĠ Ġ} ĊĊĠ Ġ@ Obsolete Cor outines Api ĊĠ Ġ@ Deprecated ( level Ġ= ĠDeprec ation Level . HIDDEN , Ġmessage Ġ= Ġ" Since Ġ 3 . 7 . 1 , Ġbinary Ġcompatibility Ġwith Ġversions Ġ<= Ġ 3 . 7 . 0 ") ĊĠ Ġoverride Ġfun Ġcancel ( cause : ĠThrowable ?): ĠBoolean Ġ{ ĊĠĠĠ Ġchannel . cancel ( Cancellation Exception ( message Ġ= Ġnull , Ġcause Ġ= Ġcause )) ĊĠĠĠ Ġreturn Ġtrue ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġcancel ( cause : ĠC ancellation Exception ?) Ġ{ ĊĠĠĠ Ġchannel . cancel ( cause ) ĊĠ Ġ} ĊĊĠ Ġ@ Deprecated ( level Ġ= ĠDeprec ation Level . ERROR , Ġmessage Ġ= Ġ" Since Ġ 3 . 7 . 1 , Ġbinary Ġcompatibility Ġwith Ġversions Ġ<= Ġ 3 . 7 . 0 ") ĊĠ Ġoverride Ġfun Ġcancel () Ġ{ ĊĠĠĠ Ġreturn Ġchannel . cancel () ĊĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* ĠCreate Ġa Ġ[ Receive Channel Handler ] Ġof Ġsome Ġtype Ġ` T `. Ċ Ġ*/ Ċ fun Ġ< T > ĠVert x . receive Channel Handler (): ĠReceive Channel Handler < T > Ġ= ĠReceive Channel Handler ( this ) Ċ Ċ /** Ċ Ġ* ĠAd ap ts Ġthe Ġcurrent Ġread Ġstream Ġto ĠKotlin Ġ[ Receive Channel ]. Ċ Ġ* Ċ Ġ* ĠThe Ġadapter Ġwill Ġfetch Ġat Ġmost Ġmax Ġchannel Ġcapacity Ġfrom Ġthe Ġstream Ġand Ġpause Ġit Ġwhen Ġthe Ġchannel Ġis Ġfull . Ċ Ġ* Ċ Ġ* Ġ@ param Ġvert x Ġthe Ġrelated Ġvert x Ġinstance Ċ Ġ* Ġ@ param Ġcapacity Ġthe Ġchannel Ġbuffer ing Ġcapacity Ċ Ġ*/ Ċ @ Deprecated (" Please Ġuse Ġto Receive Channel Ġinstead Ġto Ġavoid Ġname Ġcl ash ") Ċ fun Ġ< T > ĠRead Stream < T >. to Channel ( vertx : ĠVert x ): ĠReceive Channel < T > Ġ{ ĊĠ Ġreturn Ġto Channel ( vertx . or Create Context ) Ċ } Ċ Ċ /** Ċ Ġ* ĠAd ap ts Ġthe Ġcurrent Ġread Ġstream Ġto ĠKotlin Ġ[ Receive Channel ]. Ċ Ġ* Ċ Ġ* ĠThe Ġadapter Ġwill Ġfetch Ġat Ġmost Ġmax Ġchannel Ġcapacity Ġfrom Ġthe Ġstream Ġand Ġpause Ġit Ġwhen Ġthe Ġchannel Ġis Ġfull . Ċ Ġ* Ċ Ġ* Ġ@ param Ġvert x Ġthe Ġrelated Ġvert x Ġinstance Ċ Ġ* Ġ@ param Ġcapacity Ġthe Ġchannel Ġbuffer ing Ġcapacity Ċ Ġ*/ Ċ fun Ġ< T > ĠRead Stream < T >. to Receive Channel ( vertx : ĠVert x ): ĠReceive Channel < T > Ġ{ ĊĠ Ġreturn Ġto Channel ( vertx . or Create Context ) Ċ } Ċ Ċ /** Ċ Ġ* ĠAd ap ts Ġthe Ġcurrent Ġread Ġstream Ġto ĠKotlin Ġ[ Receive Channel ]. Ċ Ġ* Ċ Ġ* Ġ@ param Ġcontext Ġthe Ġvert x Ġcontext Ċ Ġ*/ Ċ @ Deprecated (" P le as Ġuse Ġto Receive Channel Ġinstead Ġto Ġavoid Ġname Ġcl ashes ") Ċ fun Ġ< T > ĠRead Stream < T >. to Channel ( context : ĠContext ): ĠReceive Channel < T > Ġ{ ĊĠ Ġthis . pause () ĊĠ Ġval Ġret Ġ= ĠChannel Read Stream ( ĊĠĠĠ Ġstream Ġ= Ġthis , ĊĠĠĠ Ġchannel Ġ= ĠChannel ( 0 ), ĊĠĠĠ Ġcontext Ġ= Ġcontext ĊĠ Ġ) ĊĠ Ġret . subscribe () ĊĠ Ġthis . fetch ( 1 ) ĊĠ Ġreturn Ġret Ċ } Ċ Ċ /** Ċ Ġ* ĠAd ap ts Ġthe Ġcurrent Ġread Ġstream Ġto ĠKotlin Ġ[ Receive Channel ]. Ċ Ġ* Ċ Ġ* Ġ@ param Ġcontext Ġthe Ġvert x Ġcontext Ċ Ġ*/ Ċ fun Ġ< T > ĠRead Stream < T >. to Receive Channel ( context : ĠContext ): ĠReceive Channel < T > Ġ{ ĊĠ Ġreturn Ġthis . to Channel ( context ) Ċ } Ċ Ċ private Ġclass ĠChannel Read Stream < T >( val Ġstream : ĠRead Stream < T >, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġchannel : ĠChannel < T >, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcontext : ĠContext ) Ġ: ĠChannel < T > Ġby Ġchannel , ĠCor outine Scope Ġ{ ĊĊĠ Ġoverride Ġval Ġcoroutine Context : ĠCor outine Context Ġ= Ġcontext . dispatcher () ĊĊĠ Ġfun Ġsubscribe () Ġ{ ĊĠĠĠ Ġstream . end Handler Ġ{ ĊĠĠĠĠĠ Ġclose () ĊĠĠĠ Ġ} ĊĠĠĠ Ġstream . exception Handler Ġ{ Ġerr Ġ-> ĊĠĠĠĠĠ Ġclose ( err ) ĊĠĠĠ Ġ} ĊĠĠĠ Ġstream . handler Ġ{ Ġevent Ġ-> ĊĠĠĠĠĠ Ġlaunch Ġ{ ĊĠĠĠĠĠĠĠ Ġsend ( event ) ĊĠĠĠĠĠĠĠ Ġstream . fetch ( 1 ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* ĠAd ap ts Ġthe Ġcurrent Ġwrite Ġstream Ġto ĠKotlin Ġ[ Send Channel ]. Ċ Ġ* Ċ Ġ* ĠThe Ġchannel Ġcan Ġbe Ġused Ġto Ġwrite Ġitems , Ġthe Ġcoroutine Ġis Ġsus pended Ġwhen Ġthe Ġstream Ġis Ġfull Ċ Ġ* Ġand Ġres umed Ġwhen Ġthe Ġstream Ġis Ġdr ained . Ċ Ġ* Ċ Ġ* Ġ@ param Ġvert x Ġthe Ġrelated Ġvert x Ġinstance Ċ Ġ* Ġ@ param Ġcapacity Ġthe Ġchannel Ġbuffer ing Ġcapacity Ċ Ġ*/ Ċ @ Deprecated (" Please Ġuse Ġinstead Ġto Send Channel , Ġto Ġavoid Ġname Ġcl ashes ") Ċ @ Experimental Cor outines Api Ċ fun Ġ< T > ĠWrite Stream < T >. to Channel ( vertx : ĠVert x , Ġcapacity : ĠInt Ġ= ĠDEFAULT _ CAP ACITY ): ĠSend Channel < T > Ġ{ ĊĠ Ġreturn Ġto Channel ( vertx . or Create Context , Ġcapacity ) Ċ } Ċ Ċ /** Ċ Ġ* ĠAd ap ts Ġthe Ġcurrent Ġwrite Ġstream Ġto ĠKotlin Ġ[ Send Channel ]. Ċ Ġ* Ċ Ġ* ĠThe Ġchannel Ġcan Ġbe Ġused Ġto Ġwrite Ġitems , Ġthe Ġcoroutine Ġis Ġsus pended Ġwhen Ġthe Ġstream Ġis Ġfull Ċ Ġ* Ġand Ġres umed Ġwhen Ġthe Ġstream Ġis Ġdr ained . Ċ Ġ* Ċ Ġ* Ġ@ param Ġvert x Ġthe Ġrelated Ġvert x Ġinstance Ċ Ġ* Ġ@ param Ġcapacity Ġthe Ġchannel Ġbuffer ing Ġcapacity Ċ Ġ*/ Ċ fun Ġ< T > ĠWrite Stream < T >. to Send Channel ( vertx : ĠVert x , Ġcapacity : ĠInt Ġ= ĠDEFAULT _ CAP ACITY ): ĠSend Channel < T > Ġ{ ĊĠ Ġreturn Ġto Channel ( vertx . or Create Context , Ġcapacity ) Ċ } Ċ Ċ /** Ċ Ġ* ĠAd ap ts Ġthe Ġcurrent Ġwrite Ġstream Ġto ĠKotlin Ġ[ Send Channel ]. Ċ Ġ* Ċ Ġ* ĠThe Ġchannel Ġcan Ġbe Ġused Ġto Ġwrite Ġitems , Ġthe Ġcoroutine Ġis Ġsus pended Ġwhen Ġthe Ġstream Ġis Ġfull Ċ Ġ* Ġand Ġres umed Ġwhen Ġthe Ġstream Ġis Ġdr ained . Ċ Ġ* Ċ Ġ* Ġ@ param Ġcontext Ġthe Ġvert x Ġcontext Ċ Ġ* Ġ@ param Ġcapacity Ġthe Ġchannel Ġbuffer ing Ġcapacity Ċ Ġ*/ Ċ @ Deprecated (" Please Ġuse Ġto Send Channel , Ġto Ġavoid Ġname Ġcl ash ") Ċ @ Experimental Cor outines Api Ċ fun Ġ< T > ĠWrite Stream < T >. to Channel ( context : ĠContext , Ġcapacity : ĠInt Ġ= ĠDEFAULT _ CAP ACITY ): ĠSend Channel < T > Ġ{ ĊĠ Ġval Ġret Ġ= ĠChannel Write Stream ( ĊĠĠĠ Ġstream Ġ= Ġthis , ĊĠĠĠ Ġchannel Ġ= ĠChannel ( capacity ), ĊĠĠĠ Ġcontext Ġ= Ġcontext ĊĠ Ġ) ĊĠ Ġret . subscribe () ĊĠ Ġreturn Ġret Ċ } Ċ Ċ /** Ċ Ġ* ĠAd ap ts Ġthe Ġcurrent Ġwrite Ġstream Ġto ĠKotlin Ġ[ Send Channel ]. Ċ Ġ* Ċ Ġ* ĠThe Ġchannel Ġcan Ġbe Ġused Ġto Ġwrite Ġitems , Ġthe Ġcoroutine Ġis Ġsus pended Ġwhen Ġthe Ġstream Ġis Ġfull Ċ Ġ* Ġand Ġres umed Ġwhen Ġthe Ġstream Ġis Ġdr ained . Ċ Ġ* Ċ Ġ* Ġ@ param Ġcontext Ġthe Ġvert x Ġcontext Ċ Ġ* Ġ@ param Ġcapacity Ġthe Ġchannel Ġbuffer ing Ġcapacity Ċ Ġ*/ Ċ @ Experimental Cor outines Api Ċ fun Ġ< T > ĠWrite Stream < T >. to Send Channel ( context : ĠContext , Ġcapacity : ĠInt Ġ= ĠDEFAULT _ CAP ACITY ): ĠSend Channel < T > Ġ{ ĊĠ Ġreturn Ġthis . to Channel ( context , Ġcapacity ) Ċ } Ċ Ċ private Ġclass ĠChannel Write Stream < T >( val Ġstream : ĠWrite Stream < T >, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġchannel : ĠChannel < T >, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcontext : ĠContext ) Ġ: ĠChannel < T > Ġby Ġchannel , ĠCor outine Scope Ġ{ ĊĊĠ Ġoverride Ġval Ġcoroutine Context : ĠCor outine Context Ġ= Ġcontext . dispatcher () ĊĊĠ Ġ@ Experimental Cor outines Api ĊĠ Ġfun Ġsubscribe () Ġ{ ĊĠĠĠ Ġstream . exception Handler Ġ{ ĊĠĠĠĠĠ Ġchannel . close ( it ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġlaunch Ġ{ ĊĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġres Ġ= Ġchannel . receive Catch ing () ĊĠĠĠĠĠĠĠ Ġif Ġ( res . is Success ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġval Ġelt Ġ= Ġres . get OrNull (); ĊĠĠĠĠĠĠĠĠĠ Ġif Ġ( stream . write Queue Full ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġstream . drain Handler Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( dispatch ( elt )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsubscribe () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! dispatch ( elt )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( res . is Closed ) Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġ// ĠCan Ġit Ġhappen ? ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġfun Ġdispatch ( elt : ĠT ?): ĠBoolean Ġ{ ĊĠĠĠ Ġreturn Ġif Ġ( elt Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠ Ġstream . write ( elt ) ĊĠĠĠĠĠ Ġtrue ĊĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠ Ġclose () ĊĠĠĠĠĠ Ġfalse ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġclose ( cause : ĠThrowable ?): ĠBoolean Ġ{ ĊĠĠĠ Ġval Ġret Ġ= Ġchannel . close ( cause ) ĊĠĠĠ Ġif Ġ( ret ) Ġstream . end () ĊĠĠĠ Ġreturn Ġret ĊĠ Ġ} Ċ } Ċ Ċ private Ġconst Ġval ĠDEFAULT _ CAP ACITY Ġ= Ġ 1 6 Ċ