How to promisify AWS SQS using Bluebird

Question

I have the following code that works. I started promisifying it using Bluebird, however, I am not sure how I can promisify the processing of the array of messages.

var s3 = new AWS.S3();
var sqs = new AWS.SQS();
 // This notification call is triggered by the latest message but there may
 // be earlier unprocessed messages. So, we request the maximum number of
 // messages (10) from the queue and process and then remove from the queue
 // all of them.
sqs.receiveMessage({
  QueueUrl: settings.sqsQueueUrl[prdOrDev],
  /* required */
  WaitTimeSeconds: 20, // to enable long polling, which polls all servers for any unprocessed SQS messages
  VisibilityTimeout: 120, // without this longpolling didn't work.
  MaxNumberOfMessages: 10
}, function(err, data) {
  if (err) {
    console.error('SQS receiveMessage failed: ', err, err.stack);
    return res.status(400).json({
      success: false
    });
  } else {
    var messages = data.Messages;
    messages.forEach(function(message) {
      var body = JSON.parse(message.Body);
      var sesMsg = JSON.parse(body.Message);
      s3.getObject({
        Bucket: sesMsg.receipt.action.bucketName,
        Key: sesMsg.receipt.action.objectKey
      }, function(err, data2) {
        if (err) {
          console.error('S3 getObject failed: ', err, err.stack);
        } else {
          sqs.deleteMessage({
            QueueUrl: settings.sqsQueueUrl[prdOrDev],
            /* required */
            ReceiptHandle: message.ReceiptHandle
          }, function(err, data) {
            if (err) {
              console.error('SQS deleteMessage failed: ', err, err.stack);
            }
          });
        }
      });
    });
  }
});

Here is my attempt at promisifying the code above:

var Promise = require('bluebird');
var s3 = new AWS.S3();
var sqs = new AWS.SQS();
Promise.promisifyAll(Object.getPrototypeOf(s3));
Promise.promisifyAll(Object.getPrototypeOf(sqs));

sqs.receiveMessageAsync({
  QueueUrl: settings.sqsQueueUrl[prdOrDev],
  /* required */
  WaitTimeSeconds: 20, // to enable long polling, which polls all servers for any unprocessed SQS messages
  VisibilityTimeout: 120, // without this longpolling didn't work.
  MaxNumberOfMessages: 10
}).then(function(data) {
  var messages = data.Messages;
  messages.forEach(function(message) {
    var body = JSON.parse(message.Body);
    var sesMsg = JSON.parse(body.Message);
    s3.getObjectAsync({
      Bucket: sesMsg.receipt.action.bucketName,
      Key: sesMsg.receipt.action.objectKey
    }).then(function(data2) {
      return sqs.deleteMessageAsync({
        QueueUrl: settings.sqsQueueUrl[prdOrDev],
        /* required */
        ReceiptHandle: message.ReceiptHandle
      }).catch(function(err) {
        console.log('SQS deleteMessage failed: ', err, err.stack);
      });
    }).catch(function(err) {
      console.log('S3 getObject failed: ', err, err.stack);
    });
  });
}).catch(function(err) {
  notifyAdmin('SQS receiveMessage failed: ', err, err.stack);
});

I am guessing that this is not the best way to use Promises. I am especially curious if there is a better way to deal with the forEach loop, similar to the following example from Bluebird's main page:

mongoClient.connectAsync('mongodb://localhost:27017/mydb')
    .then(function(db) {
        return db.collection('content').findAsync({})
    })
    .then(function(cursor) {
        return cursor.toArrayAsync();
    })
    .then(function(content) {
        res.status(200).json(content);
    })
    .catch(function(err) {
        throw err;
    });

So, how do I best promisify the code snippet at the top using Bluebird?

Answer 1

In the forEach loop, your then() function is breaking the chain, you're creating promises but you don't "wait" for them. The usual way is to store all promises in an array and use Promise.all(). So with your code:

sqs.receiveMessageAsync({
  QueueUrl: settings.sqsQueueUrl[prdOrDev],
  /* required */
  WaitTimeSeconds: 20, // to enable long polling, which polls all servers for any unprocessed SQS messages
  VisibilityTimeout: 120, // without this longpolling didn't work.
  MaxNumberOfMessages: 10
}).then(function(data) {
  var messages = data.Messages;
  var promises = [];
  messages.forEach(function(message) {
    var body = JSON.parse(message.Body);
    var sesMsg = JSON.parse(body.Message);

    var promise = s3.getObjectAsync({
      Bucket: sesMsg.receipt.action.bucketName,
      Key: sesMsg.receipt.action.objectKey
    }).then(function(data2) {
      return sqs.deleteMessageAsync({
        QueueUrl: settings.sqsQueueUrl[prdOrDev],
        /* required */
        ReceiptHandle: message.ReceiptHandle
      }).catch(function(err) {
        console.log('SQS deleteMessage failed: ', err, err.stack);
      });
    }).catch(function(err) {
      console.log('S3 getObject failed: ', err, err.stack);
    });

    promises.push(promise);
  });

  return Promise.all(promises);
}).then(function(result) {
  console.log('all done');
}).catch(function(err) {
  notifyAdmin('SQS receiveMessage failed: ', err, err.stack);
});

You can also simplify the promisify code to:

var s3 = Promise.promisifyAll(new AWS.S3());
var sqs = Promise.promisifyAll(new AWS.SQS());