bigluck / sqs-queue-parallel Goto Github PK

Currently, there's no way to stop listening. AWS.Request exposes an abort() that can be used to stop the queue. I'd make a pull request, but I don't drink Coffee.

Maybe store active requests returned from self.client.receiveMessage, and expose a disconnect() method on the queue?

Btw, love your use of the async module. i don't leave home without it.

Running the following reduced test case will cause the node process to balloon it's memory usage, especially if it's processing messages. If it's just waiting for new messages, the process still see memory usage increases.

var SqsQueueParallel = require('sqs-queue-parallel');

var queue = new SqsQueueParallel({
  accessKeyId: process.env.AWS_ACCESS_KEY_ID,
  secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY,
  name: process.env.INCOMING_QUEUE_NAME,
  region: process.env.AWS_QUEUE_REGION,
  debug: process.env.DEBUG
});

queue.on("message", function(m) {
  m.deleteMessage(function(err) {
    if ( err ) {
      console.log("Error deleting message:\n%s", err.stack);
    }

    m.next();
  });
});

It appears this library uses sqs.listQueues and some convoluted regexp matching in order to discover the QueueUrl it is supposed to use.

The AWS API and SDK provide a sqs.getQueueUrl method, which accepts a queue name and returns the QueueUrl. No need to provide permissions to list all queues on the account, regexp match urls whose formats are not guaranteed, or discard QueueUrls for queues that happen to begin with the name you give but don't match.

I need something like this, but see a lot of open issues and not merged pull request...
Is this still a thing?

@sdossick @bigluck I am facing this issue only when I am using this library

message:
'Access to the resource https://sqs.us-east-2.amazonaws.com/ is denied.',
code: 'AccessDenied',
time: 2021-03-01T07:03:45.606Z,
requestId: '7e964ccb-0430-5a74-9859-147d90216795',
statusCode: 403,
retryable: false,
retryDelay: 86.40930855612243

This can cause the deleteMessage to get scheduled to be called after the visibility window has expired for each message.

next() should be called after deleteMessage() or not at all so it can be called in the callback for delete message

If I have long running work that I want to process in response to a message coming in from on('message'), what is the best practice for calling next()?

I have my max messages set to 10, and concurrency set to 1. In my simple tests though, it seems that it is only processing one message at a time until I call next().

The part of the documentation that is confusing me is this:
For example, 2 concurrency queue with 5 maxNumberOfMessages can trigger a max of 5 * 2 = 10 message events; so it's very important to be carefull, expecially if you're working with I/O streams.

My understanding of this would then be, if I have max messages set to 10, and concurrency to 1, that I should be getting on('message') called ten times in quick succession, however, the next on('message') event isn't being fired until the previous one calls next(). Is this expected? Is the documentation just misleading?

you need
try JSON.parse(message.Body) catch then message.Body

try {
JSON.parse(message.Body);
} catch (_error) {
message.Body;
}