/** Ċ Ġ* ĠCopyright Ġ( c ) Ġ 2 0 2 0 ĠE mer ald Pay , ĠInc Ċ Ġ* ĠCopyright Ġ( c ) Ġ 2 0 2 0 ĠE TC DEV ĠGmbH Ċ Ġ* Ċ Ġ* Ġ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 . em eral dp ay . d sh ack le . proxy Ċ Ċ import Ġcom . fasterxml . jackson . databind . ObjectMapper Ċ import Ġio . em eral dp ay . d sh ack le . Global Ċ import Ġio . em eral dp ay . d sh ack le . rpc . Native Call Ċ import Ġio . em eral dp ay . d sh ack le . upstream . rp cc lient . Json Rpc Response Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġorg . springframework . stereotype . Service Ċ import Ġreactor . core . publisher . Flux Ċ import Ġreactor . core . publisher . Mono Ċ import Ġjava . util . function . Function Ċ Ċ /** Ċ Ġ* ĠWriter Ġfor ĠJSON ĠRPC Ġrequests Ċ Ġ*/ Ċ @ Service Ċ open Ġclass ĠWrite Rpc Json Ġ{ ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlog Ġ= ĠLoggerFactory . getLogger ( Write Rpc Json :: class . java ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval ĠobjectMapper : ĠObjectMapper Ġ= ĠGlobal . object Mapper ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠConvert ĠD sh ack le Ġprotobuf Ġbased Ġresponses Ġto ĠJSON ĠRPC Ġformatted Ġas Ġstrings ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġopen Ġfun ĠtoJson s ( call : ĠProxy Call ): ĠFunction < Flux < Native Call . Call Result >, ĠFlux < String >> Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠFunction Ġ{ Ġflux Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġflux ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġresponse Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(! call . ids . containsKey ( response . id )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . warn (" ID Ġwasn 't Ġrequested : Ġ${ response . id }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ flatMap ĠFlux . empty < String >() ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġjson Ġ= ĠtoJson ( call , Ġresponse ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( json Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠFlux . empty < String >() ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠFlux . just ( json ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. onError Resume Ġ{ Ġt Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( t Ġis ĠNative Call . Call Failure ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMono . just ( toJson ( call , Ġt ) !! ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMono . empty () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġopen Ġfun ĠtoJson ( call : ĠProxy Call , Ġresponse : ĠNative Call . Call Result ): ĠString ? Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġid Ġ= Ġcall . ids [ response . id ]?. let Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠJson Rpc Response . Id . from ( it ) ĊĠĠĠĠĠĠĠ Ġ} Ġ?: Ġreturn Ġnull ĊĠĠĠĠĠĠĠ Ġval Ġjson Ġ= Ġif Ġ( response . isError ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġerror Ġ= Ġresponse . error !! ĊĠĠĠĠĠĠĠĠĠĠĠ Ġerror . upstream Error ?. let Ġ{ Ġupstream Error Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠJson Rpc Response . error ( upstream Error , Ġid ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġ?: ĠJson Rpc Response . error (- 3 2 0 0 2 , Ġerror . message , Ġid ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠJson Rpc Response . ok ( response . result !! , Ġid ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn ĠobjectMapper . writeValue AsString ( json ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun ĠtoJson ( call : ĠProxy Call , Ġerror : ĠNative Call . Call Failure ): ĠString ? Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġid Ġ= Ġcall . ids [ error . id ] Ġ?: Ġreturn Ġnull ĊĠĠĠĠĠĠĠ Ġval Ġjson Ġ= ĠJson Rpc Response . error (- 3 2 0 0 3 , Ġerror . reason . message Ġ?: Ġ"", ĠJson Rpc Response . Id . from ( id )) ĊĠĠĠĠĠĠĠ Ġreturn ĠobjectMapper . writeValue AsString ( json ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠFormat Ġresponse Ġas ĠJSON ĠArray , Ġfor ĠBatch Ġrequests ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġas Array (): ĠFunction < Flux < String >, ĠFlux < String >> Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠFunction Ġ{ Ġflux Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbody Ġ= Ġflux . zip With ( Flux . concat ( Mono . just ( false ), ĠFlux . just ( true ). repeat ())) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( it . t 2 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"," Ġ+ Ġit . t 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . t 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ ĠFlux . concat ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMono . just (" [" ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbody , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMono . just (" ]") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ