/**
* @typedef {string} AmqpURI
*/
/**
* @typedef {string} ChannelName
*/
/**
* Options applied when asserting Queues.
* @typedef {Object} QueueArguments
* @property {number} messageTtl `0 <= n < 2^32` expires messages arriving in the queue after n milliseconds
* @property {number} expires `0 < n < 2^32` the queue will be destroyed after n milliseconds of disuse,
* where use means having consumers, being declared
* @property {number} maxLength sets a maximum number of messages the queue will hold
* @property {number} maxPriority makes the queue a priority queue
*/
/**
* @typedef {Object} ConnectionOptions
* @property {number} heartbeat the period of the connection heartbeat, in seconds. Defaults to `60`
*/
/**
* Common configuration for instances.
* @typedef Configuration
* @property {boolean} printMessages if true, the messages will be printed if logging is enabled. Defaults to `true`.
* @property {number} maxMessageLengthToPrint if the messages are to printed, this is the max
* number of bytes that a message can be to be printed. Defaults to `257`.
* @property {QueueArguments} queueArguments
* @property {ConnectionOptions} connectionOptions
*/
const amqp = require('amqplib/callback_api')
const events = require('events')
const _get = require('lodash/get')
const _set = require('lodash/set')
const _isSafeInteger = require('lodash/isSafeInteger')
const _isString = require('lodash/isString')
const _isEmpty = require('lodash/isEmpty')
const _isPlainObject = require('lodash/isPlainObject')
const _isBoolean = require('lodash/isBoolean')
const _isBuffer = require('lodash/isBuffer')
const _isNull = require('lodash/isNull')
const ConsoleLogger = require('./console_logger')
const MESSAGES = require('./internal').MESSAGES
const QUEUE_ARGUMENTS_MAP = {
messageTtl: 'x-message-ttl',
expires: 'x-expires',
maxLength: 'x-max-length',
maxPriority: 'x-max-priority'
}
class Common {
/**
* @param {AmqpURI} amqpUri
* @param {ChannelName} channelName
* @param {Configuration} [configuration]
*/
constructor (amqpUri, channelName, configuration) {
/**
* Name associated with this instance - displayed in the logs.
* Each instance should override this with some relevant name.
* @type {string}
* @protected
*/
this._name = 'COMMON'
/**
* Internal instance to log messages.
* @type {ConsoleLogger}
* @protected
*/
this._consoleLogger = new ConsoleLogger('IAMQP')
/**
* Used to emit messages and statues.
* @type {*|EventEmitter}
* @protected
*/
this._E = new events.EventEmitter()
/**
* In case the user initiates the closing of the connection
* the reconnect is not preformed.
* @type {boolean}
* @private
*/
this._manualConnectionClose = false
/**
* Internal flag for signaling of the connection if ready.
* @type {boolean}
* @protected
*/
this._isReady = false
/**
* Is the connection process going on.
* This is a flag to use as not to perform multiple connection attempts.
* @type {boolean}
* @private
*/
this._isConnecting = false
this._amqp = {
reconnectMultiplier: 1,
connection: null,
channel: null,
queue: null
}
// Lose the reference
configuration = JSON.parse(JSON.stringify(configuration || {}))
/**
* @type {{configuration: Configuration, amqpUri: AmqpURI, channelName: ChannelName}}
* @protected
*/
this._configuration = {
amqpUri,
channelName,
configuration
}
this._processConfiguration()
}
/**
* Check the configuration of this instance.
* @private
*/
_processConfiguration () {
const ERR = (message) => {
this._messageDispatcherWrapper('error', message)
throw new Error(message)
}
const CHECK = {
amqpUri: {
default: undefined,
required: true,
check: (e) => _isString(e) && e.startsWith('amqp://')
},
channelName: {
default: undefined,
required: true,
check: (e) => _isString(e) && !_isEmpty(e.trim())
},
configuration: {
default: {},
required: false,
check: (e) => _isPlainObject(e)
},
'configuration.printMessages': {
default: true,
required: false,
check: (e) => _isBoolean(e)
},
'configuration.maxMessageLengthToPrint': {
default: 257,
required: false,
check: (e) => _isSafeInteger(e) && e > 0
},
'configuration.queueArguments': {
default: {},
required: false,
check: (e) => _isPlainObject(e)
},
'configuration.connectionOptions': {
default: {},
required: false,
check: (e) => _isPlainObject(e)
},
'configuration.queueArguments.messageTtl': {
default: undefined,
required: false,
check: (e) => _isSafeInteger(e) && e >= 0 && e < 4294967296
},
'configuration.queueArguments.expires': {
default: undefined,
required: false,
check: (e) => _isSafeInteger(e) && e > 0 && e < 4294967296
},
'configuration.queueArguments.maxLength': {
default: undefined,
required: false,
check: (e) => _isSafeInteger(e)
},
'configuration.queueArguments.maxPriority': {
default: undefined,
required: false,
check: (e) => _isSafeInteger(e)
},
'configuration.connectionOptions.heartbeat': {
default: 60,
required: false,
check: (e) => _isSafeInteger(e) && e >= 0
}
}
for (const conf in CHECK) {
const ref = CHECK[conf]
let val = _get(this._configuration, conf, undefined)
if (ref.required && val === undefined) {
ERR(`${MESSAGES.MISSING_CONFIGURATION} "${conf}" `)
}
if (ref.default !== undefined && val === undefined) {
_set(this._configuration, conf, ref.default)
val = _get(this._configuration, conf, undefined)
if (!ref.check(val)) {
ERR(`${MESSAGES.INVALID_CONFIGURATION} "${conf}" `)
}
}
if (val !== undefined && !ref.check(val)) {
ERR(`${MESSAGES.INVALID_CONFIGURATION} "${conf}" `)
}
}
// Replace the queue arguments names with the valid names that go into amqplib
const queueArguments = _get(this._configuration, 'configuration.queueArguments', {})
for (const argRef in queueArguments) {
if (!Object.prototype.hasOwnProperty.call(QUEUE_ARGUMENTS_MAP, argRef)) {
continue
}
const val = queueArguments[argRef]
delete queueArguments[argRef]
queueArguments[QUEUE_ARGUMENTS_MAP[argRef]] = val
}
}
/**
* Wrapper for common message handling.
* @param {'log'|'info'|'warn'|'error'} logType one of `log, info, warn, error`
* @param {string} message message to print
* @protected
*/
_messageDispatcherWrapper (logType, message) {
this._consoleLogger[logType] && this._consoleLogger[logType](`[${this._name}] ${message.toString()}`)
}
/**
* @param {string} errorMessage
* @protected
*/
_emitAndDispatchError (errorMessage) {
this._messageDispatcherWrapper('error', errorMessage)
this._E.emit('error', new Error(errorMessage))
}
/**
* Timeout the connection attempt.
* @private
*/
_reconnectHandler () {
this._messageDispatcherWrapper('info', MESSAGES.RECONNECTING)
if (this._amqp.reconnectMultiplier < 6) {
this._amqp.reconnectMultiplier += 0.125
}
// THAT IS RIGHT YOU SONS-OF-BITCHES; THAT IS THIS; BACK TO THE FUTURE MOTHERFUCKER
const that = this
setTimeout(() => {
that._isConnecting = false
that._connect()
}, 10000 * this._amqp.reconnectMultiplier)
}
/**
* Connect the AMQP and setup event listeners.
* @private
*/
_connect () {
if (this._isConnecting) {
return
}
this._isConnecting = true
this._manualConnectionClose = false
amqp.connect(
`${this._configuration.amqpUri}?heartbeat=${this._configuration.configuration.connectionOptions.heartbeat}`,
(amqpConnErr, amqpConn) => {
if (amqpConnErr) {
this._isReady = false
this._messageDispatcherWrapper('error', `${MESSAGES.CONNECTION_ERROR} -> ${amqpConnErr.message}`)
this._E.emit('error', amqpConnErr)
this._reconnectHandler()
} else {
this._amqp.connection = amqpConn
// event ERROR
this._amqp.connection.on('error', (amqpErr) => {
this._isReady = false
this._messageDispatcherWrapper('error', `${MESSAGES.CONNECTION_ERROR} -> ${amqpErr.message}`)
// Closing connection can be a valid action
if (amqpErr.message !== 'Connection closing') {
this._E.emit('error', amqpErr)
}
})
// event CLOSE
this._amqp.connection.on('close', () => {
this._isReady = false
this._E.emit('closed')
if (this._manualConnectionClose === false) {
this._messageDispatcherWrapper('error', MESSAGES.CONNECTION_ERROR)
this._reconnectHandler()
} else {
this._messageDispatcherWrapper('info', MESSAGES.CONNECTION_CLOSE)
}
})
this._amqp.reconnectMultiplier = 1
this._messageDispatcherWrapper('info', MESSAGES.CONNECTION_OPEN)
this._isReady = true
this._isConnecting = false
this._E.emit('connected')
}
})
}
/**
* Parse the message that was consumed.
* The parsing tries parsing via the <code>JSON.parse</code>.
* If this fails, it just tries to return a string representation.
* It can parse:
* <ul>
* <li>number</li>
* <li>string</li>
* <li>plain Object</li>
* <li>Array</li>
* <li>boolean</li>
* </ul>
* @param {Object} message as received by the consumer; it expects a property <code>content</code> of type Buffer
* @returns {number | string | Object | boolean | Array | null} null is returned if the message was not parsed
* @protected
*/
_parseConsumedMessage (message) {
if (
!message ||
!Object.prototype.hasOwnProperty.call(message, 'content') ||
!_isBuffer(message.content)
) {
return null
}
try {
return JSON.parse(message.content.toString())
} catch (parseError) {
if (message.content.toString) {
return message.content.toString()
}
return null
}
}
/**
* Prepare the data to be published.
* @param {string | Object | number | boolean | Array} message
* @returns {Buffer | null}
* @protected
*/
_preparePublishingMessage (message) {
let dataToPublish = null
if (_isString(message)) {
dataToPublish = Buffer.from(message)
} else {
try {
dataToPublish = Buffer.from(JSON.stringify(message))
// If reverse fails, the message could not be send either way, so we stop it here
JSON.parse(dataToPublish.toString())
} catch (parseErr) {
this._messageDispatcherWrapper('error', `${MESSAGES.DATA_PARSE_FAIL} -> ${parseErr.message}`)
dataToPublish = null
}
}
return dataToPublish
}
/**
* @param {number | string | boolean | Object | Array} contentToPublish
* @returns {* | null}
*/
_commonPublishPreparations (contentToPublish) {
if (this._isReady !== true) {
this._messageDispatcherWrapper('error', `${MESSAGES.PUBLISHING} -> ${MESSAGES.CONNECTION_ERROR}`)
return null
}
if (!this._amqp.channel) {
this._messageDispatcherWrapper('error', `${MESSAGES.PUBLISHING} -> ${MESSAGES.CHANNEL_NOT_READY}`)
return null
}
const data = this._preparePublishingMessage(contentToPublish)
if (_isNull(data)) {
this._messageDispatcherWrapper('error', `${MESSAGES.PUBLISHING} -> ${MESSAGES.INVALID_DATA}`)
return null
}
this._logData(data, MESSAGES.PUBLISHING)
return data
}
/**
* Log the data that was received or send in the channels.
* @param {Object | Buffer} data
* @param {string} description some additional information to be logged with the data
* @protected
*/
_logData (data, description) {
if (!this._configuration.configuration.printMessages) {
return
}
let actualDataRef = null
if (_isBuffer(data)) {
actualDataRef = data
} else if (Object.prototype.hasOwnProperty.call(data, 'content') && _isBuffer(data.content)
) {
actualDataRef = data.content
}
if (actualDataRef === null) {
return
}
let thatLength = null
try {
thatLength = Buffer.byteLength(actualDataRef)
} catch (someError) {
thatLength = this._configuration.configuration.maxMessageLengthToPrint * 2
}
if (thatLength < this._configuration.configuration.maxMessageLengthToPrint) {
this._messageDispatcherWrapper('log', `${description} message: ${actualDataRef.toString()}`)
} else {
this._messageDispatcherWrapper('log', `${description} message`)
}
}
/**
* Close the connection.
* The client will not preform an automatic reconnect when this is called.
* @returns {boolean} false if there was an error
*/
closeConnection () {
try {
this._amqp.connection && this._amqp.connection.close()
this._manualConnectionClose = true
return true
} catch (cer) {
this._messageDispatcherWrapper('error', `${MESSAGES.CONNECTION_CLOSE} -> ${cer.message}`)
}
return false
}
/**
* Open the connection.
*/
openConnection () {
if (this._isReady !== true) {
this._connect()
}
}
/**
* Check if the connection is operational.
* @returns {boolean}
*/
isReady () {
return this._isReady
}
/**
* Get the EventEmitter that emits messages and errors.
* @returns {EventEmitter}
*/
getEventer () {
return this._E
}
/**
* Get a string representation of this instance.
* If prints the keys and values of all non-private members that are primitives.
* @returns {string}
* @public
*/
toString () {
return `{${[
this._name,
this._configuration.amqpUri,
this._configuration.channelName
].join(', ')}}`
}
}
module.exports = Common