import amqp , { Connection } from 'amqplib' ; import { EventEmitter , once } from 'events' ; import { TcpSocketConnectOpts } from 'net' ; import pb from 'promise-breaker' ; import { ConnectionOptions } from 'tls' ; import { URL } from 'url' ; import ChannelWrapper , { CreateChannelOpts } from './ChannelWrapper.js' ; import { wait } from './helpers.js' ; const HEARTBEAT_IN_SECONDS = 5 ; export type ConnectionUrl = string | amqp . Options . Connect | { url : string ; connectionOptions ? : AmpqConnectionOptions ; } ; export interface ConnectListener { ( arg ) : void ; } export interface ConnectFailedListener { ( arg ) : void ; } export type AmpqConnectionOptions = ( ConnectionOptions | TcpSocketConnectOpts ) & { noDelay ? : boolean ; timeout ? : number ; keepAlive ? : boolean ; keepAliveDelay ? : number ; clientProperties ? : any ; credentials ? : { mechanism : string ; username : string ; password : string ; response : ( ) => Buffer ; } | undefined ; } ; export interface AmqpConnectionManagerOptions { /** Interval to send heartbeats to broker. Defaults to 5 seconds. */ heartbeatIntervalInSeconds ? : number ; /**
     * The time to wait before trying to reconnect. If not specified, defaults
     * to `heartbeatIntervalInSeconds`.
     */ reconnectTimeInSeconds ? : number | undefined ; /**
     * `findServers` is a function that which returns one or more servers to
     * connect to. This should return either a single URL or an array of URLs.
     * This is handy when you're using a service discovery mechanism such as
     * Consul or etcd. Instead of taking a callback, this can also return a
     * Promise. Note that if this is supplied, then `urls` is ignored.
     */ findServers ? : ( ( callback : ( urls : ConnectionUrl | ConnectionUrl [ ] ) => void ) => void ) | ( ( ) => Promise < ConnectionUrl | ConnectionUrl [ ] > ) | undefined ; /** Connection options, passed as options to the amqplib.connect() method. */ connectionOptions ? : AmpqConnectionOptions ; } function neverThrows ( ) { return ( err ) => setImmediate ( ( ) => { throw new Error ( `AmqpConnectionManager - should never get here: ${ err . message }\n` + err . stack ) ; } ) ; } export interface IAmqpConnectionManager { connectionOptions ? : AmpqConnectionOptions ; heartbeatIntervalInSeconds : number ; reconnectTimeInSeconds : number ; addListener ( event , listener ) : this ; addListener ( event , listener ) : this ; addListener ( event , listener ) : this ; addListener ( event , listener ) : this ; addListener ( event , listener ) : this ; addListener ( event , listener ) : this ; listeners ( eventName ) : Function [ ] ; on ( event , listener ) : this ; on ( event , listener ) : this ; on ( event , listener ) : this ; on ( event , listener ) : this ; on ( event , listener ) : this ; on ( event , listener ) : this ; once ( event , listener ) : this ; once ( event , listener ) : this ; once ( event , listener ) : this ; once ( event , listener ) : this ; once ( event , listener ) : this ; once ( event , listener ) : this ; prependListener ( event , listener ) : this ; prependListener ( event , listener ) : this ; prependListener ( event , listener ) : this ; prependListener ( event , listener ) : this ; prependListener ( event , listener ) : this ; prependListener ( event , listener ) : this ; prependOnceListener ( event , listener ) : this ; prependOnceListener ( event , listener ) : this ; prependOnceListener ( event , listener ) : this ; prependOnceListener ( event , listener ) : this ; prependOnceListener ( event , listener ) : this ; prependOnceListener ( event , listener ) : this ; removeListener ( event , listener ) : this ; connect ( options ? ) : Promise < void > ; reconnect ( ) : void ; createChannel ( options ? ) : ChannelWrapper ; close ( ) : Promise < void > ; isConnected ( ) : boolean ; /** The current connection. */ readonly connection : Connection | undefined ; /** Returns the number of registered channels. */ readonly channelCount : number ; } export default class AmqpConnectionManager extends EventEmitter implements IAmqpConnectionManager { private _channels : ChannelWrapper [ ] ; private _currentUrl : number ; private _closed = false ; private _cancelRetriesHandler ? : ( ) => void ; private _connectPromise ? : Promise < null > ; private _currentConnection ? : Connection ; private _findServers : ( ( callback : ( urls : ConnectionUrl | ConnectionUrl [ ] ) => void ) => void ) | ( ( ) => Promise < ConnectionUrl | ConnectionUrl [ ] > ) ; private _urls ? : ConnectionUrl [ ] ; public connectionOptions : AmpqConnectionOptions | undefined ; public heartbeatIntervalInSeconds : number ; public reconnectTimeInSeconds : number ; param urls param options param options . heartbeatIntervalInSeconds param options . reconnectTimeInSeconds param options . connectionOptions param options . findServers constructor ( urls , options = { } ) { super ( ) ; if ( ! urls && ! options . findServers ) { throw new Error ( 'Must supply either `urls` or `findServers`' ) ; } this . _channels = [ ] ; this . _currentUrl = 0 ; this . connectionOptions = options . connectionOptions ; this . heartbeatIntervalInSeconds = options . heartbeatIntervalInSeconds || HEARTBEAT_IN_SECONDS ; this . reconnectTimeInSeconds = options . reconnectTimeInSeconds || this . heartbeatIntervalInSeconds ; this . setMaxListeners ( 0 ) ; this . _findServers = options . findServers || ( ( ) => Promise . resolve ( urls ) ) ; } param options param options . timeout async connect ( { timeout } = { } ) { this . _connect ( ) ; let reject ; const onConnectFailed = ( { err } ) => { if ( err . isOperational ) { reject ( err ) ; } } ; try { await Promise . race ( [ once ( this , 'connect' ) , new Promise ( ( _resolve , innerReject ) => { reject = innerReject ; this . on ( 'connectFailed' , onConnectFailed ) ; } ) , ... ( timeout ? [ wait ( timeout ) . promise . then ( ( ) => { throw new Error ( 'amqp-connection-manager: connect timeout' ) ; } ) , ] : [ ] ) , ] ) ; } finally { this . removeListener ( 'connectFailed' , onConnectFailed ) ; } } createChannel ( options = { } ) { const channel = new ChannelWrapper ( this , options ) ; this . _channels . push ( channel ) ; channel . once ( 'close' , ( ) => { this . _channels = this . _channels . filter ( ( c ) => c !== channel ) ; } ) ; return channel ; } close ( ) { if ( this . _closed ) { return Promise . resolve ( ) ; } this . _closed = true ; if ( this . _cancelRetriesHandler ) { this . _cancelRetriesHandler ( ) ; this . _cancelRetriesHandler = undefined ; } return Promise . resolve ( this . _connectPromise ) . then ( ( ) => { return Promise . all ( this . _channels . map ( ( channel ) => channel . close ( ) ) ) . catch ( function ( ) { } ) . then ( ( ) => { this . _channels = [ ] ; if ( this . _currentConnection ) { this . _currentConnection . removeAllListeners ( 'close' ) ; return this . _currentConnection . close ( ) ; } else { return null ; } } ) . then ( ( ) => { this . _currentConnection = undefined ; } ) ; } ) ; } isConnected ( ) { return ! ! this . _currentConnection ; } /** Force reconnect - noop unless connected */ reconnect ( ) { if ( this . _closed ) { throw new Error ( 'cannot reconnect after close' ) ; } if ( this . _currentConnection ) { this . _currentConnection . removeAllListeners ( ) ; this . _currentConnection . close ( ) . catch ( ( ) => { } ) . then ( ( ) => { this . _currentConnection = undefined ; this . emit ( 'disconnect' , { err : new Error ( 'forced reconnect' ) } ) ; return this . _connect ( ) ; } ) . catch ( neverThrows ) ; } } /** The current connection. */ get connection ( ) : Connection | undefined { return this . _currentConnection ; } /** Returns the number of registered channels. */ get channelCount ( ) : number { return this . _channels . length ; } private _connect ( ) { if ( this . _connectPromise ) { return this . _connectPromise ; } if ( this . _closed || this . isConnected ( ) ) { return Promise . resolve ( null ) ; } let attemptedUrl ; const result = ( this . _connectPromise = Promise . resolve ( ) . then ( ( ) => { if ( ! this . _urls || this . _currentUrl >= this . _urls . length ) { this . _currentUrl = 0 ; return pb . call ( this . _findServers , 0 , null ) ; } else { return this . _urls ; } } ) . then ( ( urls ) => { if ( Array . isArray ( urls ) ) { this . _urls = urls ; } else if ( urls ) { this . _urls = [ urls ] ; } if ( ! this . _urls || this . _urls . length === 0 ) { throw new Error ( 'amqp-connection-manager: No servers found' ) ; } const url = this . _urls [ this . _currentUrl ] ; this . _currentUrl ++ ; let connectionOptions = this . connectionOptions ; let originalUrl ; let connect ; if ( typeof url === 'object' && 'url' in url ) { originalUrl = connect = url . url ; connectionOptions = url . connectionOptions || this . connectionOptions ; } else if ( typeof url === 'string' ) { originalUrl = connect = url ; } else { originalUrl = url ; connect = { ... url , heartbeat : url . heartbeat ?? this . heartbeatIntervalInSeconds , } ; } attemptedUrl = originalUrl ; if ( typeof connect === 'string' ) { const u = new URL ( connect ) ; if ( ! u . searchParams . get ( 'heartbeat' ) ) { u . searchParams . set ( 'heartbeat' , `${ this . heartbeatIntervalInSeconds }` ) ; } connect = u . toString ( ) ; } return amqp . connect ( connect , connectionOptions ) . then ( ( connection ) => { this . _currentConnection = connection ; connection . on ( 'blocked' , ( reason ) => this . emit ( 'blocked' , { reason } ) ) ; connection . on ( 'unblocked' , ( ) => this . emit ( 'unblocked' ) ) ; connection . on ( 'error' , ( ) => { } ) ; connection . on ( 'close' , ( err ) => { this . _currentConnection = undefined ; this . emit ( 'disconnect' , { err } ) ; const handle = wait ( this . reconnectTimeInSeconds * 1000 ) ; this . _cancelRetriesHandler = handle . cancel ; handle . promise . then ( ( ) => this . _connect ( ) ) . catch ( neverThrows ) ; } ) ; this . _connectPromise = undefined ; this . emit ( 'connect' , { connection , url : originalUrl } ) ; return null ; } ) ; } ) . catch ( ( err ) => { this . emit ( 'connectFailed' , { err , url : attemptedUrl } ) ; this . _currentConnection = undefined ; this . _connectPromise = undefined ; let handle ; if ( err . name === 'OperationalError' && err . message === 'connect ETIMEDOUT' ) { handle = wait ( 0 ) ; } else { handle = wait ( this . reconnectTimeInSeconds * 1000 ) ; } this . _cancelRetriesHandler = handle . cancel ; return handle . promise . then ( ( ) => this . _connect ( ) ) ; } ) ) ; return result ; } }