Merge pull request #4 from DaPulse/addition/ran/missing_function_callbacks

addition of data in instrumentation and version upgrade

Author: RanNir143

dist/consumer.js

@@ -71,6 +71,9 @@ class Consumer extends events_1.EventEmitter {
         this.handleMessageBatch = options.handleMessageBatch;
         this.pollingStartedInstrumentCallback = options.pollingStartedInstrumentCallback;
         this.pollingFinishedInstrumentCallback = options.pollingFinishedInstrumentCallback;
+        this.batchStartedInstrumentCallBack = options.batchStartedInstrumentCallBack;
+        this.batchFinishedInstrumentCallBack = options.batchFinishedInstrumentCallBack;
+        this.batchFailedInstrumentCallBack = options.batchFailedInstrumentCallBack;
         this.handleMessageTimeout = options.handleMessageTimeout;
         this.attributeNames = options.attributeNames || [];
         this.messageAttributeNames = options.messageAttributeNames || [];
@@ -135,7 +138,8 @@ class Consumer extends events_1.EventEmitter {
             this.pollingFinishedInstrumentCallback({
                 instanceId: process.env.HOSTNAME,
                 queueUrl: this.queueUrl,
-                messagesReceived: numberOfMessages
+                messagesReceived: numberOfMessages,
+                freeConcurrentSlots: this.freeConcurrentSlots
             });
         }
         if (response) {
@@ -251,7 +255,8 @@ class Consumer extends events_1.EventEmitter {
             this.pollingStartedInstrumentCallback({
                 instanceId: process.env.HOSTNAME,
                 queueUrl: this.queueUrl,
-                pollBatchSize
+                pollBatchSize,
+                freeConcurrentSlots: this.freeConcurrentSlots
             });
         }
         let currentPollingTimeout = this.pollingWaitTimeMs;
@@ -296,31 +301,34 @@ class Consumer extends events_1.EventEmitter {
                 instanceId: process.env.HOSTNAME,
                 queueUrl: this.queueUrl,
                 batchUuid,
-                numberOfMessages: messages.length
+                numberOfMessages: messages.length,
+                freeConcurrentSlots: this.freeConcurrentSlots
             });
         }
-        try {
-            await this.handleMessageBatch(messages, this);
+        this.handleMessageBatch(messages, this)
+            .then(() => {
             if (this.batchFinishedInstrumentCallBack) {
                 this.batchFinishedInstrumentCallBack({
                     instanceId: process.env.HOSTNAME,
                     queueUrl: this.queueUrl,
                     batchUuid,
-                    numberOfMessages: messages.length
+                    numberOfMessages: messages.length,
+                    freeConcurrentSlots: this.freeConcurrentSlots
                 });
             }
-        }
-        catch (err) {
+        })
+            .catch(err => {
             if (this.batchFailedInstrumentCallBack) {
                 this.batchFailedInstrumentCallBack({
                     instanceId: process.env.HOSTNAME,
                     queueUrl: this.queueUrl,
                     batchUuid,
-                    numberOfMessages: messages.length
+                    numberOfMessages: messages.length,
+                    freeConcurrentSlots: this.freeConcurrentSlots,
+                    error: err
                 });
             }
-            throw err;
-        }
+        });
     }
 }
 exports.Consumer = Consumer;

package.json

@@ -1,6 +1,6 @@
 {
   "name": "sqs-consumer",
-  "version": "5.7.8",
+  "version": "5.7.81",
   "description": "Build SQS-based Node applications without the boilerplate",
   "main": "dist/index.js",
   "types": "dist/index.d.ts",

src/consumer.ts

@@ -140,6 +140,9 @@ export class Consumer extends EventEmitter {
     this.handleMessageBatch = options.handleMessageBatch;
     this.pollingStartedInstrumentCallback = options.pollingStartedInstrumentCallback;
     this.pollingFinishedInstrumentCallback = options.pollingFinishedInstrumentCallback;
+    this.batchStartedInstrumentCallBack = options.batchStartedInstrumentCallBack;
+    this.batchFinishedInstrumentCallBack = options.batchFinishedInstrumentCallBack;
+    this.batchFailedInstrumentCallBack = options.batchFailedInstrumentCallBack;
     this.handleMessageTimeout = options.handleMessageTimeout;
     this.attributeNames = options.attributeNames || [];
     this.messageAttributeNames = options.messageAttributeNames || [];
@@ -216,7 +219,8 @@ export class Consumer extends EventEmitter {
       this.pollingFinishedInstrumentCallback({
         instanceId: process.env.HOSTNAME,
         queueUrl: this.queueUrl,
-        messagesReceived: numberOfMessages
+        messagesReceived: numberOfMessages,
+        freeConcurrentSlots: this.freeConcurrentSlots
       });
     }
 
@@ -334,7 +338,8 @@ export class Consumer extends EventEmitter {
       this.pollingStartedInstrumentCallback({
         instanceId: process.env.HOSTNAME,
         queueUrl: this.queueUrl,
-        pollBatchSize
+        pollBatchSize,
+        freeConcurrentSlots: this.freeConcurrentSlots
       });
     }
 
@@ -384,32 +389,34 @@ export class Consumer extends EventEmitter {
         instanceId: process.env.HOSTNAME,
         queueUrl: this.queueUrl,
         batchUuid,
-        numberOfMessages: messages.length
+        numberOfMessages: messages.length,
+        freeConcurrentSlots: this.freeConcurrentSlots
       });
     }
 
-    try {
-      await this.handleMessageBatch(messages, this);
-
-      if (this.batchFinishedInstrumentCallBack) {
-        this.batchFinishedInstrumentCallBack({
-          instanceId: process.env.HOSTNAME,
-          queueUrl: this.queueUrl,
-          batchUuid,
-          numberOfMessages: messages.length
-        });
-      }
-    } catch (err) {
-      if (this.batchFailedInstrumentCallBack) {
-        this.batchFailedInstrumentCallBack({
-          instanceId: process.env.HOSTNAME,
-          queueUrl: this.queueUrl,
-          batchUuid,
-          numberOfMessages: messages.length
-        });
-      }
-
-      throw err;
-    }
+    this.handleMessageBatch(messages, this)
+      .then(() => {
+        if (this.batchFinishedInstrumentCallBack) {
+          this.batchFinishedInstrumentCallBack({
+            instanceId: process.env.HOSTNAME,
+            queueUrl: this.queueUrl,
+            batchUuid,
+            numberOfMessages: messages.length,
+            freeConcurrentSlots: this.freeConcurrentSlots
+          });
+        }
+      })
+      .catch(err => {
+        if (this.batchFailedInstrumentCallBack) {
+          this.batchFailedInstrumentCallBack({
+            instanceId: process.env.HOSTNAME,
+            queueUrl: this.queueUrl,
+            batchUuid,
+            numberOfMessages: messages.length,
+            freeConcurrentSlots: this.freeConcurrentSlots,
+            error: err
+          });
+        }
+      });
   }
 }