/* Ċ Ġ* ĠCopyright Ġ 2 0 1 5 - 2 0 2 2 Ġthe Ġoriginal Ġauthor Ġor Ġ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 . r socket . kotlin Ċ Ċ import Ġio . ktor . utils . io . core .* Ċ import Ġio . r socket . kotlin . payload .* Ċ import Ġkotlinx . coroutines .* Ċ import Ġkotlinx . coroutines . flow .* Ċ import Ġkotlin . coroutines .* Ċ Ċ public Ġclass ĠR Socket RequestHandler Builder Ġinternal Ġconstructor () Ġ{ ĊĠĠĠ Ġprivate Ġvar Ġmetadata Push : Ġ( suspend ĠR Socket .( metadata : ĠByte Read Packet ) Ġ-> ĠUnit )? Ġ= Ġnull ĊĠĠĠ Ġprivate Ġvar Ġfire And For get : Ġ( suspend ĠR Socket .( payload : ĠPayload ) Ġ-> ĠUnit )? Ġ= Ġnull ĊĠĠĠ Ġprivate Ġvar Ġrequest Response : Ġ( suspend ĠR Socket .( payload : ĠPayload ) Ġ-> ĠPayload )? Ġ= Ġnull ĊĠĠĠ Ġprivate Ġvar Ġrequest Stream : Ġ( suspend ĠR Socket .( payload : ĠPayload ) Ġ-> ĠFlow < Payload >) ? Ġ= Ġnull ĊĠĠĠ Ġprivate Ġvar Ġrequest Channel : Ġ( suspend ĠR Socket .( init Payload : ĠPayload , Ġpayload s : ĠFlow < Payload >) Ġ-> ĠFlow < Payload >) ? Ġ= ĊĠĠĠĠĠĠĠ Ġnull ĊĊĠĠĠ Ġpublic Ġfun Ġmetadata Push ( block : Ġ( suspend ĠR Socket .( metadata : ĠByte Read Packet ) Ġ-> ĠUnit )) Ġ{ ĊĠĠĠĠĠĠĠ Ġcheck ( metadata Push Ġ== Ġnull ) Ġ{ Ġ" Metadata ĠPush Ġhandler Ġalready Ġconfigured " Ġ} ĊĠĠĠĠĠĠĠ Ġmetadata Push Ġ= Ġblock ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġpublic Ġfun Ġfire And For get ( block : Ġ( suspend ĠR Socket .( payload : ĠPayload ) Ġ-> ĠUnit )) Ġ{ ĊĠĠĠĠĠĠĠ Ġcheck ( fire And For get Ġ== Ġnull ) Ġ{ Ġ" Fire Ġand ĠFor get Ġhandler Ġalready Ġconfigured " Ġ} ĊĠĠĠĠĠĠĠ Ġfire And For get Ġ= Ġblock ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġpublic Ġfun Ġrequest Response ( block : Ġ( suspend ĠR Socket .( payload : ĠPayload ) Ġ-> ĠPayload )) Ġ{ ĊĠĠĠĠĠĠĠ Ġcheck ( request Response Ġ== Ġnull ) Ġ{ Ġ" Request ĠResponse Ġhandler Ġalready Ġconfigured " Ġ} ĊĠĠĠĠĠĠĠ Ġrequest Response Ġ= Ġblock ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġpublic Ġfun Ġrequest Stream ( block : Ġsuspend Ġ( R Socket .( payload : ĠPayload ) Ġ-> ĠFlow < Payload > )) Ġ{ ĊĠĠĠĠĠĠĠ Ġcheck ( request Stream Ġ== Ġnull ) Ġ{ Ġ" Request ĠStream Ġhandler Ġalready Ġconfigured " Ġ} ĊĠĠĠĠĠĠĠ Ġrequest Stream Ġ= Ġblock ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġpublic Ġfun Ġrequest Channel ( block : Ġsuspend Ġ( R Socket .( init Payload : ĠPayload , Ġpayload s : ĠFlow < Payload >) Ġ-> ĠFlow < Payload > )) Ġ{ ĊĠĠĠĠĠĠĠ Ġcheck ( request Channel Ġ== Ġnull ) Ġ{ Ġ" Request ĠChannel Ġhandler Ġalready Ġconfigured " Ġ} ĊĠĠĠĠĠĠĠ Ġrequest Channel Ġ= Ġblock ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġinternal Ġfun Ġbuild ( parent Context : ĠCor outine Context ): ĠR Socket Ġ= ĊĠĠĠĠĠĠĠ ĠR Socket RequestHandler ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġparent Context Ġ+ ĠJob ( parent Context [ Job ]), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmetadata Push , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfire And For get , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequest Response , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequest Stream , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrequest Channel ĊĠĠĠĠĠĠĠ Ġ) Ċ } Ċ Ċ @ Suppress (" FunctionName ") Ċ public Ġfun ĠR Socket RequestHandler ( ĊĠĠĠ Ġparent Context : ĠCor outine Context Ġ= ĠEmpty Coroutine Context , ĊĠĠĠ Ġconfigure : ĠR Socket RequestHandler Builder . () Ġ-> ĠUnit Ċ ): ĠR Socket Ġ{ ĊĠĠĠ Ġval Ġbuilder Ġ= ĠR Socket RequestHandler Builder () ĊĠĠĠ Ġbuilder . configure () ĊĠĠĠ Ġreturn Ġbuilder . build ( parent Context ) Ċ } Ċ Ċ private Ġclass ĠR Socket RequestHandler ( ĊĠĠĠ Ġoverride Ġval Ġcoroutine Context : ĠCor outine Context , ĊĠĠĠ Ġprivate Ġval Ġmetadata Push : Ġ( suspend ĠR Socket .( metadata : ĠByte Read Packet ) Ġ-> ĠUnit )? Ġ= Ġnull , ĊĠĠĠ Ġprivate Ġval Ġfire And For get : Ġ( suspend ĠR Socket .( payload : ĠPayload ) Ġ-> ĠUnit )? Ġ= Ġnull , ĊĠĠĠ Ġprivate Ġval Ġrequest Response : Ġ( suspend ĠR Socket .( payload : ĠPayload ) Ġ-> ĠPayload )? Ġ= Ġnull , ĊĠĠĠ Ġprivate Ġval Ġrequest Stream : Ġ( suspend ĠR Socket .( payload : ĠPayload ) Ġ-> ĠFlow < Payload >) ? Ġ= Ġnull , ĊĠĠĠ Ġprivate Ġval Ġrequest Channel : Ġ( suspend ĠR Socket .( init Payload : ĠPayload , Ġpayload s : ĠFlow < Payload >) Ġ-> ĠFlow < Payload >) ? Ġ= Ġnull , Ċ ) Ġ: ĠR Socket Ġ{ ĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġmetadata Push ( metadata : ĠByte Read Packet ): ĠUnit Ġ= ĊĠĠĠĠĠĠĠ Ġmetadata Push ?. invoke ( this , Ġmetadata ) Ġ?: Ġsuper . metadata Push ( metadata ) ĊĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġfire And For get ( payload : ĠPayload ): ĠUnit Ġ= ĊĠĠĠĠĠĠĠ Ġfire And For get ?. invoke ( this , Ġpayload ) Ġ?: Ġsuper . fire And For get ( payload ) ĊĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġrequest Response ( payload : ĠPayload ): ĠPayload Ġ= ĊĠĠĠĠĠĠĠ Ġrequest Response ?. invoke ( this , Ġpayload ) Ġ?: Ġsuper . request Response ( payload ) ĊĊĠĠĠ Ġoverride Ġfun Ġrequest Stream ( payload : ĠPayload ): ĠFlow < Payload > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġif Ġ( this . request Stream Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow Ġ{ Ġemit All ( request Stream . invoke ( this @ R Socket RequestHandler , Ġpayload )) Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsuper . request Stream ( payload ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġrequest Channel ( init Payload : ĠPayload , Ġpayload s : ĠFlow < Payload >): ĠFlow < Payload > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġif Ġ( this . request Channel Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflow Ġ{ Ġemit All ( request Channel . invoke ( this @ R Socket RequestHandler , Ġinit Payload , Ġpayload s )) Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsuper . request Channel ( init Payload , Ġpayload s ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ Ċ } Ċ