Merge pull request #80 from bbc/unexpectedExceptions

Fix: Catch exceptions from handler and propagate err via callback.

index.js

@@ -151,7 +151,11 @@ Consumer.prototype._processMessage = function (message, cb) {
   this.emit('message_received', message);
   async.series([
     function handleMessage(done) {
-      consumer.handleMessage(message, done);
+      try {
+        consumer.handleMessage(message, done);
+      } catch (err) {
+        done(new Error('Unexpected message handler failure: ' + err.message));
+      }
     },
     function deleteMessage(done) {
       consumer._deleteMessage(message, done);
@@ -169,7 +173,7 @@ Consumer.prototype._processMessage = function (message, cb) {
           QueueUrl: consumer.queueUrl,
           ReceiptHandle: message.ReceiptHandle,
           VisibilityTimeout: 0
-        }, function(err) {
+        }, function (err) {
           if (err) consumer.emit('error', err, message);
           cb();
         });

test/index.js

@@ -104,6 +104,26 @@ describe('Consumer', function () {
       consumer.start();
     });
 
+    it('handles unexpected exceptions thrown by the handler function', function (done) {
+      consumer = new Consumer({
+        queueUrl: 'some-queue-url',
+        region: 'some-region',
+        handleMessage: function () {
+          throw new Error('unexpected parsing error');
+        },
+        sqs: sqs,
+        authenticationErrorTimeout: 20
+      });
+
+      consumer.on('processing_error', function (err) {
+        assert.ok(err);
+        assert.equal(err.message, 'Unexpected message handler failure: unexpected parsing error');
+        done();
+      });
+
+      consumer.start();
+    });
+
     it('fires an error event when an error occurs deleting a message', function (done) {
       var deleteErr = new Error('Delete error');