<commit_before>
/*!
 * Module dependencies.
 */

var snake = require('to-snake-case');
var type = require('component-type');
var debug = require('debug')('starsky:consumer');

/**
 * Export `Consumer`.
 */

module.exports = Consumer;

/**
 * Consumer constructor.
 *
 * TODO: document me...
 *
 * @param {Starsky} starsky
 * @param {String} queue
 * @constructor
 */

function Consumer (starsky, name) {
  this.starsky = starsky;
  this.name = snake(name);
  this.topics = [];
  this.inbox = {};
  this.tag = null;
  this.queue = null;
}

/**
 * Adds topics to subscribe to. Note that this only works prior
 * to the queue being created after calling `consume`.
 *
 * @param {String|Array} topic
 */

Consumer.prototype.subscribe = function (topic) {
  if (Array.isArray(topic)) {
    this.topics = this.topics.concat(topic);
  } else {
    this.topics.push(topic);
  }
  return this;
};

/**
 * Initializes the queue, topic bindings and subscription with RabbitMQ. Upon
 * each message receieved, we execute the `consume` function with the message
 * payload and pass the "next" callback.
 *
 * TODO: expose prefetch
 *
 * @param {Function} callback
 * @private
 */

Consumer.prototype.consume = function (callback) {
  debug('consume');

  if (!this.starsky.ready) {
    throw new Error('not ready');
  }

  var name = this.name;
  var namespace = this.starsky.get('namespace');
  var options = { durable: true, autoDelete: false };

  if (namespace) {
    name = namespace + ':' + name;
  }

  this.starsky.connection.queue(name, options, function (queue) {
    var exchange = this.starsky.get('mq exchange');
    var pending = this.topics.length;
    var self = this;

    this.queue = queue;
    this.topics.forEach(bind);

    function bind (topic) {
      self.queue.bind(exchange, topic, next);
    }

    function next () {
      --pending || subscribe();
    }

    function subscribe () {
      var prefetch = self.starsky.get('prefetch');
      var options = { ack: true , prefetchCount: prefetch };
      var consumer = queue.subscribe(options, self.onMessage(callback));
      consumer.addCallback(tag);
    }

    function tag (ok) {
      self.tag = ok.consumerTag;
    }
  }.bind(this));
};

/**
 * Handles incoming messages.
 *
 * TODO: timeouts?
 *
 * @param {Object} data
 * @param {Object} headers
 * @param {Object} info
 * @param {Message} msg
 * @private
 */

Consumer.prototype.onMessage = function (callback) {
  var self = this;
  return function (body, headers, info, msg) {
    // just the essentials needed for the callback
    var packet = {
      id: info.messageId,
      body: body,
      timestamp: info.timestamp,
      topic: msg.routingKey
    };

    // keep track of packets in-process, see quit()
    self.inbox[packet.id] = packet;
<commit_msg>Fix bug with multiple subscriptions<commit_after>
/*!
 * Module dependencies.
 */

var snake = require('to-snake-case');
var type = require('component-type');
var debug = require('debug')('starsky:consumer');

/**
 * Export `Consumer`.
 */

module.exports = Consumer;

/**
 * Consumer constructor.
 *
 * TODO: document me...
 *
 * @param {Starsky} starsky
 * @param {String} queue
 * @constructor
 */

function Consumer (starsky, name) {
  this.starsky = starsky;
  this.name = snake(name);
  this.topics = [];
  this.inbox = {};
  this.tag = null;
  this.queue = null;
}

/**
 * Adds topics to subscribe to. Note that this only works prior
 * to the queue being created after calling `consume`.
 *
 * @param {String|Array} topic
 */

Consumer.prototype.subscribe = function (topic) {
  if (Array.isArray(topic)) {
    this.topics = this.topics.concat(topic);
  } else {
    this.topics.push(topic);
  }
  return this;
};

/**
 * Initializes the queue, topic bindings and subscription with RabbitMQ. Upon
 * each message receieved, we execute the `consume` function with the message
 * payload and pass the "next" callback.
 *
 * TODO: expose prefetch
 *
 * @param {Function} callback
 * @private
 */

Consumer.prototype.consume = function (callback) {
  debug('consume');

  if (!this.starsky.ready) {
    throw new Error('not ready');
  }

  var name = this.name;
  var namespace = this.starsky.get('namespace');
  var options = { durable: true, autoDelete: false };

  if (namespace) {
    name = namespace + ':' + name;
  }

  this.starsky.connection.queue(name, options, function (queue) {
    debug('queue declared');

    var exchange = this.starsky.get('mq exchange');
    var self = this;

    this.queue = queue;
    this.topics.forEach(bind);

    function bind (topic) {
      debug('bind %s', topic);
      self.queue.bind(exchange, topic, next);
    }

    function next () {
      debug('next');
      subscribe();
    }

    function subscribe () {
      debug('subscribe');
      var prefetch = self.starsky.get('prefetch');
      var options = { ack: true , prefetchCount: prefetch };
      var consumer = queue.subscribe(options, self.onMessage(callback));
      consumer.addCallback(tag);
    }

    function tag (ok) {
      debug('tag');
      self.tag = ok.consumerTag;
    }
  }.bind(this));
};

/**
 * Handles incoming messages.
 *
 * TODO: timeouts?
 *
 * @param {Object} data
 * @param {Object} headers
 * @param {Object} info
 * @param {Message} msg
 * @private
 */

Consumer.prototype.onMessage = function (callback) {
  var self = this;
  return function (body, headers, info, msg) {
    // just the essentials needed for the callback
    var packet = {
      id: info.messageId,
      body: body,
      timestamp: info.timestamp,
      topic: msg.routingKey
    };

    // keep track of packets in-process, see quit()
    self.inbox[packet.id] = packet;

{'fraction_non_alphanumeric': 0.14406630223946393, 'fraction_numerical': 0.0, 'mean_word_length': 3.022695035460993, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4576539', 'n_tokens_mistral': 1813, 'n_tokens_neox': 1751, 'n_words': 490}