<commit_before>const _    = require('lodash');
const amqp = require('amqplib');
const util = require('./util');

const eventBusChannel = 'event-bus';

function RabbitBus(connectionOptions) {
  this.subscriberChannels = {};
  this.publisherChannels  = {};
  this.channels           = [];

  this.publish = (channelName, payload) => {
    return new Promise((resolve, reject) => {
      this.getPublisherChannel(channelName)
        .then(channel => {
          channel.publish(eventBusChannel, channelName, new Buffer(_.isPlainObject(payload) ? JSON.stringify(payload, util.replaceErrors) :  payload));
          resolve();
        })
        .catch(err => {
          reject(err);
        });
    });
  };

  this.subscribe = (channelName, listener) => {
    return new Promise((resolve, reject) => {
      this.getSubscriberChannel(channelName)
        .spread((channel, queue) => {
          resolve();
          channel.consume(queue, message => {
            listener(_.attempt(JSON.parse, message.content.toString()));
          });
        })
        .catch(err => {
          reject(err);
        });
    });
  };

  this.end = () => {
    return Promise.all(
      _.flatten([
        _.map(this.subscriberChannels, (channel) => {
          return channel.spread((c, q) => c.unbindExchange(eventBusChannel, 'direct', ''));
        }),
        _.map(this.publisherChannels, (channel) => {
          return channel.then(c => c.unbindExchange(eventBusChannel, 'direct', ''));
        })
      ]))<commit_msg>Turned noAck on. We don't need that, and it eventually pegs CPU if you don't ack the messages<commit_after>const _    = require('lodash');
const amqp = require('amqplib');
const util = require('./util');

const eventBusChannel = 'event-bus';

function RabbitBus(connectionOptions) {
  this.subscriberChannels = {};
  this.publisherChannels  = {};
  this.channels           = [];

  this.publish = (channelName, payload) => {
    return new Promise((resolve, reject) => {
      this.getPublisherChannel(channelName)
        .then(channel => {
          channel.publish(eventBusChannel, channelName, new Buffer(_.isPlainObject(payload) ? JSON.stringify(payload, util.replaceErrors) :  payload));
          resolve();
        })
        .catch(err => {
          reject(err);
        });
    });
  };

  this.subscribe = (channelName, listener) => {
    return new Promise((resolve, reject) => {
      this.getSubscriberChannel(channelName)
        .spread((channel, queue) => {
          resolve();
          channel.consume(queue, message => {
            listener(_.attempt(JSON.parse, message.content.toString()));
          }, { noAck: true });
        })
        .catch(err => {
          reject(err);
        });
    });
  };

  this.end = () => {
    return Promise.all(
      _.flatten([
        _.map(this.subscriberChannels, (channel) => {
          return channel.spread((c, q) => c.unbindExchange(eventBusChannel, 'direct', ''));
        }),
        _.map(this.publisherChannels, (channel) => {
          return channel.then(c => c.unbindExchange(eventBusChannel, 'direct', ''));
        })
      ]))
{'fraction_non_alphanumeric': 0.1607428754402818, 'fraction_numerical': 0.0, 'mean_word_length': 2.474972191323693, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 31, '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': '4363558', 'n_tokens_mistral': 922, 'n_tokens_neox': 893, 'n_words': 157}