lib/amqp/queue.js
function subscribe(callback, subscribeOptions) {
    let consumeOptions = {};
    _.assign(consumeOptions, options, subscribeOptions);
    consumeOptions = _.pick(consumeOptions, ['consumerTag', 'noAck', 'noBatch', 'limit', 'exclusive', 'priority', 'arguments']);

    let shouldAck = !consumeOptions.noAck;
    let shouldBatch = !consumeOptions.noBatch;

    if (shouldAck && shouldBatch) {
      messages.listenForSignal();
    }

    qLog.info(`Starting subscription ${channelName} - ${connectionName} with ${JSON.stringify(consumeOptions)}`);
    return (consumeOptions.limit !== false ? channel.prefetch(consumeOptions.limit || 500) : Promise.resolve())
      .then(() => {
        function handler(raw) {
          let ops = getResolutionOperations(raw, consumeOptions);
          qLog.silly(`Received message on queue ${channelName} - ${connectionName}`);

          if (shouldAck && shouldBatch) {
            messages.addMessage(ops.message);
          }

          try {
            callback(raw, ops);
          } catch(e) {
            qLog.error(`Error handing message on queue ${channelName} - ${connectionName}`, e);
            ops.reject();
          }
        }
        return channel.consume(channelName, handler, _.omit(consumeOptions, ['noBatch', 'limit']));
      }).then(function(result) {
        channel.tag = result.consumerTag;
        return result;
      });
  }
function subscribe(callback, subscribeOptions) {
    let consumeOptions = {};
    _.assign(consumeOptions, options, subscribeOptions);
    consumeOptions = _.pick(consumeOptions, ['consumerTag', 'noAck', 'noBatch', 'limit', 'exclusive', 'priority', 'arguments']);

    let shouldAck = !consumeOptions.noAck;
    let shouldBatch = !consumeOptions.noBatch;

    if (shouldAck && shouldBatch) {
      messages.listenForSignal();
    }

    qLog.info(`Starting subscription ${channelName} - ${connectionName} with ${JSON.stringify(consumeOptions)}`);
    return (consumeOptions.limit !== false ? channel.prefetch(consumeOptions.limit || 500) : Promise.resolve())
      .then(() => {
        function handler(raw) {
          let ops = getResolutionOperations(raw, consumeOptions);
          qLog.silly(`Received message on queue ${channelName} - ${connectionName}`);

          if (shouldAck && shouldBatch) {
            messages.addMessage(ops.message);
          }

          try {
            callback(raw, ops);
          } catch(e) {
            qLog.error(`Error handing message on queue ${channelName} - ${connectionName}`, e);
            ops.reject();
          }
        }
        return channel.consume(channelName, handler, _.omit(consumeOptions, ['noBatch', 'limit']));
      }).then(function(result) {
        channel.tag = result.consumerTag;
        return result;
      });
  }
function subscribe ( callback , subscribeOptions ) { let consumeOptions = { } ; _ . assign ( consumeOptions , options , subscribeOptions ) ; consumeOptions = _ . pick ( consumeOptions , [ 'consumerTag' , 'noAck' , 'noBatch' , 'limit' , 'exclusive' , 'priority' , 'arguments' ] ) ; let shouldAck = ! consumeOptions . noAck ; let shouldBatch = ! consumeOptions . noBatch ; if ( shouldAck && shouldBatch ) { messages . listenForSignal ( ) ; } qLog . info ( ` ${ channelName } ${ connectionName } ${ JSON . stringify ( consumeOptions ) } ` ) ; return ( consumeOptions . limit !== false ? channel . prefetch ( consumeOptions . limit || 500 ) : Promise . resolve ( ) ) . then ( ( ) => { function handler ( raw ) { let ops = getResolutionOperations ( raw , consumeOptions ) ; qLog . silly ( ` ${ channelName } ${ connectionName } ` ) ; if ( shouldAck && shouldBatch ) { messages . addMessage ( ops . message ) ; } try { callback ( raw , ops ) ; } catch ( e ) { qLog . error ( ` ${ channelName } ${ connectionName } ` , e ) ; ops . reject ( ) ; } } return channel . consume ( channelName , handler , _ . omit ( consumeOptions , [ 'noBatch' , 'limit' ] ) ) ; } ) . then ( function ( result ) { channel . tag = result . consumerTag ; return result ; } ) ; }
Subscribe to the queue's messages
@public
@param {Function} callback - the function to call for each message received from queue. Should have signature function (message, operations) where operations contains ack, nack, and reject functions
@param {Object} subscribeOptions - details in consuming from the queue. Overrides any equivalent options sent in to Queue constructor - passed to amqplib's consume function
@param {Number} subscribeOptions.limit - the channel prefetch limit
@param {bool} subscribeOptions.noAck - if true, the broker won't expect an acknowledgement of messages delivered to this consumer; i.e., it will dequeue messages as soon as they've been sent down the wire. Defaults to false (i.e., you will be expected to acknowledge messages).
@param {bool} subscribeOptions.noBatch - if true, ack/nack/reject operations will execute immediately and not be batched
@param {String} subscribeOptions.consumerTag - a name which the server will use to distinguish message deliveries for the consumer; mustn't be already in use on the channel. It's usually easier to omit this, in which case the server will create a random name and supply it in the reply.
@param {bool} subscribeOptions.exclusive - if true, the broker won't let anyone else consume from this queue; if there already is a consumer, there goes your channel (so usually only useful if you've made a 'private' queue by letting the server choose its name).
@param {Number} subscribeOptions.priority - gives a priority to the consumer; higher priority consumers get messages in preference to lower priority consumers. See this RabbitMQ extension's documentation
@param {Object} subscribeOptions.arguments -  arbitrary arguments. Go to town.
Subscribe to the queue s messages