nl_tokens: publish : hides the complexity of publishing to the different queues
pl_tokens: function ( destactorid , type , message , socket ) { var self = this ; var qmsg = { } ; qmsg . type = type ; qmsg . message = <mask> ; if ( typeof destactorid !== 'string' ) { logger . error ( "Trying to publish on a bad queue: " + JSON . stringify ( destactorid ) ) ; if ( socket ) // Prevent new config deployment issues { socket . emit ( 'error' , "Trying to publish on a bad queue: " + JSON . stringify ( destactorid ) ) ; } return ; } logger . debug ( util . format ( "Publishing '%s' on Queue [%s]" , type , destactorid ) ) ; // First, we declare the Queue using passive arg, in order to perform checks (Queue exists + has consumers) this . amqp . queue ( destactorid , { passive : true } , function ( queue ) { } ) . once ( 'error' , function ( err ) { var errMsg = util . format ( "The queue [%s] does not exist (%s). The following message (type='%s') *is LOST*: %s" , destactorid , err , type , JSON . stringify ( message ) ) ; logger . error ( errMsg ) ; self . _storeNagiosEvent ( new NagiosCheckResponse ( NagiosCheckResponseCodes . ERROR , EVENTS_CONTROLLER_SUBSYSTEM , errMsg ) ) ; if ( socket ) // Prevent new config deployment issues { socket . emit ( 'error' , util . format ( 'Sorry, your message could not be delivered. Error=The queue [%s] does not exist.' , destactorid ) ) ; } } ) . once ( 'open' , function ( queueName , messageCount , consumerCount ) { // If the queue exists, we publish (publishing on a non existing queue will lose the message anyway) if ( consumerCount < 1 ) { var warnMsg = util . format ( 'Queue [%s] does not have any consumers. (Pending message(s) before publishing: %d)' , queueName , messageCount ) ; logger . warn ( warnMsg ) ; self . _storeNagiosEvent ( new NagiosCheckResponse ( NagiosCheckResponseCodes . WARNING , EVENTS_CONTROLLER_SUBSYSTEM , warnMsg ) ) ; } else { logger . trace ( util . format ( 'Before publishing, Queue [%s] had %d consumer(s) and %d message(s) pending.' , queueName , consumerCount , messageCount ) ) ; } self . amqp . publish ( destactorid , qmsg , { contentType : 'application/json' } ) ; } ) ; if ( this . amqp . channelCounter >= self . MAX_CHANNEL_PER_CONNECTION && ! self . creatingAmqpConnection ) { self . _createAmqpConnection ( ) ; } }