package Ġfund . cyber . dump . bitcoin Ċ Ċ import Ġfund . cyber . cassandra . bitcoin . model . C ql Bitcoin Block Ċ import Ġfund . cyber . cassandra . bitcoin . model . C ql Bitcoin Contract M ined Block Ċ import Ġfund . cyber . cassandra . bitcoin . repository . Bitcoin Block Repository Ċ import Ġfund . cyber . cassandra . bitcoin . repository . Bitcoin Contract M ined Block Repository Ċ import Ġfund . cyber . dump . common . to Flux Ċ import Ġfund . cyber . search . model . bitcoin . Bitcoin Block Ċ import Ġfund . cyber . search . model . chains . Chain Info Ċ import Ġfund . cyber . search . model . events . Pump Event Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Record Ċ import Ġorg . reactiv est ream s . Publisher Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġorg . springframework . kafka . listener . Batch Message Listener Ċ import Ġreactor . core . publisher . Flux Ċ import Ġreactor . core . publisher . Mono ĊĊ Ċ class ĠBlock Dump Process ( ĊĠĠĠ Ġprivate Ġval Ġblock Repository : ĠBitcoin Block Repository , ĊĠĠĠ Ġprivate Ġval Ġcontract M ined Block Repository : ĠBitcoin Contract M ined Block Repository , ĊĠĠĠ Ġprivate Ġval Ġchain Info : ĠChain Info Ċ ) Ġ: ĠBatch Message Listener < Pump Event , ĠBitcoin Block > Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġlog Ġ= ĠLoggerFactory . getLogger ( Batch Message Listener :: class . java ) ĊĊĠĠĠ Ġoverride Ġfun Ġon Message ( records : ĠList < Consumer Record < Pump Event , ĠBitcoin Block >>) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġlog . info (" Dump ing Ġbatch Ġof Ġ${ records . size } Ġ${ chain Info . name } Ġblocks Ġfrom Ġoffset Ġ${ records . first (). offset () }") ĊĊĠĠĠĠĠĠĠ Ġrecords . to Flux Ġ{ Ġevent , Ġblock Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn @ to Flux Ġwhen Ġ( event ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠP ump Event . NEW _ BLOCK Ġ-> Ġblock . to New Block Publisher () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠP ump Event . NEW _ POOL _ TX Ġ-> ĠMono . empty () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠP ump Event . DROP PED _ BLOCK Ġ-> Ġblock . to Drop Block Publisher () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}. collect List (). block () ĊĊĠĠĠĠĠĠĠ Ġlog . info (" Finish Ġdump Ġbatch Ġof Ġ${ records . size } Ġ${ chain Info . name } Ġblocks Ġfrom " Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ" Ġoffset Ġ${ records . first (). offset () }") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠBitcoin Block . to New Block Publisher (): ĠPublisher < Any > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġsave Block Mono Ġ= Ġblock Repository . save ( C ql Bitcoin Block ( this )) ĊĠĠĠĠĠĠĠ Ġval Ġsave Contract Block Mono Ġ= Ġcontract M ined Block Repository . save ( C ql Bitcoin Contract M ined Block ( this )) ĊĊĠĠĠĠĠĠĠ Ġreturn ĠFlux . merge ( save Block Mono , Ġsave Contract Block Mono ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠBitcoin Block . to Drop Block Publisher (): ĠPublisher < Any > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġdelete Block Mono Ġ= Ġblock Repository . delete ( C ql Bitcoin Block ( this )) ĊĠĠĠĠĠĠĠ Ġval Ġdelete Contract Block Mono Ġ= Ġcontract M ined Block Repository . delete ( C ql Bitcoin Contract M ined Block ( this )) ĊĊĠĠĠĠĠĠĠ Ġreturn ĠFlux . merge ( delete Block Mono , Ġdelete Contract Block Mono ) ĊĠĠĠ Ġ} Ċ } Ċ