package Ġk af kas noop . avro Ċ Ċ import Ġcom . google . gson . JsonObject Ċ import Ġorg . apache . avro . generic . Generic Data Ċ import Ġorg . apache . avro . generic . Generic Datum Reader Ċ import Ġorg . apache . avro . io . Datum Reader Ċ import Ġorg . apache . avro . io . Decoder Factory Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġjava . nio . ByteBuffer Ċ Ċ /** Ċ Ġ* ĠDeserial iser Ġwith Ġsupport Ġfor ĠAV RO ĠEn velo pes Ċ Ġ*/ Ċ class ĠEn velope Deserial iser ( ĊĠĠĠ Ġval Ġschema Registry : ĠSchema Registry , ĊĠĠĠ Ġval Ġav ro Deserial iser : ĠDeserial iser Ġ= ĠDeserial iser ( schema Registry ) Ċ ) Ġ{ ĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠLoggerFactory . getLogger ( Envelope Deserial iser :: class . java ) ĊĠĠĠ Ġ} ĊĠĠĠ Ġval Ġdecoder Factory : ĠDecoder Factory Ġ= ĠDecoder Factory . get () ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠDecode Ġ[ msg ] Ġusing Ġschema Ġidentified Ġby Ġthe Ġvalue Ġof Ġ[ schema Name Field ] Ġand Ġbinary ĊĠĠĠĠ Ġ* Ġfrom Ġthe Ġ[ message Field ] Ġvalue . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġdecode ( ĊĠĠĠĠĠĠĠ Ġmsg : ĠByteArray , ĊĠĠĠĠĠĠĠ Ġenvelope Schema Name : ĠString , ĊĠĠĠĠĠĠĠ Ġschema Name Field : ĠString , ĊĠĠĠĠĠĠĠ Ġmessage Field : ĠString ĊĠĠĠ Ġ): ĠJsonObject Ġ{ ĊĠĠĠĠĠĠĠ Ġcheck ( envelope Schema Name . isNotBlank ()) Ġ{ Ġ" envelope Schema Name Ġmust Ġnot Ġbe Ġblank " Ġ} ĊĠĠĠĠĠĠĠ Ġcheck ( schema Name Field . isNotBlank ()) Ġ{ Ġ" schema Name Field Ġmust Ġnot Ġbe Ġblank " Ġ} ĊĠĠĠĠĠĠĠ Ġcheck ( message Field . isNotBlank ()) Ġ{ Ġ" message Field Ġmust Ġnot Ġbe Ġblank " Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġenvelope Ġ= Ġdecode Envelope ( msg , Ġenvelope Schema Name ) ĊĠĠĠĠĠĠĠ Ġval Ġschema Name Ġ= Ġenvelope . getString ( schema Name Field ) ĊĠĠĠĠĠĠĠ Ġval Ġbytes Ġ= Ġenvelope . get ByteArray ( message Field ) ĊĊĠĠĠĠĠĠĠ Ġreturn Ġav ro Deserial iser . decode ( bytes , Ġschema Name ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠDecode Ġ[ msg ] Ġusing Ġschema Ġidentified Ġby Ġthe Ġvalue Ġof Ġ[ schema F inger Print Field ] Ġand Ġbinary ĊĠĠĠĠ Ġ* Ġfrom Ġthe Ġ[ message Field ] Ġvalue . ĊĠĠĠĠ Ġ* ĠIf Ġ[ schema F inger Print Al gor itm Field ] Ġis Ġprovided , Ġfingerprint Ġalgorithm Ġis Ġpart Ġof Ġthe Ġmessage ĊĠĠĠĠ Ġ* Ġenvelope , Ġotherwise Ġ[ schema F inger Print Algorithm ] Ġwill Ġbe Ġused . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġdecode From F inger Print ( ĊĠĠĠĠĠĠĠ Ġmsg : ĠByteArray , ĊĠĠĠĠĠĠĠ Ġenvelope Schema Name : ĠString , ĊĠĠĠĠĠĠĠ Ġschema F inger Print Field : ĠString , ĊĠĠĠĠĠĠĠ Ġmessage Field : ĠString , ĊĠĠĠĠĠĠĠ Ġschema F inger Print Al gor itm Field : ĠString ? Ġ= Ġnull , ĊĠĠĠĠĠĠĠ Ġschema F inger Print Algorithm : ĠString Ġ= Ġ" SHA - 2 5 6 " ĊĠĠĠ Ġ): ĠJsonObject Ġ{ ĊĠĠĠĠĠĠĠ Ġcheck ( envelope Schema Name . isNotBlank ()) Ġ{ Ġ" envelope Schema Name Ġmust Ġnot Ġbe Ġblank " Ġ} ĊĠĠĠĠĠĠĠ Ġcheck ( schema F inger Print Field . isNotBlank ()) Ġ{ Ġ" schema F inger Print Field Ġmust Ġnot Ġbe Ġblank " Ġ} ĊĠĠĠĠĠĠĠ Ġcheck ( message Field . isNotBlank ()) Ġ{ Ġ" message Field Ġmust Ġnot Ġbe Ġblank " Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġenvelope Ġ= Ġdecode Envelope ( msg , Ġenvelope Schema Name ) ĊĠĠĠĠĠĠĠ Ġval Ġschema F inger Print Ġ= Ġenvelope . get ByteArray ( schema F inger Print Field ) ĊĠĠĠĠĠĠĠ Ġval Ġbytes Ġ= Ġenvelope . get ByteArray ( message Field ) ĊĊĠĠĠĠĠĠĠ Ġval Ġfinger Print Algo Ġ= Ġif Ġ( null Ġ!= Ġschema F inger Print Al gor itm Field ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġenvelope . getString ( schema F inger Print Al gor itm Field ) Ġelse Ġschema F inger Print Algorithm ĊĊĠĠĠĠĠĠĠ Ġval Ġschema Name Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ Ġschema Registry . getBy F inger Print ( schema F inger Print , Ġfinger Print Algo )?. fullName ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġthrow ĠAv ro Serial isation Exception (" Cannot Ġfind Ġschema Ġwith Ġfingerprint : Ġ$ schema F inger Print ") ĊĠĠĠĠĠĠĠ Ġreturn Ġav ro Deserial iser . decode ( bytes , Ġschema Name ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġdecode Envelope ( msg : ĠByteArray , Ġschema Name : ĠString ): ĠGeneric Data . Record Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġschema Ġ= Ġschema Registry . get ByName ( schema Name ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġthrow ĠAv ro Serial isation Exception (" Could Ġnot Ġfind Ġschema Ġ$ schema Name ") ĊĠĠĠĠĠĠĠ Ġval Ġdatum Reader : ĠDatum Reader < Any > Ġ= ĠGeneric Datum Reader ( schema ) ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdecoder Ġ= Ġdecoder Factory . binary Decoder ( msg , Ġnull ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrecord Ġ= Ġdatum Reader . read ( null , Ġdecoder ) Ġas ĠGeneric Data . Record ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġrecord ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġe Msg Ġ= Ġ" Cannot Ġconvert Ġmsg Ġenvelope Ġwith Ġ${ schema . fullName }" ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn ("$ e Msg Ġ- Ġ${ e . message }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠAv ro Serial isation Exception ( e Msg , Ġe ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ private Ġfun ĠGeneric Data . Record . getString ( field : ĠString ): ĠString Ġ{ ĊĠĠĠ Ġreturn Ġthis . get ( field ). toString () Ċ } Ċ Ċ private Ġfun ĠGeneric Data . Record . get ByteArray ( field : ĠString ): ĠByteArray Ġ{ ĊĠĠĠ Ġval Ġrecord Ġ= Ġthis . get ( field ) ĊĠĠĠ Ġif Ġ( record Ġis ĠGeneric Data . Fixed ) ĊĠĠĠĠĠĠĠ Ġreturn Ġrecord . bytes () ĊĠĠĠ Ġelse Ġif Ġ( record Ġis ĠByteBuffer ) ĊĠĠĠĠĠĠĠ Ġreturn Ġrecord . array () ĊĠĠĠ Ġthrow ĠAv ro Serial isation Exception (" Cannot Ġconvert Ġtype Ġ$ record ") Ċ } Ċ