package Ġio . con fluent . dev x . k ly fft . web so ket Ċ Ċ import Ġcom . fasterxml . jackson . databind . ObjectMapper Ċ import Ġorg . apache . kafka . clients . consumer . Kafka Consumer Ċ import Ġorg . apache . kafka . clients . producer . Producer Record Ċ import Ġorg . springframework . beans . factory . annotation . Autowired Ċ import Ġorg . springframework . kafka . core . Consumer Factory Ċ import Ġorg . springframework . kafka . core . Kafka Template Ċ import Ġorg . springframework . stereotype . Component Ċ import Ġorg . springframework . web . socket . Close Status Ċ import Ġorg . springframework . web . socket . Text Message Ċ import Ġorg . springframework . web . socket . WebSocket Session Ċ import Ġorg . springframework . web . socket . handler . Text WebSocket Handler Ċ import Ġjava . time . Duration Ċ Ċ @ Component Ċ class ĠWebSocket Handler (@ Autowired Ġval Ġmapper : ĠObjectMapper , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Autowired Ġval Ġkafka Producer : ĠKafka Template < String , ĠString >, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Autowired Ġval Ġkafka Consumer Factory : ĠConsumer Factory < String , ĠString >) Ġ: ĠText WebSocket Handler () Ġ{ ĊĊĠ Ġprivate Ġfun ĠWebSocket Session . create Kafka Consumer ( topic : ĠString ): ĠKafka Consumer < *, Ġ*> Ġ{ ĊĠĠĠ Ġval Ġattributes Ġ= Ġthis . attributes ĊĠĠĠ Ġreturn Ġattributes . compute If Absent (" consumer ") Ġ{ ĊĠĠĠĠĠ Ġkafka Consumer Factory . create Consumer (). apply Ġ{ ĊĠĠĠĠĠĠĠ Ġsubscribe ( listOf ( topic )) ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ġas ĠKafka Consumer < *, Ġ*> ĊĠ Ġ} ĊĊĊĠ Ġoverride Ġfun Ġafter Connection Closed ( session : ĠWebSocket Session , Ġstatus : ĠClose Status ) Ġ{ ĊĠĠĠ Ġval Ġconsumer Ġ= Ġsession . attributes [" consumer "] ĊĠĠĠ Ġif Ġ( consumer Ġis ĠKafka Consumer < *, Ġ* >) Ġwith ( consumer ) Ġ{ ĊĠĠĠĠĠ Ġunsubscribe () ĊĠĠĠĠĠ Ġclose () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġhandle Text Message ( session : ĠWebSocket Session , Ġtext Message : ĠText Message ) Ġ{ ĊĊĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠ Ġval Ġdriver Ġ= Ġsession . uri !!. path . contains (" driver - ws ") ĊĊĠĠĠĠĠ Ġval Ġsink Ġ= Ġif Ġ( driver ) Ġ" driver " Ġelse Ġ" r ider " ĊĠĠĠĠĠ Ġval Ġsource Ġ= Ġif Ġ( driver ) Ġ" r ider " Ġelse Ġ" driver " ĊĊĠĠĠĠĠ Ġval Ġkafka Consumer Ġ= Ġsession . create Kafka Consumer ( source ) ĊĊĠĠĠĠĠ Ġval Ġpayload Ġ= Ġtext Message . payload ĊĠĠĠĠĠ Ġval Ġmessage Ġ= Ġmapper . read Tree ( payload ) ĊĠĠĠĠĠ Ġ// val Ġtimestamp : ĠString Ġ= Ġmessage . get (" timestamp "). as Text () ĊĠĠĠĠĠ Ġval Ġkey : ĠString Ġ= Ġif Ġ(" driver " Ġ== Ġsink ) Ġmessage . get (" driver "). as Text () Ġelse Ġmessage . get (" r ider "). as Text () ĊĊĠĠĠĠĠ Ġkafka Producer . send ( Producer Record < String , ĠString >( sink , Ġkey , Ġpayload )) ĊĊĠĠĠĠĠ Ġkafka Consumer . poll ( Duration . of Millis ( 1 0 0 )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . value () Ġas ĠString ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsession . sendMessage ( Text Message ( it )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠ Ġe . printStackTrace () ĊĠĠĠĠĠ Ġthrow Ġe ĊĠĠĠ Ġ} ĊĠ Ġ} Ċ } ĊĊĊ