import { Logger } from 'log4js' ; import { LogFactory } from '../../factory/log-factory' ; import { RabbitManagerExchangeInterface } from '../../interface/rabbit-manager/rabbit-manager-exchange-interface' ; import { RabbitManagerHealthInterface } from '../../interface/rabbit-manager/rabbit-manager-health-interface' ; import { RabbitManagerQueueInterface } from '../../interface/rabbit-manager/rabbit-manager-queue-interface' ; import { RabbitManagerVhostInterface } from '../../interface/rabbit-manager/rabbit-manager-vhost-interface' ; import { RMChannel } from '../../model/rabbit-manager/rabbit-manager-channel-namespace' ; import { RMConnect } from '../../model/rabbit-manager/rabbit-manager-connect-namespace' ; import { RMExchange } from '../../model/rabbit-manager/rabbit-manager-exchange-namespace' ; import { RMHealth } from '../../model/rabbit-manager/rabbit-manager-health-namespace' ; import { RMVhost } from '../../model/rabbit-manager/rabbit-manager-vhost-namespace' ; import { RabbitManager } from '../../model/rabbit-manager/rebbit-manager-namespace' ; import { ApiRequestUtil } from '../../util/api-request-util' ; import { RabbitManagerResponseUtil } from '../../util/api-response/rabbit-manager-response-util' ; import { EnvLoaderUtil } from '../../util/env-loader-util' ; export class RabbitManagerService implements RabbitManagerHealthInterface , RabbitManagerVhostInterface , RabbitManagerExchangeInterface , RabbitManagerQueueInterface { private readonly logger : Logger ; private readonly managerApi : string | undefined ; private readonly instanceName : string ; private readonly apiRequester : ApiRequestUtil ; param { string } name param { string } managerApi param { string } auth constructor ( name , managerApi ? , auth ? ) { this . instanceName = name ; this . logger = LogFactory . create ( `${ this . instanceName }-manager-service` ) ; this . managerApi = managerApi ; if ( ! this . managerApi ) { this . managerApi = EnvLoaderUtil . getInstance ( ) . getPublicConfig ( ) . managerApiUrl ; } let currentAuth = auth ; if ( ! currentAuth ) { currentAuth = EnvLoaderUtil . getInstance ( ) . getPublicConfig ( ) . managerAuth ; } if ( ! this . managerApi || ! currentAuth ) { throw new Error ( 'Must Specific Manager ApiUrl And Auth, You Can Defined At Env Or Params' ) ; } this . apiRequester = new ApiRequestUtil ( new RabbitManagerResponseUtil ( this . logger ) , { 'Authorization' : `Basic ${ Buffer . from ( currentAuth ) . toString ( 'base64' ) }` , } ) ; } return { Promise < void > } public async ready ( ) { return await this . checkAlarms ( ) !== false ; } return { Promise < any > } async checkAlarms ( ) { return this . apiRequester . getRequest ( `${ this . managerApi }/health/checks/alarms` ) ; } param { number } leftTime param { "days" | "weeks" | "months" | "years" } unit return { Promise < any > } async checkCertificate ( leftTime , unit ) { return this . apiRequester . getRequest ( `${ this . managerApi }/health/checks/certificate-expiration/${ leftTime }/${ unit }` ) ; } return { Promise < any > } async checkLocalAlarms ( ) { return this . apiRequester . getRequest ( `${ this . managerApi }/health/checks/local-alarms` ) ; } return { Promise < HealthCheckBaseRs | boolean > } async checkNodeMirrorSyncCritical ( ) { return this . apiRequester . getRequest ( `${ this . managerApi }/health/checks/node-is-mirror-sync-critical` ) ; } return { Promise < HealthCheckBaseRs | boolean > } async checkNodeQuorumCritical ( ) { return this . apiRequester . getRequest ( `${ this . managerApi }/health/checks/node-is-quorum-critical` ) ; } param { number } port return { Promise < HealthCheckPortRs | boolean > } async checkPortHealth ( port ) { return this . apiRequester . getRequest ( `${ this . managerApi }/health/checks/port-listener/${ port }` ) ; } param { "amqp091" | "amqp10" | "mqtt" | "stomp" | "web-mqtt" | "web-stomp" } name return { Promise < HealthCheckProtocolRs | boolean > } public async checkProtocol ( name ) { return this . apiRequester . getRequest ( `${ this . managerApi }/health/checks/protocol-listener/${ name }` ) ; } return { Promise < any > } checkVirtualHosts ( ) { return this . apiRequester . getRequest ( `${ this . managerApi }/health/checks/virtual-hosts` ) ; } param { string } name param { CreateVhostParams } params return { Promise < boolean > } async createVhost ( name , params ? ) { const rs = await this . apiRequester . putRequest ( `${ this . managerApi }/vhosts/${ name }` , params ) ; if ( ! rs ) { this . logger . warn ( `Target VHost Already Exist:${ name }` ) ; } return rs ; } param { string } name param { boolean } checkImmediately param { CreateVhostParams } params return { Promise < boolean > } async deleteVhost ( name , checkImmediately = false , params ? ) { const rs = await this . apiRequester . deleteRequest ( `${ this . managerApi }/vhosts/${ name }` , params ) ; if ( checkImmediately ) { return await this . getVhost ( name ) === false ; } return rs ; } return { Promise < > } public async getVhost ( ) ; public async getVhost ( name ) ; public async getVhost ( name ? ) { const currentName = name ? `/${ name }` : '' ; return this . apiRequester . getRequest ( `${ this . managerApi }/vhosts${ currentName }` ) ; } param { string } name return { Promise < RMConnect . VhostConnectionsStatus [ ] | boolean > } async getVhostConnection ( name ) { return this . apiRequester . getRequest ( `${ this . managerApi }/vhosts/${ name }/connections` ) ; } param { string } name param { RabbitManager . PaginationParams } options return { Promise < RMChannel . VhostChannelStatus [ ] | boolean > } async getVhostOpenChannels ( name , options ? ) { return this . apiRequester . getRequest ( `${ this . managerApi }/vhosts/${ name }/channels` , options ) ; } param { string } vhostName param { string } nodeName return { Promise < any > } async startNode ( vhostName , nodeName ) { return this . apiRequester . postRequest ( `${ this . managerApi }/vhosts/${ vhostName }/start/${ nodeName }` ) ; } param { string } vhost param { string } name param { RMExchange . CreateOptions } options return { Promise < boolean > } public async createExchange ( vhost , name , options ) { return this . apiRequester . putRequest ( `${ this . managerApi }/exchanges/${ vhost }/${ name }` , options ) ; } param { string } vhost param { string } name param { boolean } checkImmediately param { RMExchange . DeleteOptions } options return { Promise < boolean > } public async deleteExchange ( vhost , name , checkImmediately = false , options ? ) { const rs = await this . apiRequester . deleteRequest ( `${ this . managerApi }/exchanges/${ vhost }/${ name }` , options ) ; if ( checkImmediately ) { return await this . getExchange ( vhost , name ) === false ; } return rs ; } return { Promise < RMExchange . ExchangeStatusModel [ ] | RMExchange . ExchangeStatusModel | boolean > } public async getExchange ( ) ; public async getExchange ( vhost ) ; public async getExchange ( vhost , name ) ; public async getExchange ( vhost ? , name ? ) { if ( ! vhost && name ) { throw new Error ( 'Must Specific Vhost When Getting Current Exchange' ) ; } const currentVhost = vhost ? `/${ vhost }` : '' ; const currentName = name ? `/${ name }` : '' ; return this . apiRequester . getRequest ( `${ this . managerApi }/exchanges${ currentVhost }${ currentName }` ) ; } }