<commit_before>'use strict'

var util = require('util')
var EventEmitter = require('events').EventEmitter
var AWS = require('aws-sdk')
var debug = require('debug')('simple-sqs')

var defaultOpts = {
  apiVersion: '2012-11-05',
  region: 'us-east-1'
}

module.exports = function (opts) {
  return function (queue, cb) {
    queue = new Queue(queue, opts)
    if (cb) queue.on('message', cb)
    return queue
  }
}

var Queue = function (queue, opts) {
  if (!(this instanceof Queue)) return new Queue(queue, opts)

  EventEmitter.call(this)

  this.queue = queue
  this.opts = opts || defaultOpts
  this.once('newListener', this._start.bind(this))
}

util.inherits(Queue, EventEmitter)

Queue.prototype._start = function (event, fn) {
  if (event !== 'message') return this.once('newListener', this._start.bind(this))
  this.sqs = new AWS.SQS(this.opts)
  process.nextTick(this.poll.bind(this))
}

Queue.prototype.deleteMsg = function (msg) {
  var self = this
  var opts = {
    QueueUrl: this.queue,
    ReceiptHandle: msg.ReceiptHandle
  }
  debug('[%s] Deleting message...', msg.MessageId)
  this.sqs.deleteMessage(opts, function (err) {
    if (err) self.emit('error', err)
  })
}

Queue.prototype.poll = function () {
  var self = this
  var opts = {
    QueueUrl: this.queue,
    AttributeNames: [ 'All' ]
  }
  debug('Polling queue...')
  this.sqs.receiveMessage(opts, function (err, data) {
    if (err) {
      self.emit('error', err)
      debug('Waiting 5 seconds before retrying...')
      setTimeout(self.poll.bind(self), 1000 * 5)
      return
    }
    (data.Messages || []).forEach(self._processMsg.bind(self))
    self.poll()
  })
}

Queue.prototype._processMsg = function (msg) {
  var self = this
  debug('[%s] Processing message...', msg.MessageId)
  try {
    msg.Body = JSON.parse(msg.Body)
  } catch (e) {
    debug('[%s] Could not parse message', msg.MessageId)
    if (!self.opts.ignoreParseErrors) {
      e.MessageId = msg.MessageId
      e.Body = msg.Body
      this.emit('error', e)
      this.deleteMsg(msg) // since the message could not be parsed, there is no need in trying again
      return
    }
  }
  this.emit('message', msg, function (err) {
    if (err) return self.emit('error', err)
    self.deleteMsg(msg)
  })
}
<commit_msg>Add support for SQS Message Attributes. Now for real<commit_after>'use strict'

var util = require('util')
var EventEmitter = require('events').EventEmitter
var AWS = require('aws-sdk')
var debug = require('debug')('simple-sqs')

var defaultOpts = {
  apiVersion: '2012-11-05',
  region: 'us-east-1'
}

module.exports = function (opts) {
  return function (queue, cb) {
    queue = new Queue(queue, opts)
    if (cb) queue.on('message', cb)
    return queue
  }
}

var Queue = function (queue, opts) {
  if (!(this instanceof Queue)) return new Queue(queue, opts)

  EventEmitter.call(this)

  this.queue = queue
  this.opts = opts || defaultOpts
  this.once('newListener', this._start.bind(this))
}

util.inherits(Queue, EventEmitter)

Queue.prototype._start = function (event, fn) {
  if (event !== 'message') return this.once('newListener', this._start.bind(this))
  this.sqs = new AWS.SQS(this.opts)
  process.nextTick(this.poll.bind(this))
}

Queue.prototype.deleteMsg = function (msg) {
  var self = this
  var opts = {
    QueueUrl: this.queue,
    ReceiptHandle: msg.ReceiptHandle
  }
  debug('[%s] Deleting message...', msg.MessageId)
  this.sqs.deleteMessage(opts, function (err) {
    if (err) self.emit('error', err)
  })
}

Queue.prototype.poll = function () {
  var self = this
  var opts = {
    QueueUrl: this.queue,
    AttributeNames: [ 'All' ],
    MessageAttributeNames: [ 'All' ]
  }
  debug('Polling queue...')
  this.sqs.receiveMessage(opts, function (err, data) {
    if (err) {
      self.emit('error', err)
      debug('Waiting 5 seconds before retrying...')
      setTimeout(self.poll.bind(self), 1000 * 5)
      return
    }
    (data.Messages || []).forEach(self._processMsg.bind(self))
    self.poll()
  })
}

Queue.prototype._processMsg = function (msg) {
  var self = this
  debug('[%s] Processing message...', msg.MessageId)
  try {
    msg.Body = JSON.parse(msg.Body)
  } catch (e) {
    debug('[%s] Could not parse message', msg.MessageId)
    if (!self.opts.ignoreParseErrors) {
      e.MessageId = msg.MessageId
      e.Body = msg.Body
      this.emit('error', e)
      this.deleteMsg(msg) // since the message could not be parsed, there is no need in trying again
      return
    }
  }
  this.emit('message', msg, function (err) {
    if (err) return self.emit('error', err)
    self.deleteMsg(msg)
  })
}

{'fraction_non_alphanumeric': 0.15543030565792326, 'fraction_numerical': 0.006503360069369174, 'mean_word_length': 3.2881040892193307, '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': '386673', 'n_tokens_mistral': 1602, 'n_tokens_neox': 1572, 'n_words': 380}