CODEWITHSUNDEEP
node.jsamazon-web-servicesamazon-sqs
squeekyDave
squeekyDave

Reputation: 962

Get all messages from AWS SQS in NodeJS

I have the following function that gets a message from aws SQS, the problem is I get one at a time and I wish to get all of them, because I need to check the ID for each message:

function getSQSMessages() {

    const params = {
        QueueUrl: 'some url',
    };

    sqs.receiveMessage(params, (err, data) => {
        if(err) {
            console.log(err, err.stack)
            return(err);
        }
        return data.Messages;
    });

};

function sendMessagesBack() {

    return new Promise((resolve, reject) => {
        if(Array.isArray(getSQSMessages())) {
            resolve(getSQSMessages());
        } else {
            reject(getSQSMessages());
        };
    });

};

The function sendMessagesBack() is used in another async/await function. I am not sure how to get all of the messages, as I was looking on how to get them, people mention loops but I could not figure how to implement it in my case. I assume I have to put sqs.receiveMessage() in a loop, but then I get confused on what do I need to check and when to stop the loop so I can get the ID of each message?

If anyone has any tips, please share. Thank you.

Upvotes: 5

Views: 10514

Answers (4)

Frito
Frito

Reputation: 1443

I know this is a bit of a necro but I landed here last night while trying to pull some all messages from a dead letter queue in SQS. While the accepted answer, "you cannot guarantee to get all messages" from the queue is absolutely correct I did want to drop an answer for anyone that may land here as well and needs to get around the 10 message limit per request from AWS.

Dependencies

In my case I have a few dependencies already in my project that I used to make life simpler.

  • lodash - This is something we use in our code for help making things functional. I don't think I used it below but I'm including it since it's in the file.
  • cli-progress - This gives you a nice little progress bar on your CLI.

Disclaimer

The below was thrown together during troubleshooting some production errors integrating with another system. Our DLQ messages contain some identifiers that I need in order to formulate cloud watch queries for troubleshooting. Given that these are two different GUIs in AWS switching back and forth is cumbersome given that our AWS session are via a form of federation and the session only lasts for one hour max.

The script

#!/usr/bin/env node

const _ = require('lodash');
const aswSdk = require('aws-sdk');
const cliProgress = require('cli-progress');

const queueUrl = 'https://[put-your-url-here]';
const queueRegion = 'us-west-1';

const getMessages = async (sqs) => {
  const resp = await sqs.receiveMessage({
    QueueUrl: queueUrl,
    MaxNumberOfMessages: 10,
  }).promise();

  return resp.Messages;
};

const main = async () => {
  const sqs = new aswSdk.SQS({ region: queueRegion });

  // First thing we need to do is get the current number of messages in the DLQ. 
  const attributes = await sqs.getQueueAttributes({
    QueueUrl: queueUrl,
    AttributeNames: ['All'], // Probably could thin this down but its late
  }).promise();

  const numberOfMessage = Number(attributes.Attributes.ApproximateNumberOfMessages);

  // Next we create a in-memory cache for the messages
  const allMessages = {};
  let running = true;

  // Honesty here: The examples we have in existing code use the multi-bar. It was about 10PM and I had 28 DLQ messages I was looking into. I didn't feel it was worth converting the multi-bar to a single-bar. Look into the docs on the github page if this is really a sticking point for you.
  const progress = new cliProgress.MultiBar({
    format: ' {bar} | {name} | {value}/{total}',
    hideCursor: true,
    clearOnComplete: true,
    stopOnComplete: true
  }, cliProgress.Presets.shades_grey);
  const progressBar = progress.create(numberOfMessage, 0, { name: 'Messages' });

  // TODO: put in a time limit to avoid an infinite loop. 
  // NOTE: For 28 messages I managed to get them all with this approach in about 15 seconds. When/if I cleanup this script I plan to add the time based short-circuit at that point.
  while (running) {
    // Fetch all the messages we can from the queue. The number of messages is not guaranteed per the AWS documentation. 
    let messages = await getMessages(sqs);
    for (let i = 0; i < messages.length; i++) {
      // Loop though the existing messages and only copy messages we have not already cached.
      let message = messages[i];
      let data = allMessages[message.MessageId];
      if (data === undefined) {
        allMessages[message.MessageId] = message;
      }
    }

    // Update our progress bar with the current progress
    const discoveredMessageCount = Object.keys(allMessages).length;
    progressBar.update(discoveredMessageCount);

    // Give a quick pause just to make sure we don't get rate limited or something
    await new Promise((resolve) => setTimeout(resolve, 1000));
    running = discoveredMessageCount !== numberOfMessage;
  }

  // Now that we have all the messages I printed them to console so I could copy/paste the output into LibreCalc (excel-like tool). I split on the semicolon for rows out of habit since sometimes similar scripts deal with data that has commas in it.
  const keys = Object.keys(allMessages);
  console.log('Message ID;ID');
  for (let i = 0; i < keys.length; i++) {
    const message = allMessages[keys[i]];
    const decodedBody = JSON.parse(message.Body);
    console.log(`${message.MessageId};${decodedBody.id}`);
  }
};

main();

Upvotes: 1

E.J. Brennan
E.J. Brennan

Reputation: 46841

You can never be guaranteed to get all the messages in a queue, unless after you get some of them, you delete them from the queue - thus ensuring that the next requests returns a different selection of records.

Each request will return 'upto' 10 messages, if you don't delete them, then there is a good chance that the next request for 'upto' 10 messages will return a mix of messages you have already seen, and some new ones - so you will never really know when you have seen them all.

It maybe that a queue is not the right tool to use in your use case - but since I don't know your use case, its hard to say.

Upvotes: 2

Swarup Bam
Swarup Bam

Reputation: 199

SQS doesn't return more than 10 messages in the response. To get all the available messages, you need to call the getSQSMessages function recursively. If you return a promise from getSQSMessages, you can do something like this.

getSQSMessages()
.then(data => {
  if(!data.Messages || data.Messages.length === 0){
      // no messages are available. return
  }
  // continue processing for each message or push the messages into array and call 
 //getSQSMessages function again. 
});

Upvotes: 4

Alexandru Olaru
Alexandru Olaru

Reputation: 7092

I suggest you to use the Promise api, and it will give you the possibility to use async/await syntax right away.

const { Messages } = await sqs.receiveMessage(params).promise();
// Messages will contain all your needed info

await sqs.sendMessage(params).promise();

In this way, you will not need to wrap the callback API with Promises.

Upvotes: 7

Related Questions