import * as amqp from 'amqplib' ; import { Connection } from 'amqplib' ; import { Options } from 'amqplib/properties' ; import { Logger } from 'log4js' ; import { LogFactory } from '../../factory/log-factory' ; import { AmqpConnectInterface } from '../../interface/amqp/amqp-connect-interface' ; import { EnvLoaderUtil } from '../../util/env-loader-util' ; import { AmqpOriginalChannelWrapper } from '../../wrapper/amqp-original-channel-wrapper' ; import { AmqpOriginalConfirmChannelWrapper } from '../../wrapper/amqp-original-confirm-channel-wrapper' ; import { AmqpChannelService } from './amqp-channel-service' ; import { AmqpConfirmChannelService } from './amqp-confirm-channel-service' ; since date author Luminous(BGLuminous)
  export class AmqpConnectService implements AmqpConnectInterface { public currentConnection : Connection | undefined ; public readonly currentAmqpServerUrl : string | Options . Connect = '' ; public readonly instanceName : string ; public readonly MAX_CHANNEL_COUNT : number = 2047 ; private readonly logger : Logger ; private wrapperChannelPool : ( AmqpOriginalChannelWrapper | AmqpOriginalConfirmChannelWrapper ) [ ] = [ ] ; private serviceChannelPool : ( AmqpChannelService | AmqpConfirmChannelService ) [ ] = [ ] ; private channelPosition = - 1 ; param { string } name param { string | Options . Connect } amqpUrl constructor ( name , amqpUrl ? ) { this . instanceName = `${ name }-amqp-connect` ; this . logger = LogFactory . create ( this . instanceName ) ; if ( amqpUrl ) { this . currentAmqpServerUrl = amqpUrl ; } if ( ! this . currentAmqpServerUrl ) { const loadFromEnv = EnvLoaderUtil . getInstance ( ) . getPublicConfig ( ) . amqpServerUrl ; if ( loadFromEnv ) { this . currentAmqpServerUrl = loadFromEnv ; } } const maxChannelCount = EnvLoaderUtil . getInstance ( ) . getPublicConfig ( ) . maxChannelCount ; if ( maxChannelCount && ! isNaN ( maxChannelCount ) ) { this . MAX_CHANNEL_COUNT = maxChannelCount ; } if ( ! this . currentAmqpServerUrl ) { throw new Error ( 'Must Specific Amqp Server Url' ) ; } } param init param retry param options public async ready ( init ? , retry = 0 , options ? ) { const retryInfo = retry > 0 ? `Retry: ${ retry } times` : '' ; this . logger . debug ( `Initialize ${ this . instanceName } Creating Amqp Server: ${ this . currentAmqpServerUrl } ${ retryInfo }` ) ; try { await this . close ( ) ; this . currentConnection = await amqp . connect ( this . currentAmqpServerUrl , options , ( err ) => { this . logger . error ( `Connection(${ this . instanceName }) Connecting Server Got Error: ${ err }` ) ; throw new Error ( err ) ; } ) ; this . addDefaultListener ( ) ; this . logger . debug ( `${ this . instanceName } Connected Amqp Server` ) ; if ( ( init ) ) { init ( this ) ; } } catch ( err ) { this . logger . error ( `${ this . instanceName } Connect Got Error: ${ err . stack }` ) ; if ( retry >= 0 && EnvLoaderUtil . getInstance ( ) . getPublicConfig ( ) . autoReconnect ) { await new Promise ( r => setTimeout ( r , 5 * 1000 ) ) ; await this . ready ( init , ++ retry , options ) ; } throw err ; } } private private addDefaultListener ( ) { this . addErrorListener ( ( err ) => { this . logger . error ( `Amqp Connect(${ this . instanceName }) Got Error: ${ err } ${ JSON . stringify ( err ) }` ) ; } ) ; this . addCloseListener ( ( err ) => { this . logger . error ( `Amqp Connect(${ this . instanceName }) Closed: ${ err } ${ JSON . stringify ( err ) }` ) ; } ) ; } param process public addCloseListener ( process ) { this . currentConnection ?. on ( 'close' , err => { process ( err ) ; } ) ; } param process public addErrorListener ( process ) { this . currentConnection ?. on ( 'error' , err => process ( err ) ) ; } param { string } channelName return { Promise < void > } public async killChannel ( channelName ) { let targetChannel ; targetChannel = this . serviceChannelPool . find ( element => element . instanceName === channelName ) ; if ( targetChannel ) { this . serviceChannelPool = this . serviceChannelPool . filter ( element => element . instanceName !== channelName ) ; return true ; } targetChannel = this . wrapperChannelPool . find ( element => element . instanceName === channelName ) ; if ( targetChannel ) { await targetChannel . close ( ) ; this . wrapperChannelPool = this . wrapperChannelPool . filter ( element => element . instanceName !== channelName ) ; return true ; } this . logger . error ( `Can't Kill Unknown Channel ${ channelName }` ) ; return false ; } return { Promise < void > } async close ( init = true ) { try { if ( ! this . currentConnection ) { this . logger . warn ( 'Please ready this connection first' ) ; } await this . killWrapperChannel ( ) ; await this . killServiceChannel ( ) ; await this . currentConnection ?. close ( ) ; this . logger . warn ( `Amqp Server Instance ${ this . instanceName } Closed` ) ; } catch ( err ) { if ( ! init ) { this . logger . warn ( `Cant Close Instance ${ this . instanceName } Closed` ) ; } } } param { 'true' | 'false' | boolean } confirmChannel return { Promise < AmqpConfirmChannelService | AmqpChannelService > } public async createChannelService ( confirmChannel ) ; public async createChannelService ( confirmChannel ) ; public async createChannelService ( confirmChannel ) { if ( ! this . currentConnection ) { throw new Error ( 'Please ready this connection first' ) ; } if ( this . checkChannelCountOvered ( ) ) { return ; } if ( confirmChannel ) { const confirmChannelService = new AmqpConfirmChannelService ( this . getNextChannelName ( ) , await this . currentConnection . createConfirmChannel ( ) ) ; this . serviceChannelPool . push ( confirmChannelService ) ; return confirmChannelService ; } const channelService = new AmqpChannelService ( this . getNextChannelName ( ) , await this . currentConnection . createChannel ( ) ) ; this . serviceChannelPool . push ( channelService ) ; return channelService ; } param { 'true' | 'false' | boolean } confirmChannel return { Promise < AmqpOriginalConfirmChannelWrapper | undefined > } public async createChannelWrapper ( confirmChannel ) ; public async createChannelWrapper ( confirmChannel ) ; public async createChannelWrapper ( confirmChannel ) { if ( ! this . currentConnection ) { throw new Error ( 'Please ready this connection first' ) ; } if ( this . checkChannelCountOvered ( ) ) { return ; } if ( confirmChannel ) { const confirmChannelInstance = new AmqpOriginalConfirmChannelWrapper ( await this . currentConnection . createConfirmChannel ( ) , this . getNextChannelName ( ) ) ; this . wrapperChannelPool . push ( confirmChannelInstance ) ; return confirmChannelInstance ; } const channelInstance = new AmqpOriginalChannelWrapper ( await this . currentConnection . createChannel ( ) , this . getNextChannelName ( ) ) ; this . wrapperChannelPool . push ( channelInstance ) ; return channelInstance ; } return { Promise < void > } private private async killWrapperChannel ( ) { for ( const channelPoolElement of this . wrapperChannelPool ) { await channelPoolElement . channel . close ( ) ; this . logger . warn ( `Killed Channel ${ channelPoolElement . instanceName }` ) ; } } return { Promise < void > } private private async killServiceChannel ( ) { await Promise . all ( this . serviceChannelPool . map ( element => element . close ( ) ) ) ; } return { Promise < boolean > } protected private checkChannelCountOvered ( ) { if ( this . wrapperChannelPool . length + this . serviceChannelPool . length > this . MAX_CHANNEL_COUNT ) { this . logger . error ( `Can't Create New Channel, Active Channels Count Exceeds Total Connection Supports: MAX:${ this . MAX_CHANNEL_COUNT }` ) ; return true ; } return false ; } return { string } private private getNextChannelName ( ) { this . channelPosition ++ ; return `${ this . instanceName }-channel-${ this . channelPosition }` ; } }