package Ġfinancial . shared . streams . kafka . connector Ċ Ċ import Ġcom . fasterxml . jackson . databind . JsonNode Ċ import Ġcom . github . kit tin un f . fuel . Fuel Ċ import Ġcom . github . kit tin un f . fuel . jackson . response Object Ċ import Ġcom . github . kit tin un f . result . Result Ċ import Ġfinancial . shared . log . Logger Ċ import Ġfinancial . shared . streams . Connector Factory Ċ import Ġfinancial . shared . streams . Connector Factory . Connector Ċ import Ġfinancial . shared . streams . Stream Settings Ċ import Ġkotlinx . coroutines . delay Ċ import Ġjava . nio . file . Files Ċ import Ġjava . nio . file . Paths Ċ Ċ class ĠKafka Connector Factory ( private Ġval Ġlogger : ĠLogger , Ġprivate Ġval Ġstream Settings : ĠStream Settings ) Ġ: ĠConnector Factory Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġurl Ġ= Ġstream Settings . connector Url ĊĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġregister () Ġ{ ĊĠĠĠĠĠĠĠ Ġwaiting For Connection () ĊĊĠĠĠĠĠĠĠ Ġval Ġconnector s Ġ= Ġlist From Broker () ĊĊĠĠĠĠĠĠĠ Ġlist From Classpath () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġ! connectors . contains ( it . name ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ(_, Ġ_, Ġresult ) Ġ= ĠF uel ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. post ( path Ġ= Ġ"${ url }/ connectors ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. body ( it . data ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. append Header (" Content - Type ", Ġ" application / json ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. response String () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( result ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠResult . Success Ġ-> Ġlogger ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. logger For ( this :: class ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. info (" The Ġconnector Ġ'${ it . name }' Ġhas Ġbeen Ġregistered ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠResult . Failure Ġ-> Ġlogger ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. logger For ( this :: class ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. error (" The Ġconnector Ġ'${ it . name }' Ġhas Ġnot Ġbeen Ġregistered Ġ- Ġ(${ result . error . message }) ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġwaiting For Connection () Ġ{ ĊĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġ(_, Ġ_, Ġresult ) Ġ= ĠF uel ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. get ( path Ġ= Ġurl ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. append Header (" Content - Type ", Ġ" application / json ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. response String () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( result ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠResult . Success Ġ-> Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠResult . Failure Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . logger For ( this :: class ). info (" Waiting Ġconnection Ġwith Ġkafka Ġconnect ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 3 0 0 0 L ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġlist From Broker (): ĠList < String > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġ(_, Ġ_, Ġresult ) Ġ= ĠF uel ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. get ( path Ġ= Ġ"${ url }/ connectors ", Ġparameters Ġ= ĠlistOf (" expand " Ġto Ġ" status ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. response Object < JsonNode >() ĊĊĠĠĠĠĠĠĠ Ġreturn Ġwhen Ġ( result ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠResult . Failure Ġ-> Ġempty List () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠResult . Success Ġ-> Ġresult ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. value ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. fields () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. as Sequence () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . value [" status "] Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit [" name "]. as Text () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. toList () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġlist From Classpath (): ĠList < Connector > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠFiles ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. walk ( Paths . get ( stream Settings . connectors Path )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ ĠFiles . is Regular File ( it ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . to File () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit . name . endsWith (". json ") Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĠConnector ( name Ġ= Ġit . name . remove Suffix (". json "), Ġdata Ġ= Ġit . read Text ()) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. toList () ĊĠĠĠ Ġ} Ċ }