import amqp from 'amqplib' ; import { Options , Replies } from 'amqplib/properties' ; import { Logger } from 'log4js' ; import { LogFactory } from '../../factory/log-factory' ; import { AmqpExchangeInterface } from '../../interface/amqp/amqp-exchange-interface' ; import { AmqpExtraInterface } from '../../interface/amqp/amqp-extra-interface' ; import { AmqpMessageInterface } from '../../interface/amqp/amqp-message-interface' ; import { AmqpQueueInterface } from '../../interface/amqp/amqp-queue-interface' ; import { AmqpChannelNamespace } from '../../model/amqp/amqp-channel-namespace' ; import { AmqpOriginalConsumerWrapper } from '../../wrapper/amqp-original-consumer-wrapper' ; since date author Luminous(BGLuminous)
  export class AmqpChannelService implements AmqpQueueInterface , AmqpExchangeInterface , AmqpMessageInterface , AmqpExtraInterface { public readonly logger : Logger ; public readonly currentChannelInstance : amqp . Channel | amqp . ConfirmChannel ; public readonly instanceName : string ; private originalConsumerPool : AmqpOriginalConsumerWrapper [ ] = [ ] ; private consumerPosition = - 1 ; param { string } name param { amqp . Channel | amqp . ConfirmChannel } channel constructor ( name , channel ) { this . instanceName = name ; this . logger = LogFactory . create ( name . substr ( 0 , this . instanceName . lastIndexOf ( '-' ) ) ) ; this . logger . info ( `Creating Channel Service ${ this . instanceName }` ) ; if ( ! channel ) { throw new Error ( 'Channel Service Inject Error, Current Amqp Channel Undefined' ) ; } this . currentChannelInstance = channel ; this . addDefaultListener ( ) ; } private private addDefaultListener ( ) { this . addErrorListener ( ( err ) => { this . logger . error ( `Amqp Channel(${ this . instanceName }) Got Error: ${ err } ${ JSON . stringify ( err ) }` ) ; } ) ; this . addCloseListener ( ( ) => { this . logger . error ( `Amqp Channel(${ this . instanceName }) Closed` ) ; } ) ; } param process public addCloseListener ( process ) { this . currentChannelInstance ?. on ( 'close' , ( ) => { process ( ) ; } ) ; } param process public addErrorListener ( process ) { this . currentChannelInstance ?. on ( 'error' , err => process ( err ) ) ; } param process public addReturnListener ( process ) { this . currentChannelInstance ?. on ( 'return' , err => process ( err ) ) ; } param process public addDrainListener ( process ) { this . currentChannelInstance ?. on ( 'drain' , ( ) => process ( ) ) ; } param config public async initMetaConfigure ( config ) { if ( config . queueList && config . queueList . length !== 0 ) { await this . initQueueConfigure ( config . queueList ) ; } if ( config . exchangeList && config . exchangeList . length !== 0 ) { await this . initExchangeConfigure ( config . exchangeList ) ; } if ( config . bindList && config . bindList . length !== 0 ) { await this . initBindConfigure ( config . bindList ) ; } } param config param cleanMode public async initQueueConfigure ( config , cleanMode = false ) { try { for ( let amqpQueue of config ) { if ( cleanMode ) { await this . currentChannelInstance . deleteQueue ( amqpQueue . name , amqpQueue . cleanOption ) ; } await this . currentChannelInstance . assertQueue ( amqpQueue . name , amqpQueue . options ) ; } } catch ( err ) { this . logger . error ( `${ this . instanceName } initQueueConfigure() Got Error: ${ JSON . stringify ( err ) }` ) ; throw err ; } } param config param cleanMode public async initExchangeConfigure ( config , cleanMode = false ) { try { for ( let amqpExchange of config ) { if ( cleanMode ) { await this . currentChannelInstance . deleteExchange ( amqpExchange . name , amqpExchange . cleanOption ) ; } await this . currentChannelInstance . assertExchange ( amqpExchange . name , amqpExchange . type , amqpExchange . options ) ; } } catch ( err ) { this . logger . error ( `${ this . instanceName } initExchangeConfigure() Got Error: ${ JSON . stringify ( err ) }` ) ; throw err ; } } param config public async initBindConfigure ( config ) { try { for ( let amqpBind of config ) { if ( amqpBind . type === 'qte' ) { await this . bindQueueToExchange ( amqpBind . fromSourceName , amqpBind . targetSourceName , amqpBind . key , amqpBind . options ) ; } else if ( amqpBind . type === 'ete' ) { await this . bindExchangeToExchange ( amqpBind . fromSourceName , amqpBind . targetSourceName , amqpBind . key , amqpBind . options ) ; } } } catch ( err ) { this . logger . error ( `${ this . instanceName } initBindConfigure() Got Error: ${ JSON . stringify ( err ) }` ) ; throw err ; } } public async ackAllMessage ( ) { await this . currentChannelInstance . ackAll ( ) ; } public async ackMessage ( message , allUp ? ) { await this . currentChannelInstance . ack ( message , allUp ) ; } public consume ( queue , processor , options ? ) { const consumerName = options ?. consumerTag || this . getNextConsumerName ( ) ; const originalConsumer = new AmqpOriginalConsumerWrapper ( consumerName , this , queue , processor , options ) ; this . originalConsumerPool . push ( originalConsumer ) ; return originalConsumer ; } public async getCurrentMessage ( queueName , options ? ) { return this . currentChannelInstance . get ( queueName , options ) ; } public async killConsume ( consumerName ) { const targetConsumer = this . originalConsumerPool . find ( element => element . consumerName === consumerName ) ; if ( targetConsumer ) { await targetConsumer . kill ( ) ; this . originalConsumerPool = this . originalConsumerPool . filter ( element => element . consumerName !== consumerName ) ; return true ; } this . logger . error ( `Can't Kill Unknown Consumer ${ consumerName }` ) ; return false ; } public async nackAllMessage ( reQueue ? ) { this . currentChannelInstance . nackAll ( reQueue ) ; } public async recover ( ) { return this . currentChannelInstance . recover ( ) ; } public async rejectMessage ( message , allUp ? , reQueue ? ) { try { this . currentChannelInstance . reject ( message , reQueue ) ; } catch ( err ) { this . currentChannelInstance . nack ( message , allUp , reQueue ) ; } } public async sendMessageToExchange ( exchangeName , key , content , options ? ) { return this . currentChannelInstance . publish ( exchangeName , key , content , options ) ; } public async sendMessageToQueue ( queueName , content , options ? ) { return this . currentChannelInstance . sendToQueue ( queueName , content , options ) ; } public async setPrefetchCount ( count , global ? ) { return this . currentChannelInstance . prefetch ( count , global ) ; } public async bindExchangeToExchange ( target , from , key , args ? ) { return this . currentChannelInstance . bindExchange ( target , from , key , args ) ; } public async createExchangeIfNotExist ( name , type , options ? ) { return this . currentChannelInstance . assertExchange ( name , type , options ) ; } public async deleteExchange ( name , options ? ) { return this . currentChannelInstance . deleteExchange ( name , options ) ; } public async getExchangeStatus ( name ) { return this . currentChannelInstance . checkExchange ( name ) ; } public async unbindExchangeToExchange ( target , from , key , args ? ) { return this . currentChannelInstance . unbindExchange ( target , from , key , args ) ; } public async bindQueueToExchange ( name , exchange , bindKey , args ? ) { return this . currentChannelInstance . bindQueue ( name , exchange , bindKey , args ) ; } public async createQueueIfNotExist ( name , options ? ) { return this . currentChannelInstance . assertQueue ( name , options ) ; } public async deleteQueue ( name , options ? ) { return this . currentChannelInstance . deleteQueue ( name , options ) ; } public async getQueueStatus ( name ) { return this . currentChannelInstance . checkQueue ( name ) ; } public async purgeQueue ( name ) { this . logger . warn ( `Purging Queue ${ name }` ) ; return this . currentChannelInstance . purgeQueue ( name ) ; } public async unbindQueueToExchange ( name , exchange , bindKey , args ? ) { return this . currentChannelInstance . unbindQueue ( name , exchange , bindKey , args ) ; } public async close ( ) { await Promise . all ( this . originalConsumerPool . map ( element => element . kill ( ) ) ) ; await this . currentChannelInstance . close ( ) ; this . logger . warn ( `Killed Channel ${ this . instanceName }` ) ; } private getNextConsumerName ( ) { this . consumerPosition ++ ; return `${ this . instanceName }-consumer-${ this . consumerPosition }` ; } }