/* Ċ Ġ* ĠCopyright Ġ( C ) Ġ 2 0 2 1 ĠSquare , ĠInc . Ċ Ġ* Ċ Ġ* Ġ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 Ġapp . cash . zi pline Ċ Ċ import Ġkotlinx . coroutines . Experimental Cor outines Api Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . flow . channel Flow Ċ import Ġkotlinx . serialization . K Serializer Ċ import Ġkotlinx . serialization . descriptors . Serial Descriptor Ċ import Ġkotlinx . serialization . encoding . Decoder Ċ import Ġkotlinx . serialization . encoding . Encoder Ċ import Ġkotlinx . serialization . json . Json Ċ Ċ /** ĠA Ġflow Ġreference Ġis Ġa Ġ[ Flow ] Ġthat Ġcan Ġbe Ġtransmitted Ġover Ġa ĠZ i pline Ġcall . Ġ*/ Ċ class ĠFlow Reference < T > Ġ@ Published Api Ġinternal Ġconstructor ( ĊĠ Ġinternal Ġval Ġreference Flow Reference : ĠZ i pline Reference < Reference Flow >, ĊĠ Ġinternal Ġval Ġitem Serializer : ĠK Serializer < T >, Ċ ) Ġ{ ĊĠ Ġ@ Opt In ( Experimental Cor outines Api :: class ) ĊĠ Ġprivate Ġfun Ġget De coding Flow ( reference Flow : ĠReference Flow ): ĠFlow < T > Ġ{ ĊĠĠĠ Ġreturn Ġchannel Flow Ġ{ ĊĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcollector : ĠZ i pline Flow Collector Ġ= Ġobject Ġ: ĠZ i pline Flow Collector Ġ{ ĊĠĠĠĠĠĠĠĠĠ Ġoverride Ġsuspend Ġfun Ġemit ( value : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġitem Ġ= Ġjson . decode FromString ( item Serializer , Ġvalue ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis @ channel Flow . send ( item ) ĊĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġcollector Reference Ġ= ĠZ i pline Reference ( collector ) ĊĠĠĠĠĠĠĠ Ġreference Flow . collect Json ( collector Reference ) ĊĠĠĠĠĠĠĠ Ġthis @ channel Flow . close () ĊĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠ Ġreference Flow Reference . close () ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġfun Ġtake (): ĠFlow < T > Ġ{ ĊĠĠĠ Ġreturn Ġwhen Ġ( val Ġreference Flow Ġ= Ġreference Flow Reference . take ()) Ġ{ ĊĠĠĠĠĠ Ġis ĠReal Reference Flow < *> Ġ-> Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠIf Ġit 's Ġa ĠReal Reference Flow , Ġthen Ġthe Ġinstance Ġdidn 't Ġpass Ġthrough ĠZ i pline . ĠDon 't Ġattempt ĊĠĠĠĠĠĠĠ Ġ// Ġserialization Ġboth Ġbecause Ġit 's Ġunnecessary , Ġand Ġbecause Ġthe Ġserializer Ġisn 't Ġconnected . ĊĠĠĠĠĠĠĠ Ġreference Flow . flow Ġas ĠFlow < T > ĊĠĠĠĠĠ Ġ} ĊĠĠĠĠĠ Ġelse Ġ-> Ġget De coding Flow ( reference Flow ) ĊĠĠĠ Ġ} ĊĠ Ġ} Ċ } Ċ Ċ fun Ġ< T > ĠFlow < T >. as Flow Reference (): ĠFlow Reference < T > Ġ{ ĊĠ Ġval Ġserializer Ġ= ĠDeferred Serializer < T >() ĊĠ Ġval Ġreference Flow Ġ= ĠReal Reference Flow ( this , Ġserializer ) ĊĠ Ġval Ġz i pline Reference Ġ= ĠZ i pline Reference < Reference Flow >( reference Flow ) ĊĠ Ġreturn ĠFlow Reference ( zi pline Reference , Ġserializer ) Ċ } Ċ Ċ // ĠZ i pline Ġcan Ġonly Ġbridge Ġinterfaces , Ġnot Ġimplementations , Ġso Ġsplit Ġthis Ġin Ġtwo . Ċ internal Ġinterface ĠReference Flow Ġ: ĠZ i pline Service Ġ{ ĊĠ Ġsuspend Ġfun Ġcollect Json ( collector Reference : ĠZ i pline Reference < Zi pline Flow Collector >) Ċ } Ċ Ċ private Ġclass ĠReal Reference Flow < T >( ĊĠ Ġval Ġflow : ĠFlow < T >, ĊĠ Ġprivate Ġval Ġserializer : ĠK Serializer < T >, Ċ ) Ġ: ĠReference Flow Ġ{ ĊĠ Ġoverride Ġsuspend Ġfun Ġcollect Json ( ĊĠĠĠ Ġcollector Reference : ĠZ i pline Reference < Zi pline Flow Collector > ĊĠ Ġ) Ġ{ ĊĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠ Ġval Ġcollector Ġ= Ġcollector Reference . take () ĊĠĠĠĠĠ Ġflow . collect Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġvalue Ġ= Ġjson . encode ToString ( serializer , Ġit ) ĊĠĠĠĠĠĠĠ Ġcollector . emit ( value ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠ Ġcollector Reference . close () ĊĠĠĠ Ġ} ĊĠ Ġ} Ċ } Ċ Ċ internal Ġclass ĠFlow Reference Serializer < T >( ĊĠ Ġprivate Ġval Ġz i pline Reference Serializer : ĠK Serializer < Zi pline Reference < Reference Flow >>, ĊĠ Ġprivate Ġval Ġitem Serializer : ĠK Serializer < T >, Ċ ) Ġ: ĠK Serializer < Flow Reference < T >> Ġ{ ĊĠ Ġoverride Ġval Ġdescriptor Ġget () Ġ= Ġz i pline Reference Serializer . descriptor ĊĊĠ Ġoverride Ġfun Ġdeserialize ( decoder : ĠDecoder ): ĠFlow Reference < T > Ġ{ ĊĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠ Ġval Ġz i pline Reference Ġ= Ġz i pline Reference Serializer . deserialize ( decoder ) ĊĠĠĠ Ġreturn ĠFlow Reference ( zi pline Reference , Ġitem Serializer ) ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġserialize ( encoder : ĠEncoder , Ġvalue : ĠFlow Reference < T >) Ġ{ ĊĠĠĠ Ġval Ġitem Serializer Ġ= Ġvalue . item Serializer ĊĠĠĠ Ġif Ġ( item Serializer Ġis ĠDeferred Serializer < T >) Ġ{ ĊĠĠĠĠĠ Ġitem Serializer . delegate Ġ= Ġthis . item Serializer ĊĠĠĠ Ġ} ĊĠĠĠ Ġz i pline Reference Serializer . serialize ( encoder , Ġvalue . reference Flow Reference ) ĊĠ Ġ} Ċ } Ċ Ċ internal Ġinterface ĠZ i pline Flow Collector Ġ: ĠZ i pline Service Ġ{ ĊĠ Ġsuspend Ġfun Ġemit ( value : ĠString ) Ċ } Ċ Ċ /** ĠThis Ġ[ K Serializer ] Ġcan 't Ġdecode Ġanything Ġuntil Ġ[ delegate ] Ġis Ġset . Ġ*/ Ċ private Ġclass ĠDeferred Serializer < T >: ĠK Serializer < T > Ġ{ ĊĠ Ġvar Ġdelegate : ĠK Serializer < T >? Ġ= Ġnull ĊĊĠ Ġoverride Ġval Ġdescriptor : ĠSerial Descriptor ĊĠĠĠ Ġget () Ġ{ ĊĠĠĠĠĠ Ġval Ġdelegate Ġ= Ġthis . delegate Ġ?: Ġthrow ĠIllegalStateException (" no Ġdelegate Ġset ") ĊĠĠĠĠĠ Ġreturn Ġdelegate . descriptor ĊĠĠĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġdeserialize ( decoder : ĠDecoder ): ĠT Ġ{ ĊĠĠĠ Ġval Ġdelegate Ġ= Ġthis . delegate Ġ?: Ġthrow ĠIllegalStateException (" no Ġdelegate Ġset ") ĊĠĠĠ Ġreturn Ġdelegate . deserialize ( decoder ) ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġserialize ( encoder : ĠEncoder , Ġvalue : ĠT ) Ġ{ ĊĠĠĠ Ġval Ġdelegate Ġ= Ġthis . delegate Ġ?: Ġthrow ĠIllegalStateException (" not Ġconnected ") ĊĠĠĠ Ġdelegate . serialize ( encoder , Ġvalue ) ĊĠ Ġ} Ċ } Ċ Ċ private Ġval Ġjson Ġ= ĠJson Ġ{ ĊĠ Ġuse Array Pol ym orphism Ġ= Ġtrue Ċ } Ċ