handle endpoint option in config, fix delay integer parsing, improve shutdown behavior

Author: snlamm

package.json

@@ -16,7 +16,8 @@
     "beanstalk",
     "ampq",
     "queue",
-    "faktory"
+    "faktory",
+    "sqs"
   ],
   "peerDependencies": {
     "grind-cli": "^0.7.0",

src/Drivers/SQSDriver.js

@@ -31,8 +31,11 @@ export class SQSDriver extends BaseDriver {
 
 		const params = {
 			QueueUrl: queueUrl,
-			MessageBody: JSON.stringify(payload),
-			DelaySeconds: payload.delay || 0
+			MessageBody: JSON.stringify(payload)
+		}
+
+		if(payload.delay > 0) {
+			params.DelaySeconds = Math.round(payload.delay / 1000)
 		}
 
 		return this.client.put(params)
@@ -98,6 +101,7 @@ export class SQSDriver extends BaseDriver {
 	}
 
 	destroy() {
+		this.client.isShutdown = true
 		this.client.constructor.queueUrls = { }
 		return super.destroy()
 	}

src/Support/SQS.js

@@ -32,6 +32,7 @@ export class SQS {
 
 	client = null
 	queueConfigs = null
+	isShutdown = false
 
 	constructor(config) {
 		loadPackage()
@@ -47,6 +48,12 @@ export class SQS {
 			serviceConfig.secretAccessKey = config.secret_key
 		}
 
+		// Point endpoint to a mock SQS server for local testing
+		// for example, use elasticmq standalone server or docker image
+		if(!config.endpoint.isNil) {
+			serviceConfig.endpoint = config.endpoint
+		}
+
 		this.client = new sqs(serviceConfig)
 	}
 
@@ -161,6 +168,10 @@ export class SQS {
 	}
 
 	async watch(queue, concurrency, handler) {
+		if(this.isShutdown) {
+			return
+		}
+
 		const queueUrl = this.queueUrls[queue]
 
 		if(queueUrl.isNil) {
@@ -177,7 +188,11 @@ export class SQS {
 
 		const messages = await new Promise(resolve => {
 			return this.client.receiveMessage(params, (err, data) => {
-				if(err) {
+				if(!err.isNil) {
+					if(!this.isShutdown) {
+						Log.error(err)
+					}
+
 					return resolve()
 				} else if((data.Messages === void 0) || (data.Messages.length === 0)) {
 					return resolve()
@@ -187,6 +202,11 @@ export class SQS {
 			})
 		})
 
+		// No built-in way to interrupt receiveMessage polling once it starts, so shutdown exits before handling
+		if(this.isShutdown) {
+			return
+		}
+
 		if(Array.isArray(messages)) {
 			// Delete jobs from SQS so they are not reprocessed, then process them
 			await this.deleteFromQueue(queueUrl, messages)