import type * as amqplib from 'amqplib' ; import { Options } from 'amqplib' ; import { EventEmitter } from 'events' ; import pb from 'promise-breaker' ; import { IAmqpConnectionManager } from './AmqpConnectionManager.js' ; const MAX_MESSAGES_PER_BATCH = 1000 ; export type Channel = amqplib . ConfirmChannel | amqplib . Channel ; export type SetupFunc = ( ( channel : Channel , callback : ( error ? : Error ) => void ) => void ) | ( ( channel : Channel ) => Promise < void > ) | ( ( channel : amqplib . ConfirmChannel , callback : ( error ? : Error ) => void ) => void ) | ( ( channel : amqplib . ConfirmChannel ) => Promise < void > ) ; export interface CreateChannelOpts { /**  Name for this channel. Used for debugging. */ name ? : string ; /**
     * A function to call whenever we reconnect to the broker (and therefore create a new underlying channel.)
     * This function should either accept a callback, or return a Promise. See addSetup below
     */ setup ? : SetupFunc ; /**
     * True to create a ConfirmChannel (default). False to create a regular Channel.
     */ confirm ? : boolean ; /**
     * if true, then ChannelWrapper assumes all messages passed to publish() and sendToQueue() are plain JSON objects.
     * These will be encoded automatically before being sent.
     */ json ? : boolean ; /**
     * Default publish timeout in ms. Messages not published within the given time are rejected with a timeout error.
     */ publishTimeout ? : number ; } interface PublishMessage { type : 'publish' ; exchange : string ; routingKey : string ; content : Buffer ; options ? : amqplib . Options . Publish ; resolve : ( result : boolean ) => void ; reject : ( err : Error ) => void ; timeout ? : NodeJS . Timeout ; isTimedout : boolean ; } interface SendToQueueMessage { type : 'sendToQueue' ; queue : string ; content : Buffer ; options ? : amqplib . Options . Publish ; resolve : ( result : boolean ) => void ; reject : ( err : Error ) => void ; timeout ? : NodeJS . Timeout ; isTimedout : boolean ; } interface PublishOptions extends Options . Publish { /** Message will be rejected after timeout ms */ timeout ? : number ; } interface ConsumerOptions extends amqplib . Options . Consume { prefetch ? : number ; } interface Consumer { consumerTag : string | null ; queue : string ; onMessage : ( msg : amqplib . ConsumeMessage ) => void ; options : ConsumerOptions ; } type Message = PublishMessage | SendToQueueMessage ; const IRRECOVERABLE_ERRORS = [ 403 , 404 , 406 , 501 , 502 , 503 , 504 , 505 , 530 , 540 , 541 , ] ; /**
 * Calls to `publish()` or `sendToQueue()` work just like in amqplib, but messages are queued internally and
 * are guaranteed to be delivered.  If the underlying connection drops, ChannelWrapper will wait for a new
 * connection and continue.
 *
 * Events:
 * * `connect` - emitted every time this channel connects or reconnects.
 * * `error(err, {name})` - emitted if an error occurs setting up the channel.
 * * `drop({message, err})` - called when a JSON message was dropped because it could not be encoded.
 * * `close` - emitted when this channel closes via a call to `close()`
 *
 */ export default class ChannelWrapper extends EventEmitter { private _connectionManager : IAmqpConnectionManager ; private _json : boolean ; /** If we're in the process of creating a channel, this is a Promise which
     * will resolve when the channel is set up.  Otherwise, this is `null`.
     */ private _settingUp : Promise < void > | undefined = undefined ; private _setups : SetupFunc [ ] ; /** Queued messages, not yet sent. */ private _messages : Message [ ] = [ ] ; /** Oublished, but not yet confirmed messages. */ private _unconfirmedMessages : Message [ ] = [ ] ; /** Reason code during publish or sendtoqueue messages. */ private _irrecoverableCode : number | undefined ; /** Consumers which will be reconnected on channel errors etc. */ private _consumers : Consumer [ ] = [ ] ; /**
     * The currently connected channel.  Note that not all setup functions
     * have been run on this channel until `@_settingUp` is either null or
     * resolved.
     */ private _channel ? : Channel ; /**
     * True to create a ConfirmChannel. False to create a regular Channel.
     */ private _confirm = true ; /**
     * True if the "worker" is busy sending messages.  False if we need to
     * start the worker to get stuff done.
     */ private _working = false ; /**
     *  We kill off workers when we disconnect.  Whenever we start a new
     * worker, we bump up the `_workerNumber` - this makes it so if stale
     * workers ever do wake up, they'll know to stop working.
     */ private _workerNumber = 0 ; /**
     * True if the underlying channel has room for more messages.
     */ private _channelHasRoom = true ; /**
     * Default publish timeout
     */ private _publishTimeout ? : number ; public name ? : string ; addListener ( event , listener ) ; addListener ( event , listener ) ; addListener ( event , listener ) ; addListener ( event , listener ) ; addListener ( event , listener ) { return super . addListener ( event , listener ) ; } on ( event , listener ) ; on ( event , listener ) ; on ( event , listener ) ; on ( event , listener ) ; on ( event , listener ) { return super . on ( event , listener ) ; } once ( event , listener ) ; once ( event , listener ) ; once ( event , listener ) ; once ( event , listener ) ; once ( event , listener ) { return super . once ( event , listener ) ; } prependListener ( event , listener ) ; prependListener ( event , listener ) ; prependListener ( event , listener ) ; prependListener ( event , listener ) ; prependListener ( event , listener ) { return super . prependListener ( event , listener ) ; } prependOnceListener ( event , listener ) ; prependOnceListener ( event , listener ) ; prependOnceListener ( event , listener ) ; prependOnceListener ( event , listener ) ; prependOnceListener ( event , listener ) { return super . prependOnceListener ( event , listener ) ; } param setup param done returns addSetup ( setup , done ? ) { return pb . addCallback ( done , ( this . _settingUp || Promise . resolve ( ) ) . then ( ( ) => { this . _setups . push ( setup ) ; if ( this . _channel ) { return pb . call ( setup , this , this . _channel ) ; } else { return undefined ; } } ) ) ; } param { function } setup param { function } teardown param { function } done returns { void | Promise } removeSetup ( setup , teardown ? , done ? ) { return pb . addCallback ( done , ( ) => { this . _setups = this . _setups . filter ( ( s ) => s !== setup ) ; return ( this . _settingUp || Promise . resolve ( ) ) . then ( ( ) => this . _channel && teardown ? pb . call ( teardown , this , this . _channel ) : undefined ) ; } ) ; } param done returns waitForConnect ( done ? ) { return pb . addCallback ( done , this . _channel && ! this . _settingUp ? Promise . resolve ( ) : new Promise ( ( resolve ) => this . once ( 'connect' , resolve ) ) ) ; } publish ( exchange , routingKey , content , options ? , done ? ) { return pb . addCallback ( done , new Promise < boolean > ( ( resolve , reject ) => { const { timeout , ... opts } = options || { } ; this . _enqueueMessage ( { type : 'publish' , exchange , routingKey , content : this . _getEncodedMessage ( content ) , resolve , reject , options : opts , isTimedout : false , } , timeout || this . _publishTimeout ) ; this . _startWorker ( ) ; } ) ) ; } sendToQueue ( queue , content , options ? , done ? ) { const encodedContent = this . _getEncodedMessage ( content ) ; return pb . addCallback ( done , new Promise < boolean > ( ( resolve , reject ) => { const { timeout , ... opts } = options || { } ; this . _enqueueMessage ( { type : 'sendToQueue' , queue , content : encodedContent , resolve , reject , options : opts , isTimedout : false , } , timeout || this . _publishTimeout ) ; this . _startWorker ( ) ; } ) ) ; } private _enqueueMessage ( message , timeout ? ) { if ( timeout ) { message . timeout = setTimeout ( ( ) => { let idx = this . _messages . indexOf ( message ) ; if ( idx !== - 1 ) { this . _messages . splice ( idx , 1 ) ; } else { idx = this . _unconfirmedMessages . indexOf ( message ) ; if ( idx !== - 1 ) { this . _unconfirmedMessages . splice ( idx , 1 ) ; } } message . isTimedout = true ; message . reject ( new Error ( 'timeout' ) ) ; } , timeout ) ; } this . _messages . push ( message ) ; } param connectionManager param options param options . name param options . setup param options . json constructor ( connectionManager , options = { } ) { super ( ) ; this . _onConnect = this . _onConnect . bind ( this ) ; this . _onDisconnect = this . _onDisconnect . bind ( this ) ; this . _connectionManager = connectionManager ; this . _confirm = options . confirm ?? true ; this . name = options . name ; this . _publishTimeout = options . publishTimeout ; this . _json = options . json ?? false ; this . _setups = [ ] ; this . _consumers = [ ] ; if ( options . setup ) { this . _setups . push ( options . setup ) ; } const connection = connectionManager . connection ; if ( connection ) { this . _onConnect ( { connection } ) ; } connectionManager . on ( 'connect' , this . _onConnect ) ; connectionManager . on ( 'disconnect' , this . _onDisconnect ) ; } private async _onConnect ( { connection } ) { this . _irrecoverableCode = undefined ; try { let channel ; if ( this . _confirm ) { channel = await connection . createConfirmChannel ( ) ; } else { channel = await connection . createChannel ( ) ; } this . _channel = channel ; this . _channelHasRoom = true ; channel . on ( 'close' , ( ) => this . _onChannelClose ( channel ) ) ; channel . on ( 'drain' , ( ) => this . _onChannelDrain ( ) ) ; this . _settingUp = Promise . all ( this . _setups . map ( ( setupFn ) => pb . call ( setupFn , this , channel ) . catch ( ( err ) => { if ( err . name === 'IllegalOperationError' ) { return ; } this . emit ( 'error' , err , { name : this . name } ) ; } ) ) ) . then ( ( ) => { return Promise . all ( this . _consumers . map ( ( c ) => this . _reconnectConsumer ( c ) ) ) ; } ) . then ( ( ) => { this . _settingUp = undefined ; } ) ; await this . _settingUp ; if ( ! this . _channel ) { return ; } this . _startWorker ( ) ; this . emit ( 'connect' ) ; } catch ( err ) { this . emit ( 'error' , err , { name : this . name } ) ; this . _settingUp = undefined ; this . _channel = undefined ; } } private _onChannelClose ( channel ) { if ( this . _channel === channel ) { this . _channel = undefined ; } } /** Called whenever the channel drains. */ private _onChannelDrain ( ) { this . _channelHasRoom = true ; this . _startWorker ( ) ; } private _onDisconnect ( ex ) { this . _irrecoverableCode = ex . err instanceof Error ? ex . err . code : undefined ; this . _channel = undefined ; this . _settingUp = undefined ; this . _working = false ; } queueLength ( ) { return this . _messages . length ; } close ( ) { return Promise . resolve ( ) . then ( ( ) => { this . _working = false ; if ( this . _messages . length !== 0 ) { this . _messages . forEach ( ( message ) => { if ( message . timeout ) { clearTimeout ( message . timeout ) ; } message . reject ( new Error ( 'Channel closed' ) ) ; } ) ; } if ( this . _unconfirmedMessages . length !== 0 ) { this . _unconfirmedMessages . forEach ( ( message ) => { if ( message . timeout ) { clearTimeout ( message . timeout ) ; } message . reject ( new Error ( 'Channel closed' ) ) ; } ) ; } this . _connectionManager . removeListener ( 'connect' , this . _onConnect ) ; this . _connectionManager . removeListener ( 'disconnect' , this . _onDisconnect ) ; const answer = ( this . _channel && this . _channel . close ( ) ) || undefined ; this . _channel = undefined ; this . emit ( 'close' ) ; return answer ; } ) ; } private _shouldPublish ( ) { return ( this . _messages . length > 0 && ! this . _settingUp && ! ! this . _channel && this . _channelHasRoom ) ; } private _startWorker ( ) { if ( ! this . _working && this . _shouldPublish ( ) ) { this . _working = true ; this . _workerNumber ++ ; this . _publishQueuedMessages ( this . _workerNumber ) ; } } private _canWaitReconnection ( ) { return ! this . _irrecoverableCode || ! IRRECOVERABLE_ERRORS . includes ( this . _irrecoverableCode ) ; } private _messageResolved ( message , result ) { removeUnconfirmedMessage ( this . _unconfirmedMessages , message ) ; message . resolve ( result ) ; } private _messageRejected ( message , err ) { if ( ! this . _channel && this . _canWaitReconnection ( ) ) { removeUnconfirmedMessage ( this . _unconfirmedMessages , message ) ; this . _messages . push ( message ) ; } else { removeUnconfirmedMessage ( this . _unconfirmedMessages , message ) ; message . reject ( err ) ; } } private _getEncodedMessage ( content ) { let encodedMessage ; if ( this . _json ) { encodedMessage = Buffer . from ( JSON . stringify ( content ) ) ; } else if ( typeof content === 'string' ) { encodedMessage = Buffer . from ( content ) ; } else if ( content instanceof Buffer ) { encodedMessage = content ; } else if ( typeof content === 'object' && typeof ( content as any ) . toString === 'function' ) { encodedMessage = Buffer . from ( ( content as any ) . toString ( ) ) ; } else { console . warn ( 'amqp-connection-manager: Sending JSON message, but json option not speicifed' ) ; encodedMessage = Buffer . from ( JSON . stringify ( content ) ) ; } return encodedMessage ; } private _publishQueuedMessages ( workerNumber ) { const channel = this . _channel ; if ( ! channel || ! this . _shouldPublish ( ) || ! this . _working || workerNumber !== this . _workerNumber ) { this . _working = false ; return ; } try { let sendsLeft = MAX_MESSAGES_PER_BATCH ; while ( this . _channelHasRoom && this . _messages . length > 0 && sendsLeft > 0 ) { sendsLeft -- ; const message = this . _messages . shift ( ) ; if ( ! message ) { break ; } let thisCanSend = true ; switch ( message . type ) { case 'publish' : { if ( this . _confirm ) { this . _unconfirmedMessages . push ( message ) ; thisCanSend = this . _channelHasRoom = channel . publish ( message . exchange , message . routingKey , message . content , message . options , ( err ) => { if ( message . isTimedout ) { return ; } if ( message . timeout ) { clearTimeout ( message . timeout ) ; } if ( err ) { this . _messageRejected ( message , err ) ; } else { this . _messageResolved ( message , thisCanSend ) ; } } ) ; } else { if ( message . timeout ) { clearTimeout ( message . timeout ) ; } thisCanSend = this . _channelHasRoom = channel . publish ( message . exchange , message . routingKey , message . content , message . options ) ; message . resolve ( thisCanSend ) ; } break ; } case 'sendToQueue' : { if ( this . _confirm ) { this . _unconfirmedMessages . push ( message ) ; thisCanSend = this . _channelHasRoom = channel . sendToQueue ( message . queue , message . content , message . options , ( err ) => { if ( message . isTimedout ) { return ; } if ( message . timeout ) { clearTimeout ( message . timeout ) ; } if ( err ) { this . _messageRejected ( message , err ) ; } else { this . _messageResolved ( message , thisCanSend ) ; } } ) ; } else { if ( message . timeout ) { clearTimeout ( message . timeout ) ; } thisCanSend = this . _channelHasRoom = channel . sendToQueue ( message . queue , message . content , message . options ) ; message . resolve ( thisCanSend ) ; } break ; } default : throw new Error ( `Unhandled message type ${ ( message as any ) . type }` ) ; } } if ( this . _channelHasRoom && this . _messages . length > 0 ) { setImmediate ( ( ) => this . _publishQueuedMessages ( workerNumber ) ) ; } else { this . _working = false ; } } catch ( err ) { this . _working = false ; this . emit ( 'error' , err ) ; } } /**
     * Setup a consumer
     * This consumer will be reconnected on cancellation and channel errors.
     */ async consume ( queue , onMessage , options = { } ) { const consumer = { consumerTag : null , queue , onMessage , options , } ; this . _consumers . push ( consumer ) ; await this . _consume ( consumer ) ; } private async _consume ( consumer ) { if ( ! this . _channel ) { return ; } const { prefetch , ... options } = consumer . options ; if ( typeof prefetch === 'number' ) { this . _channel . prefetch ( prefetch , false ) ; } const { consumerTag } = await this . _channel . consume ( consumer . queue , ( msg ) => { if ( ! msg ) { consumer . consumerTag = null ; this . _reconnectConsumer ( consumer ) . catch ( ( err ) => { if ( err . isOperational && err . message . includes ( 'BasicConsume; 404' ) ) { return ; } throw err ; } ) ; return ; } consumer . onMessage ( msg ) ; } , options ) ; consumer . consumerTag = consumerTag ; } private async _reconnectConsumer ( consumer ) { if ( ! this . _consumers . includes ( consumer ) ) { return ; } await this . _consume ( consumer ) ; } /**
     * Cancel all consumers
     */ async cancelAll ( ) { const consumers = this . _consumers ; this . _consumers = [ ] ; if ( ! this . _channel ) { return ; } const channel = this . _channel ; await Promise . all ( consumers . reduce < any [ ] > ( ( acc , consumer ) => { if ( consumer . consumerTag ) { acc . push ( channel . cancel ( consumer . consumerTag ) ) ; } return acc ; } , [ ] ) ) ; } /** Send an `ack` to the underlying channel. */ ack ( message , allUpTo ? ) { this . _channel && this . _channel . ack ( message , allUpTo ) ; } /** Send an `ackAll` to the underlying channel. */ ackAll ( ) { this . _channel && this . _channel . ackAll ( ) ; } /** Send a `nack` to the underlying channel. */ nack ( message , allUpTo ? , requeue ? ) { this . _channel && this . _channel . nack ( message , allUpTo , requeue ) ; } /** Send a `nackAll` to the underlying channel. */ nackAll ( requeue ? ) { this . _channel && this . _channel . nackAll ( requeue ) ; } /** Send a `purgeQueue` to the underlying channel. */ async purgeQueue ( queue ) { if ( this . _channel ) { return await this . _channel . purgeQueue ( queue ) ; } else { throw new Error ( `Not connected.` ) ; } } /** Send a `checkQueue` to the underlying channel. */ async checkQueue ( queue ) { if ( this . _channel ) { return await this . _channel . checkQueue ( queue ) ; } else { throw new Error ( `Not connected.` ) ; } } /** Send a `assertQueue` to the underlying channel. */ async assertQueue ( queue , options ? ) { if ( this . _channel ) { return await this . _channel . assertQueue ( queue , options ) ; } else { return { queue , messageCount : 0 , consumerCount : 0 } ; } } /** Send a `bindQueue` to the underlying channel. */ async bindQueue ( queue , source , pattern , args ? ) { if ( this . _channel ) { await this . _channel . bindQueue ( queue , source , pattern , args ) ; } } /** Send a `deleteQueue` to the underlying channel. */ async deleteQueue ( queue , options ? ) { if ( this . _channel ) { return await this . _channel . deleteQueue ( queue , options ) ; } else { throw new Error ( `Not connected.` ) ; } } /** Send a `assertExchange` to the underlying channel. */ async assertExchange ( exchange , type , options ? ) { if ( this . _channel ) { return await this . _channel . assertExchange ( exchange , type , options ) ; } else { return { exchange } ; } } /** Send a `bindExchange` to the underlying channel. */ async bindExchange ( destination , source , pattern , args ? ) { if ( this . _channel ) { return await this . _channel . bindExchange ( destination , source , pattern , args ) ; } else { throw new Error ( `Not connected.` ) ; } } /** Send a `checkExchange` to the underlying channel. */ async checkExchange ( exchange ) { if ( this . _channel ) { return await this . _channel . checkExchange ( exchange ) ; } else { throw new Error ( `Not connected.` ) ; } } /** Send a `deleteExchange` to the underlying channel. */ async deleteExchange ( exchange , options ? ) { if ( this . _channel ) { return await this . _channel . deleteExchange ( exchange , options ) ; } else { throw new Error ( `Not connected.` ) ; } } /** Send a `unbindExchange` to the underlying channel. */ async unbindExchange ( destination , source , pattern , args ? ) { if ( this . _channel ) { return await this . _channel . unbindExchange ( destination , source , pattern , args ) ; } else { throw new Error ( `Not connected.` ) ; } } /** Send a `get` to the underlying channel. */ async get ( queue , options ? ) { if ( this . _channel ) { return await this . _channel . get ( queue , options ) ; } else { throw new Error ( `Not connected.` ) ; } } } function removeUnconfirmedMessage ( arr , message ) { const toRemove = arr . indexOf ( message ) ; if ( toRemove === - 1 ) { throw new Error ( `Message is not in _unconfirmedMessages!` ) ; } const removed = arr . splice ( toRemove , 1 ) ; return removed [ 0 ] ; }