/* Ċ Ġ* ĠCopyright Ġ 2 0 1 5 - 2 0 2 0 Ġ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 . internal . flow Ċ Ċ import Ġio . r socket . kotlin .* Ċ import Ġio . r socket . kotlin . frame .* Ċ import Ġio . r socket . kotlin . internal .* Ċ import Ġio . r socket . kotlin . payload .* Ċ import Ġkotlinx . atomic fu .* Ċ import Ġkotlinx . coroutines .* Ċ import Ġkotlinx . coroutines . channels .* Ċ import Ġkotlinx . coroutines . flow .* Ċ Ċ @ Opt In ( Experimental Streams Api :: class ) Ċ internal Ġclass ĠRequest Channel Responder Flow ( ĊĠĠĠ Ġprivate Ġval Ġstream Id : ĠInt , ĊĠĠĠ Ġprivate Ġval Ġreceiver : ĠReceive Channel < Request Frame >, ĊĠĠĠ Ġprivate Ġval Ġstate : ĠR Socket State , Ċ ) Ġ: ĠFlow < Payload > Ġ{ ĊĠĠĠ Ġprivate Ġval Ġconsumed Ġ= Ġatomic ( false ) ĊĊĠĠĠ Ġ@ Internal Cor outines Api ĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġcollect ( collector : ĠFlow Collector < Payload >): ĠUnit Ġ= Ġwith ( state ) Ġ{ ĊĠĠĠĠĠĠĠ Ġcheck (! consumed . get AndSet ( true )) Ġ{ Ġ" R Socket . request Channel Ġ` payload s ` Ġcan Ġbe Ġcollected Ġjust Ġonce " Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġstrategy Ġ= Ġcurrent Coroutine Context (). request Strategy () ĊĠĠĠĠĠĠĠ Ġval Ġinitial Request Ġ= Ġstrategy . first Request () ĊĠĠĠĠĠĠĠ Ġsend ( Request N Frame ( stream Id , Ġinitial Request )) ĊĠĠĠĠĠĠĠ Ġcollect Stream ( stream Id , Ġreceiver , Ġstrategy , Ġcollector ) ĊĠĠĠ Ġ} Ċ } Ċ