failed parse messages is not handled #48 (#49)

package.json

@@ -1,6 +1,6 @@
 {
   "name": "squiss-ts",
-  "version": "4.0.3",
+  "version": "4.0.4",
   "description": "High-volume SQS poller",
   "main": "dist/index.js",
   "types": "dist/index.d.ts",

src/Message.ts

@@ -27,6 +27,7 @@ interface SNSBody {
 }
 
 interface IMessageEvents {
+    error: Error;
     delQueued: void;
     handled: void;
     released: void;
@@ -150,8 +151,9 @@ export class Message extends (EventEmitter as new() => MessageEmitter) {
         if (!this._handled) {
             this._handled = true;
             return this._squiss.releaseMessage(this)
-                .catch(() => {
+                .catch((e) => {
                     this._handled = false;
+                    this.emit('error', e);
                 });
         }
         return Promise.resolve();

src/Squiss.ts

@@ -207,11 +207,12 @@ export class Squiss extends (EventEmitter as new() => SquissEmitter) {
 
     public releaseMessage(msg: Message): Promise<void> {
         this.handledMessage(msg);
-        return this.changeMessageVisibility(msg, 0).then((res) => {
-            msg.emit('released');
-            this.emit('released', msg);
-            return res;
-        });
+        return this.changeMessageVisibility(msg, 0)
+            .then((res) => {
+                msg.emit('released');
+                this.emit('released', msg);
+                return res;
+            });
     }
 
     public purgeQueue(): Promise<void> {
@@ -366,6 +367,10 @@ export class Squiss extends (EventEmitter as new() => SquissEmitter) {
             message.parse()
                 .then(() => {
                     this.emit('message', message);
+                })
+                .catch((e: Error) => {
+                    this.emit('error', e);
+                    message.release();
                 });
         });
     }

src/test/src/Message.spec.ts

@@ -5,6 +5,8 @@ import {SquissStub} from '../stubs/SquissStub';
 import {SQS, S3} from 'aws-sdk';
 import {Blobs, S3Stub} from '../stubs/S3Stub';
 import delay from 'delay';
+// @ts-ignore
+import * as sinon from 'sinon';
 
 const wait = (ms?: number) => delay(ms === undefined ? 20 : ms);
 
@@ -393,15 +395,21 @@ describe('Message', () => {
       bodyFormat: 'json',
       squiss: {
         releaseMessage: (toDel: Message) => {
-          return Promise.reject();
+          return Promise.reject(new Error('failed to release'));
         },
       } as any as Squiss,
       s3Retriever: getS3Stub(),
       s3Retain: false,
     });
+    const errorSpy = sinon.spy();
+    msg.on('error', errorSpy);
     return msg.release()
       .then(() => {
         msg.isHandled().should.eql(false);
+        return wait().then(() => {
+          errorSpy.should.be.calledOnce();
+          errorSpy.should.be.calledWith(sinon.match.instanceOf(Error));
+        });
       });
   });
   it('calls Squiss.handledMessage on keep', (done) => {

src/test/src/index.spec.ts

@@ -330,6 +330,22 @@ describe('index', () => {
         ]);
       });
     });
+    it('emits error on message parse error', () => {
+      const msgSpy = sinon.spy();
+      const errorSpy = sinon.spy();
+      inst = new SquissPatched({bodyFormat: 'json', queueUrl: 'foo', maxInFlight: 15,
+        receiveBatchSize: 10} as ISquissOptions);
+      inst!.sqs = new SQSStub(0, 0) as any as SQS;
+      inst!.on('message', msgSpy);
+      inst!.on('error', errorSpy);
+      inst!.start();
+      inst!.sendMessage('{sdfsdf');
+      return wait().then(() => {
+        msgSpy.should.not.be.called();
+        errorSpy.should.be.calledOnce();
+        errorSpy.should.be.calledWith(sinon.match.instanceOf(Error));
+      });
+    });
     it('emits queueEmpty event with no messages', () => {
       const msgSpy = sinon.spy();
       const qeSpy = sinon.spy();