Programatically create a SQS queue and subscribe it to a specific SNS topic

redis-lifecycle-events-to-sns-to-sqs.js

const AWS = require('aws-sdk');

// should be configurable
const ACCOUNT_ID = '';
const TOPIC_ARN = "arn:aws:sns:eu-central-1:"+ACCOUNT_ID+":"+TOPIC_ID;
const REGION = 'eu-central-1';

AWS.config.region = REGION;
AWS.config.apiVerison = 'latest';

const chain = new AWS.CredentialProviderChain();
chain.providers.push(new AWS.SharedIniFileCredentials()); // Only for dev purpose
chain.providers.push(new AWS.EC2MetadataCredentials({httpOptions: {timeout: 5000}}));

const sqs = new AWS.SQS();
const sns = new AWS.SNS();
const metadata = new AWS.MetadataService({httpOptions: {timeout: 5000}});

function getQueuePolicy(topicArn, queueName) {
  return {
    "Version": "2012-10-17",
    "Statement": [
      {
        "Effect": "Allow",
        "Principal": "*",
        "Action": [
          "SQS:SendMessage"
        ],
        "Resource": "arn:aws:sqs:"+REGION+":"+ACCOUNT_ID+":"+queueName,
        "Condition": {
          "ArnEquals": {
            "aws:SourceArn": topicArn
          }
        }
      }
    ]
  };
}

chain.resolve(function (err) {
  if (err) {
    console.error('resolve', err);
    return;
  }

  console.log('authorized!');

  metadata.request('/latest/meta-data/instance-id', function (err, instanceId) {
    if (err) {
      /*console.error(err);
      return;*/
      instanceId = 'i-9a51dc27';
    }

    createQueue(instanceId);
  });
});

function createQueue(instanceId) {
  const queueName = 'redis_notif_' + instanceId;
  const queueParams = {
    QueueName: queueName,
    Attributes: {
      MessageRetentionPeriod: new Number(60*60).toString(), // 1 hour
      Policy: JSON.stringify(getQueuePolicy(TOPIC_ARN, queueName))
    }
  };

  sqs.createQueue(queueParams, function (err, queue) {
    if (err) {
      console.error(err);
      return;
    }

    console.log(queue);

    sqs.getQueueAttributes({QueueUrl: queue.QueueUrl, AttributeNames: ['QueueArn']}, function(err, data) {
      if (err) {
        console.error(err);
        return;
      }

      console.log(data);

      subscribeToTopic(queue.QueueUrl, data.Attributes.QueueArn);
    });
  });
}

function subscribeToTopic(queueUrl, queueArn) {
  const params = {
    TopicArn: TOPIC_ARN,
    Protocol: "sqs",
    Endpoint: queueArn
  };

  sns.subscribe(params, function (err, subscription) {
    if (err) {
      console.error(err);
      return;
    }

    console.log(subscription);

    receiveMessage(queueUrl, subscription.SubscriptionArn);
  });
}

function receiveMessage(queueUrl, subscriptionArn) {
  sqs.receiveMessage({QueueUrl: queueUrl}, function (err, msg) {
    if (err) {
      console.error(err);
      return;
    }

    console.log(msg);

    deleteMessage(queueUrl, msg.Messages[0].ReceiptHandle);
  });
}

function deleteMessage(queueUrl, receiptHandle) {
  const params = {QueueUrl: queueUrl, ReceiptHandle: receiptHandle};
  console.log(params);

  sqs.deleteMessage(params, function (err, data) {
    if (err) {
      console.error(err);
      return;
    }

    console.log(data);

    // unsubscribeFromTopic(queueUrl, subscriptionArn);
  });
}

function unsubscribeFromTopic(queueUrl, subscriptionArn) {
  sns.unsubscribe({SubscriptionArn: subscriptionArn}, function (err, data) {
    if (err) {
      console.error(err);
      return;
    }

    console.log(data);

    deleteQueue(queueUrl);
  });
}

function deleteQueue(queueUrl) {
  sqs.deleteQueue({QueueUrl: queueUrl}, function (err, data) {
    if (err) {
      console.error(err);
      return;
    }

    console.log(data);
  })
}