package Ġio . enk rypt . bolt . processors Ċ Ċ import Ġcom . mongodb . client . Mongo Collection Ċ import Ġcom . mongodb . client . model . Delete One Model Ċ import Ġcom . mongodb . client . model . Update One Model Ċ import Ġcom . mongodb . client . model . Update Options Ċ import Ġcom . mongodb . client . model . Write Model Ċ import Ġio . enk rypt . bolt . extensions . to Document Ċ import Ġio . enk rypt . bolt . kafka . processors . Mongo Processor Ċ import Ġio . enk rypt . bolt . kafka . ser des . RLP Account State Ser de Ċ import Ġmu . Kotlin Logging Ċ import Ġorg . apache . kafka . common . serialization . Ser des Ċ import Ġorg . apache . kafka . streams . Kafka Streams Ċ import Ġorg . apache . kafka . streams . KeyValue Ċ import Ġorg . apache . kafka . streams . Streams Builder Ċ import Ġorg . apache . kafka . streams . Streams Config Ċ import Ġorg . apache . kafka . streams . k stream . Cons umed Ċ import Ġorg . apache . kafka . streams . processor . C ancellable Ċ import Ġorg . apache . kafka . streams . processor . Processor Context Ċ import Ġorg . apache . kafka . streams . processor . P unctuation Type Ċ import Ġorg . bson . Document Ċ import Ġorg . ethereum . core . Account State Ċ import Ġorg . ethereum . util . Byte Util Ċ import Ġorg . k oin . standalone . get Ċ import Ġjava . util .* Ċ Ċ /** Ċ Ġ* ĠThis Ġprocessor Ġprocesses Ġaddresses Ġbalances Ġand Ġtype Ġ( if Ġis Ġa Ġsmart Ġcontract Ġor Ġnot ). Ċ Ġ*/ Ċ class ĠAccount State Processor Ġ: ĠAbstract Base Processor () Ġ{ ĊĊĠ Ġoverride Ġval Ġid : ĠString Ġ= Ġ" account - state - processor " ĊĊĠ Ġprivate Ġval Ġkafka Props : ĠProperties Ġ= ĠProperties ( base Kafka Props ) ĊĠĠĠ Ġ. apply Ġ{ ĊĠĠĠĠĠ Ġput All ( base Kafka Props . toMap ()) ĊĠĠĠĠĠ Ġput ( Streams Config . APPLICATION _ ID _ CONFIG , Ġid ) ĊĠĠĠĠĠ Ġput ( Streams Config . NUM _ STREAM _ THREADS _ CONFIG , Ġ 1 ) ĊĠĠĠ Ġ} ĊĊĠ Ġoverride Ġval Ġlogger Ġ= ĠKotlin Logging . logger Ġ{} ĊĊĠ Ġoverride Ġfun Ġon Prepare Processor () Ġ{ ĊĠĠĠ Ġ// ĠR LP ĠAccount ĠSer de ĊĠĠĠ Ġval Ġaccount Ser de Ġ= ĠR LP Account State Ser de () ĊĊĠĠĠ Ġ// ĠCreate Ġstream Ġbuilder ĊĠĠĠ Ġval Ġbuilder Ġ= ĠStreams Builder () ĊĊĠĠĠ Ġval Ġ( Ġ_, Ġ_, Ġaccount State Ġ) Ġ= Ġapp Config . kafka . topics Config ĊĊĠĠĠ Ġbuilder ĊĠĠĠĠĠ Ġ. stream ( account State , ĠCon su med . with ( Ser des . ByteArray (), Ġaccount Ser de )) ĊĠĠĠĠĠ Ġ. map Ġ{ Ġk , Ġv Ġ-> ĠKey Value ( Byte Util . to HexString ( k ), Ġv ) Ġ} ĊĠĠĠĠĠ Ġ. process ({ Ġget < Account State Mongo Processor >() Ġ}, Ġnull ) ĊĊĠĠĠ Ġ// ĠGenerate Ġthe Ġtopology ĊĠĠĠ Ġval Ġtopology Ġ= Ġbuilder . build () ĊĊĠĠĠ Ġ// ĠCreate Ġstreams ĊĠĠĠ Ġstreams Ġ= ĠKafka Streams ( topology , Ġkafka Props ) ĊĠ Ġ} Ċ Ċ } Ċ Ċ class ĠAccount State Mongo Processor Ġ: ĠMongo Processor < String , ĠAccount State >() Ġ{ ĊĊĠ Ġprivate Ġval Ġaccounts Collection : ĠMongo Collection < Document > Ġby Ġlazy Ġ{ ĊĠĠĠ Ġmongo DB . get Collection ( config . mongo . accounts Collection ) ĊĠ Ġ} ĊĊĠ Ġoverride Ġval Ġbatch Size Ġ= Ġ 1 0 0 ĊĊĠ Ġprivate Ġvar Ġbatch Ġ= Ġmap Of < String , ĠAccount State >() ĊĠ Ġprivate Ġvar Ġscheduled Write : ĠC ancellable ? Ġ= Ġnull ĊĊĠ Ġoverride Ġfun Ġinit ( context : ĠProcessor Context ) Ġ{ ĊĠĠĠ Ġsuper . init ( context ) ĊĠĠĠ Ġthis . scheduled Write Ġ= Ġcontext . schedule ( timeout Ms , ĠP unctuation Type . WALL _ CLOCK _ TIME ) Ġ{ Ġ_ Ġ-> Ġtry To Write () Ġ} ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġprocess ( key : ĠString , Ġvalue : ĠAccount State ) Ġ{ ĊĠĠĠ Ġbatch Ġ+= Ġkey Ġto Ġvalue ĊĠĠĠ Ġif Ġ( batch . size Ġ== Ġbatch Size ) Ġ{ ĊĠĠĠĠĠ Ġtry To Write () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġprivate Ġfun Ġtry To Write () Ġ{ ĊĠĠĠ Ġif Ġ(! running Ġ|| Ġbatch . isEmpty ()) Ġ{ ĊĠĠĠĠĠ Ġreturn ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġval Ġstart Ms Ġ= ĠSystem . currentTimeMillis () ĊĊĠĠĠ Ġval Ġops Ġ= Ġbatch . toList (). map < Pair < String , ĠAccount State >, ĠWrite Model < Document >> Ġ{ Ġpair Ġ-> ĊĊĠĠĠĠĠ Ġval Ġaddress Ġ= Ġpair . first ĊĠĠĠĠĠ Ġval Ġstate Ġ= Ġpair . second ĊĊĠĠĠĠĠ Ġval Ġfilter Ġ= ĠDocument ( map Of ("_ id " Ġto Ġaddress )) ĊĠĠĠĠĠ Ġval Ġupdate Options Ġ= ĠUpdate Options (). upsert ( true ) ĊĊĠĠĠĠĠ Ġif Ġ( state . isEmpty ) Ġ{ ĊĠĠĠĠĠĠĠ ĠDelete One Model ( filter ) ĊĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġupdate Ġ= ĠDocument ( map Of ("\ $ set " Ġto Ġstate . to Document ( address ))) ĊĠĠĠĠĠĠĠ ĠUpdate One Model ( filter , Ġupdate , Ġupdate Options ) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġtry Ġ{ ĊĊĠĠĠĠĠ Ġaccounts Collection . bulk Write ( ops ) ĊĊĠĠĠĠĠ Ġcontext . commit () ĊĊĠĠĠĠĠ Ġval Ġelapsed Ms Ġ= ĠSystem . currentTimeMillis () Ġ- Ġstart Ms ĊĠĠĠĠĠ Ġlogger . debug Ġ{ Ġ"${ batch . size } Ġaccounts Ġupdated Ġin Ġ$ elapsed Ms Ġms " Ġ} ĊĊĠĠĠĠĠ Ġbatch Ġ= Ġempty Map () ĊĊĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠ Ġlogger . error Ġ{ Ġ" Failed Ġto Ġupdate Ġaccounts . Ġ$ e " Ġ} ĊĠĠĠ Ġ} ĊĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġclose () Ġ{ ĊĠĠĠ Ġrunning Ġ= Ġfalse ĊĠĠĠ Ġscheduled Write ?. cancel () ĊĠ Ġ} Ċ Ċ } Ċ