Merge pull request #34 from AthennaIO/develop

feat(driver): lock jobs inside drivers

Author: jlenon7

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@athenna/queue",
-  "version": "5.17.0",
+  "version": "5.18.0",
   "description": "The Athenna queue handler.",
   "license": "MIT",
   "author": "João Lenon <lenon@athenna.io>",

package.json

@@ -387,20 +387,52 @@ export class AwsSqsDriver extends Driver<SQSClient> {
 
     AwsSqsDriver.ackedIds.delete(job.id)
 
+    const heartbeatDelay = this.calculateHeartbeatDelay()
+
+    let heartbeatTimeout: NodeJS.Timeout
+
+    const startHeartbeat = () => {
+      if (heartbeatDelay <= 0) {
+        return
+      }
+
+      heartbeatTimeout = setInterval(() => {
+        this.changeJobVisibility(
+          job.id,
+          this.msToS(this.visibilityTimeout)
+        ).catch(() => {})
+      }, heartbeatDelay)
+    }
+
+    const stopHeartbeat = () => {
+      if (!heartbeatTimeout) {
+        return
+      }
+
+      clearInterval(heartbeatTimeout)
+      heartbeatTimeout = undefined
+    }
+
     try {
+      startHeartbeat()
+
       await processor({
         id: job.id,
         attempts: job.attempts,
         data: job.data
       })
 
+      stopHeartbeat()
+
       if (!AwsSqsDriver.ackedIds.has(job.id)) {
         await this.changeJobVisibility(
           job.id,
           this.msToS(this.noAckDelayMs + requeueJitterMs)
         )
       }
     } catch (err) {
+      stopHeartbeat()
+
       const receiveCount = Number(
         job.metadata.Attributes?.ApproximateReceiveCount ?? '1'
       )