package Ġfund . cyber . mark ets . connector Ċ Ċ import Ġfund . cyber . mark ets . common . model . Token Ċ import Ġfund . cyber . mark ets . common . model . Tokens Pair Ċ import Ġfund . cyber . mark ets . connector . configuration . TRA DES _ TOPIC _ PREFIX Ċ import Ġinfo . bitr ich . x change stream . core . Product Subscription Ċ import Ġinfo . bitr ich . x change stream . core . Streaming Exchange Ċ import Ġio . micro meter . core . instrument . Meter Registry Ċ import Ġio . reactivex . dis pos ables . Disposable Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġorg . springframework . kafka . core . Kafka Template Ċ Ċ abstract Ġclass ĠAbstract X change Connector Ġ: ĠConnector Ġ{ ĊĠĠĠ Ġval Ġlog Ġ= ĠLoggerFactory . getLogger ( java Class ) !! ĊĠĠĠ Ġlateinit Ġvar Ġmonitoring : ĠMeter Registry ĊĊĠĠĠ Ġlateinit Ġvar Ġexchange : ĠStreaming Exchange ĊĠĠĠ Ġval Ġexchange Name Ġby Ġlazy Ġ{ Ġexchange . exchange Specification . exchange Name . toUpperCase () Ġ} ĊĠĠĠ Ġval Ġexchange Tokens Pairs Ġby Ġlazy Ġ{ Ġexchange . exchange Symbols Ġ} ĊĠĠĠ Ġval Ġsubscriptions Ġ= Ġmutable ListOf < Disposable >() ĊĊĠĠĠ Ġval Ġtr ades Topic Name Ġby Ġlazy Ġ{ ĠTR A DES _ TOPIC _ PREFIX Ġ+ Ġexchange Name Ġ} ĊĠĠĠ Ġlateinit Ġvar Ġkafka Template : ĠKafka Template < String , ĠAny > ĊĊĠĠĠ Ġoverride Ġfun Ġconnect () Ġ{ ĊĠĠĠĠĠĠĠ Ġexchange . connect ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġbuild Subscription () ĊĠĠĠĠĠĠĠ Ġ). blocking Await () ĊĠĠĠĠĠĠĠ Ġlog . info (" Connected Ġto Ġthe Ġ$ exchange Name Ġexchange ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġabstract Ġfun Ġbuild Subscription (): ĠProduct Subscription ĊĊĠĠĠ Ġoverride Ġfun Ġdisconnect () Ġ{ ĊĠĠĠĠĠĠĠ Ġsubscriptions . forEach Ġ{ Ġsubscription Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsubscription . dispose () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġexchange . disconnect (). subscribe Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Disconnected Ġfrom Ġthe Ġ$ exchange Name Ġexchange ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġis Alive (): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġ// todo : Ġinvestigate Ġhow Ġto Ġcheck Ġhealth Ġof Ġa Ġexchange ĊĠĠĠĠĠĠĠ Ġreturn Ġexchange . is Alive ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġupdate Tokens Pairs () Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . warn (" Update Ġtokens Ġpair Ġfor Ġ$ exchange Name Ġexchange Ġnot Ġimplemented Ġyet ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġget Tokens Pairs (): ĠSet < Tokens Pair > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġpairs Ġ= Ġmutable Set Of < Tokens Pair >() ĊĊĠĠĠĠĠĠĠ Ġexchange Tokens Pairs . forEach Ġ{ Ġtokens Pair Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpairs . add ( Tokens Pair ( tokens Pair . base . currency Code , Ġtokens Pair . counter . currency Code )) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġpairs ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġget Tokens (): ĠSet < Token > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtokens Ġ= Ġmutable Set Of < Token >() ĊĊĠĠĠĠĠĠĠ Ġexchange Tokens Pairs . forEach Ġ{ Ġtokens Pair Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtokens . add ( Token ( tokens Pair . base . displayName , Ġtokens Pair . base . currency Code )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtokens . add ( Token ( tokens Pair . counter . displayName , Ġtokens Pair . counter . currency Code )) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġtokens ĊĠĠĠ Ġ} Ċ }