<commit_before>'use strict'

if (!process.env.CLOUDAMQP_URL) {
  throw new Error('CLOUDAMQP_URL environment variable must be set')
}

var _ = require('lodash')

module.exports = function (options) {
  options = _.defaults(options, {
    urls: [process.env.CLOUDAMQP_URL],
    verbose: Boolean(process.env.DEBUG)
  })

  var amqp = require('amqp-connection-manager')
  var connection = amqp.connect(options.urls)

  connection.on('connect', function () {
    console.log('AMQP: connected')
  })
  connection.on('disconnect', function (params) {
    console.log('AMQP: disconnected', params.err)
  })

  var prepareMessage = function (onMessage) {
    return function (msg) {
      return onMessage(msg, JSON.parse(msg.content.toString()))
    }
  }

  var createChannel = function (name, setup, verbose) {
    var channelWrapper = connection.createChannel({
      name: name,
      json: true,
      setup: setup
    })

    channelWrapper.on('error', function (err) {
      console.log('AMQP channel:', name, 'error', err)
    })

    if (verbose) {
      channelWrapper.on('connect', function () {
        console.log('AMQP channel:', name, 'connected')
      })
      channelWrapper.on('close', function () {
        console.log('AMQP channel:', name, 'closed')
      })
    }

    return channelWrapper
  }

  var client = function (onMessage) {
    var channelWrapper = createChannel('client', function (channel) {
      return channel.assertQueue('', {exclusive: true, autoDelete: true}).then(function (q) {
        channelWrapper.replyTo = q.queue
        return channel.consume(q.queue, prepareMessage(onMessage), {noAck: true})
      })
    }, options.verbose)
    return channelWrapper
  }

  var server = function (queueName, onMessage) {
    return createChannel('server', function (channel) {
      return channel.assertQueue(queueName, {durable: false, expires: 60 * 1000}).then(function () {
        return channel.prefetch(1)
      }).then(function () {
        return channel.consume(queueName, prepareMessage(onMessage), {noAck: true})
      })
    }, options.verbose)
  }

  return {
    client: client,
    server: server
  }
}<commit_msg>Have development and production queues<commit_after>'use strict'

if (!process.env.CLOUDAMQP_URL) {
  throw new Error('CLOUDAMQP_URL environment variable must be set')
}

var _ = require('lodash')

module.exports = function (options) {
  options = _.defaults(options, {
    urls: [process.env.CLOUDAMQP_URL],
    verbose: Boolean(process.env.DEBUG),
    queuePrefix: process.env.NODE_ENV || 'development'
  })

  var amqp = require('amqp-connection-manager')
  var connection = amqp.connect(options.urls)

  connection.on('connect', function () {
    console.log('AMQP: connected')
  })
  connection.on('disconnect', function (params) {
    console.log('AMQP: disconnected', params.err)
  })

  var prepareMessage = function (onMessage) {
    return function (msg) {
      return onMessage(msg, JSON.parse(msg.content.toString()))
    }
  }

  var createChannel = function (name, setup, verbose) {
    var channelWrapper = connection.createChannel({
      name: name,
      json: true,
      setup: setup
    })

    channelWrapper.on('error', function (err) {
      console.log('AMQP channel:', name, 'error', err)
    })

    if (verbose) {
      channelWrapper.on('connect', function () {
        console.log('AMQP channel:', name, 'connected')
      })
      channelWrapper.on('close', function () {
        console.log('AMQP channel:', name, 'closed')
      })
    }

    return channelWrapper
  }

  var client = function (onMessage) {
    var channelWrapper = createChannel('client', function (channel) {
      return channel.assertQueue('', {exclusive: true, autoDelete: true}).then(function (q) {
        channelWrapper.replyTo = q.queue
        return channel.consume(q.queue, prepareMessage(onMessage), {noAck: true})
      })
    }, options.verbose)

    var orgSendToQueue = channelWrapper.sendToQueue.bind(channelWrapper)
    channelWrapper.sendToQueue = function (queueName, msg, opts, next) {
      orgSendToQueue(options.queuePrefix + '.' + queueName, msg, opts, next)
    }

    return channelWrapper
  }

  var server = function (queueName, onMessage) {
    queueName = options.queuePrefix + '.' + queueName
    return createChannel('server', function (channel) {
      return channel.assertQueue(queueName, {durable: false, expires: 60 * 1000}).then(function () {
        return channel.prefetch(1)
      }).then(function () {
        return channel.consume(queueName, prepareMessage(onMessage), {noAck: true})
      })
    }, options.verbose)
  }

  return {
    client: client,
    server: server
  }
}
{'fraction_non_alphanumeric': 0.1371208885091841, 'fraction_numerical': 0.00299017513882956, 'mean_word_length': 3.3201107011070112, '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': '1937282', 'n_tokens_mistral': 1444, 'n_tokens_neox': 1387, 'n_words': 312}