SNSSQS-MSGBUS

A message bus using AWS SNS Topics and SQS Queues

Install

npm

$ npm install snssqs-msgbus

yarn

$ yarn add snssqs-msgbus

Usage

This lib provides two classes, MessageBus and MessageQueue. MessageBus will send command messages (AWS SNS), while MessageQueue will gather messages from a queue (AWS SQS).

Both MessageBus and MessageQueue will retrieve credentials from the Default credentials chain.

MessageBus

Send messages to services through SNS Topic.

import { createMessageBus, getMessageBus } from 'snssqs-msgbus';
// Provide the region ans SNS ARN
const msgBus = createMessageBus(process.env.AWS_REGION, process.env.AWS_SNS_ARN);

// the MessageBus instance can be reached from msgBus and also from
// getMessageBus(), which will return the previously created MessageBus instance.

// ...

// Somewhere in your code
const commandData = { whatheverContent: 1, someValue: 'this string' };

getMessageBus().sendCommand('PascalCasedCommandName', JSON.stringify(commandData));

// This will send a SNS message with Attribute command=PascalCasedCommandName and MessageBody=commandData

MessageQueue

Receive commands and data from a SQS Queue.

import { createMessageQueue, getMessageQueue } from 'snssqs-msgbus';

// Create the MesssageQueue instance
createMessageQueue(process.env.AWS_REGION);

// Create a subscription to a Queue.
// First arg is for logging purposes only
// Then provide Queue URL and callback. If callback is async, it will be awaited.
getMessageQueue().addSubscription('MyCommandName', process.env.MY_CMD_QUEUE_URL, (messageId, messageBody) => {
  console.log(`Received a message with MessageId:${messageId} and body: ${messageBody}`);
});

// To start polling queues invoke run()
getMessageQueue().run();

// It can be stopped also with getMessageQueue().stop();

AWS Setup:

1. Create SNS Topic_X

2. Create a SQS Queue_Y

3. On SNS Topic_X:

• Add subscription with Queue_Y's ARN

• Select Enable raw message delivery (it MUST BE SELECTED !)

• On subscription filter policy add this (which commands will enter the queue)

{
  "command": [
    "CommandName"
  ]
}

Where "CommandName" is the identifier to route an operation notification to certain queue. (command is a sns notification attribute, not a field on message body)

4. On SQS Queues > Queue_Y

• Select Access Policy, and edit with the following Policy

{
  "Version": "2012-10-17",
  "Id": "Policy1598555915018",
  "Statement": [
    {
      "Sid": "Stmt1598555833567",
      "Effect": "Allow",
      "Principal": "*",
      "Action": "sqs:SendMessage",
      "Resource": "<QUEUE_Y_ARN>",
      "Condition": {
        "StringEquals": {
          "aws:SourceArn": "<TOPIC_X_ARN>"
        }
      }
    },
    {
      "Sid": "Stmt1598555909987",
      "Effect": "Allow",
      "Principal": "*",
      "Action": [
        "sqs:ChangeMessageVisibility",
        "sqs:DeleteMessage",
        "sqs:GetQueueAttributes",
        "sqs:GetQueueUrl",
        "sqs:ReceiveMessage"
      ],
      "Resource": "<QUEUE_Y_ARN>"
    }
  ]
}

• The first statement allows SNS to send messages to the queue, and restricts sending messages only be from this SNS topic.
• The second defines who can receive messages from the Queue. (Adjust principal as needed to control access)

(Without the first Statement messages will NOT arrive to the queue)