package Ġcloud . k oma . kc ui . service Ċ Ċ import Ġcloud . k oma . kc ui . Application Config Ċ import Ġcloud . k oma . kc ui . model . Connector Status Ċ import Ġio . mic ronaut . http . HttpResponse Ċ import Ġjak arta . inject . Singleton Ċ import Ġorg . reactiv est ream s . Publisher Ċ import Ġreactor . core . publisher . Flux Ċ import Ġreactor . core . publisher . Mono ĊĊ Ċ @ Singleton Ċ class ĠConnector Service ( ĊĠĠĠ Ġprivate Ġval Ġconfig : ĠApplication Config , ĊĠĠĠ Ġprivate Ġval Ġconnector Client : ĠConnector Client Ċ ) Ġ{ ĊĊĠĠĠ Ġfun Ġfetch Connectors ( cluster Name : ĠString ): ĠPublisher < List < Connector Status >> Ġ{ ĊĠĠĠĠĠĠĠ Ġconfig . clusters [ cluster Name ]?. let Ġ{ Ġurl Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġconnector Names Ġ= Ġconnector Client . fetch Connectors ( url ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġ ĠMono . from ( connector Names ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġlist Ġ-> ĠFlux . merge ( list . map Ġ{ Ġconn Ġ-> Ġfetch Connector Status ( url , Ġconn ) Ġ}). collect List () Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn ĠMono . empty () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġfetch Connector ( cluster Name : ĠString , Ġconnector Name : ĠString ): ĠPublisher < Connector Status > Ġ{ ĊĠĠĠĠĠĠĠ Ġconfig . clusters [ cluster Name ]?. let Ġ{ Ġurl Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġconnector Client . fetch Connector ( url , Ġconnector Name ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn ĠMono . empty () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġfetch Connector Status ( cluster Url : ĠString , Ġconnector Name : ĠString ): ĠPublisher < Connector Status > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġconnector Client . fetch Connector Status ( cluster Url , Ġconnector Name ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġfetch Connector Config ( cluster Name : ĠString , Ġconnector Name : ĠString ): ĠPublisher < Map < String , ĠString >> Ġ{ ĊĠĠĠĠĠĠĠ Ġconfig . clusters [ cluster Name ]?. let Ġ{ Ġurl Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġconnector Client . fetch Connector Config ( url , Ġconnector Name ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn ĠMono . empty () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġinvoke Connector Action ( cluster Name : ĠString , Ġconnector Names : ĠList < String >, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction : Ġ( cluster Url : ĠString , Ġconnector : ĠString ) Ġ-> ĠPublisher < HttpResponse < String >> ): ĠPublisher < List < Connector Status >> Ġ{ ĊĠĠĠĠĠĠĠ Ġconfig . clusters [ cluster Name ]?. let Ġ{ Ġurl Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠFlux . concat ( connector Names . map Ġ{ Ġconn Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMono . from ( action ( url , Ġconn )). flatMap Ġ{ ĠMono . from ( fetch Connector Status ( url , Ġconn )) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}). collect List () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn ĠMono . empty () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġrestart Connectors ( cluster Name : ĠString , Ġconnector Names : ĠList < String >): ĠPublisher < List < Connector Status >> Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġinvoke Connector Action ( cluster Name , Ġconnector Names , Ġconnector Client :: restart Connector ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġpause Connectors ( cluster Name : ĠString , Ġconnector Names : ĠList < String >): ĠPublisher < List < Connector Status >> Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġinvoke Connector Action ( cluster Name , Ġconnector Names , Ġconnector Client :: pause Connector ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġresume Connectors ( cluster Name : ĠString , Ġconnector Names : ĠList < String >): ĠPublisher < List < Connector Status >> Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġinvoke Connector Action ( cluster Name , Ġconnector Names , Ġconnector Client :: resume Connector ) ĊĠĠĠ Ġ} Ċ Ċ }