SNSSQS-MSGBUS
A message bus using AWS SNS Topics and SQS Queues
Install
npm
$ npm install snssqs-msgbusyarn
$ yarn add snssqs-msgbusUsage
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=commandDataMessageQueue
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:
-
Create SNS
Topic_X -
Create a SQS
Queue_Y -
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)
- 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)