fix: timeout ids are not being set correctly for parallel running (#380)

* fix:`heartbeatTimeoutId` should be set within the process function * fix: the same needs to be done for handleMessageTimeoutId
bbc · Mar 18, 2023 · 7a47154 · 7a47154
1 parent c791903
commit 7a47154
Showing 1 changed file with 13 additions and 11 deletions.
diff --git a/src/consumer.ts b/src/consumer.ts
@@ -35,8 +35,6 @@ const debug = Debug('sqs-consumer');
  */
 export class Consumer extends TypedEventEmitter {
   private pollingTimeoutId: NodeJS.Timeout | undefined = undefined;
-  private heartbeatTimeoutId: NodeJS.Timeout | undefined = undefined;
-  private handleMessageTimeoutId: NodeJS.Timeout | undefined = undefined;
   private stopped = true;
   private queueUrl: string;
   private handleMessage: (message: Message) => Promise<Message | void>;
@@ -242,11 +240,13 @@ export class Consumer extends TypedEventEmitter {
    * @param message The message that was delivered from SQS
    */
   private async processMessage(message: Message): Promise<void> {
+    let heartbeatTimeoutId: NodeJS.Timeout | undefined = undefined;
+
     try {
       this.emit('message_received', message);
 
       if (this.heartbeatInterval) {
-        this.heartbeatTimeoutId = this.startHeartbeat(message);
+        heartbeatTimeoutId = this.startHeartbeat(message);
       }
 
       const ackedMessage = await this.executeHandler(message);
@@ -263,8 +263,7 @@ export class Consumer extends TypedEventEmitter {
         await this.changeVisibilityTimeout(message, 0);
       }
     } finally {
-      clearInterval(this.heartbeatTimeoutId);
-      this.heartbeatTimeoutId = undefined;
+      clearInterval(heartbeatTimeoutId);
     }
   }
 
@@ -273,13 +272,15 @@ export class Consumer extends TypedEventEmitter {
    * @param messages The messages that were delivered from SQS
    */
   private async processMessageBatch(messages: Message[]): Promise<void> {
+    let heartbeatTimeoutId: NodeJS.Timeout | undefined = undefined;
+
     try {
       messages.forEach((message) => {
         this.emit('message_received', message);
       });
 
       if (this.heartbeatInterval) {
-        this.heartbeatTimeoutId = this.startHeartbeat(null, messages);
+        heartbeatTimeoutId = this.startHeartbeat(null, messages);
       }
 
       const ackedMessages = await this.executeBatchHandler(messages);
@@ -298,8 +299,7 @@ export class Consumer extends TypedEventEmitter {
         await this.changeVisibilityTimeoutBatch(messages, 0);
       }
     } finally {
-      clearInterval(this.heartbeatTimeoutId);
-      this.heartbeatTimeoutId = undefined;
+      clearInterval(heartbeatTimeoutId);
     }
   }
 
@@ -387,12 +387,14 @@ export class Consumer extends TypedEventEmitter {
    * @param message The message that was received from SQS
    */
   private async executeHandler(message: Message): Promise<Message> {
+    let handleMessageTimeoutId: NodeJS.Timeout | undefined = undefined;
+
     try {
       let result;
 
       if (this.handleMessageTimeout) {
         const pending = new Promise((_, reject) => {
-          this.handleMessageTimeoutId = setTimeout((): void => {
+          handleMessageTimeoutId = setTimeout((): void => {
             reject(new TimeoutError());
           }, this.handleMessageTimeout);
         });
@@ -409,8 +411,8 @@ export class Consumer extends TypedEventEmitter {
           : `Unexpected message handler failure: ${err.message}`;
       throw err;
     } finally {
-      if (this.handleMessageTimeoutId) {
-        clearTimeout(this.handleMessageTimeoutId);
+      if (handleMessageTimeoutId) {
+        clearTimeout(handleMessageTimeoutId);
       }
     }
   }