Send Queue Messages in AWS SQS using Node JS to build a micro-service application
Hi there,
AWS is one of cloud computing platform that offers a variety of cloud computing services to developers and companies to build their application faster.
Following are few services offered by AWS:
- Computing (EC2)
- Storage
- Databases
- Analytics
- Machine Learning
- Networking
- Security
To get the AWS services, AWS offers both the console and command-line along with Software Development Kits (SDKs). AWS users are billed for how much resources they used.
Through offering services in different regions and availability zones AWS provides the high availability, scalability and resiliency.
In that way, through this article I am going to share about one of the service offered by AWS to send message queues. i.e: SQS (Simple Queue Service).
This service is a fully-managed messaging queuing service. This service enables the application to be decoupled between its component services and increases the scalability of the application through designing the application as micro-service architecture.
SQS allows the services to communicate with each other asynchronously by sending and receiving messages through a highly-scalable and secure queue. That means without connecting components of the application directly to each-other one component can send the message and the other component can able to receive it. This increases the fault tolerance and risk of data loss.
The SQS can be configured to do flexible configuration options, including message ordering, delivery guarantees and retention periods.
Following are two types of queues that are available in SQS;
- FIFO (First-In-First-Out): follows strict messaging order, exactly-one processing and limited number of transactions per seconds
- Standard: offers high throughput, best-effort ordering, unlimited no. of transactions per seconds
Lets build a sample back-end app that publishes a message to the AWS queue and another service that listens to that message queue and receives the message and process it and do further operations.
All before that lets create a queue in AWS console:
Login to the AWS console and Search for SQS:
Click the Simple Queue Service and then Click “Create queue”
Now, I am going to create a new Queue with Fifi (First In First Out) type.
Remember: Once you created a queue type you can’t change that type.
Now lets get into the coding phase:
Create a project folder named SQS-MICROSERVICE-NODEJS:
inside it create 2 services as provider and consumer.
We need to create package.json file in both the consumer & provider project seperately.
Here you need the following node js dependencies:
- express
- aws-sdk
- dotenv
Create a package.json file using
npm initThen execute the following command to install the packages
npm install - save express aws-sdk dotenv.env File also needs to be created inside both of the project:
- .env file for Provider project
AWS_ACCESS_KEY_ID=<aws_access_key>
AWS_SECRET_ACCESS_KEY=<aws_secret_access_key>
AWS_DEFAULT_REGION=<aws_default_region>
PROVIDER_PORT=3000- .env file for Consumer Project
AWS_ACCESS_KEY_ID=<aws_access_key>
AWS_SECRET_ACCESS_KEY=<aws_secret_access_key>
AWS_DEFAULT_REGION=<aws_default_region>
AWS_SQS_QUEUE_URL="https://sqs.ap-south-1.amazonaws.com/<aws_id>/<queue_name_that_created>"
CONSUMER_PORT=5000Provider project implementation:
Provider is a service that creates a new message and sends that message to AWS queue.
Create a new index.js file inside provider folder and paste the following code:
const dotenv = require('dotenv');
dotenv.config()
const express = require('express');
const app = express();
app.use(express.json())
const AWS = require('aws-sdk');
AWS.config.update({
accessKeyId: process.env.AWS_ACCESS_KEY_ID,
secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY,
region: process.env.AWS_DEFAULT_REGION
})
const listQueues = (queueName) => {
const sqs = new AWS.SQS();
return new Promise((resolve, reject) => {
sqs.listQueues({
QueueNamePrefix: queueName
}, (err, data) => {
if (err) {
reject(err)
}else {
resolve(data)
}
})
})
}
const createQueue = (queueName) => {
const sqs = new AWS.SQS();
return new Promise((resolve, reject) => {
sqs.createQueue({
QueueName: queueName
}, (err, data) => {
if (err) {
reject(err)
}else {
resolve(data)
}
})
})
}
const sendMessage = async (queueName, message) => {
const listOfQueues = await listQueues(queueName)
if ( !listOfQueues.QueryUrls ){
const qIsCreated = await createQueue(queueName);
console.log(qIsCreated);
}
const sqs = new AWS.SQS();
const params = {
QueryUrl: AWS_SQS_QUEUE_URL + queueName,
MessageBody: message
}
return new Promise((resolve, reject) => {
sqs.sendMessage(params, (err, data) => {
if (err) {
reject(err);
} else {
resolve(data);
}
})
})
}
app.post('/', async (req, res) => {
const result = await sendMessage(req.body.queueName, JSON.stringify({
message: req.body
}))
res.json({
message: result
})
})
app.listen(process.env.PROVIDER_PORT, () => {
console.log(`Provider is listening on port ${process.env.PROVIDER_PORT}`)
})In the provider index file, a new message will be send to a particular queue name (based on the value we made through a POST request).
listQueues function will list all the existing queues for a given queue name.
createQueue function will create a new queue for the given queue name.
sendMessage function will publish a new message to the queue.
Consumer Project Implementation:
const dotenv = require('dotenv');
dotenv.config()
const express = require('express');
const app = express();
app.use(express.json())
const AWS = require('aws-sdk');
AWS.config.update({
accessKeyId: process.env.AWS_ACCESS_KEY_ID,
secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY,
region: process.env.AWS_DEFAULT_REGION
})
const listQueues = (queueName) => {
const sqs = new AWS.SQS();
return new Promise((resolve, reject) => {
sqs.listQueues({
QueueNamePrefix: queueName
}, (err, data) => {
if (err) {
reject(err)
}else {
resolve(data)
}
})
})
}
const deleteMessage = async (queueName, receiptHandler) => {
const sqs = new AWS.SQS();
const params = {
QueryUrl: process.env.AWS_SQS_QUEUE_URL + queueName,
ReceiptHandler: receiptHandler
}
return new Promise((resolve, reject) => {
sqs.deleteMessage(params, (err, data) => {
if (err) {
reject(err)
}else {
resolve(data)
}
})
})
}
const receiveMessage = async (queueName) => {
const sqs = new AWS.SQS();
const params = {
QueryUrl: process.env.AWS_SQS_QUEUE_URL + queueName
}
return new Promise((resolve, reject) => {
sqs.receiveMessage(params, (err, data) => {
if (err) {
reject(err)
}else {
resolve(data)
}
})
})
}
app.get('/', async (req, res) => {
const result = await receiveMessage(req.query.queueName);
if (result.Message) {
const message = result.Message[0];
const receiptHandle = message.ReceiptHandler;
const messageBody = message.Body;
const deleteMessageResult = await deleteMessage(req.query.queueName, receiptHandle);
res.send(messageBody);
}else {
res.send("No message in queue");
}
})
app.listen(process.env.CONSUMER_PORT, () => {
console.log(`Consumer is listening on port ${process.env.CONSUMER_PORT}`)
})When we make a GET request to Consumer it will return all the messages in the queue.
deleteMessage function will delete the taken message from queue to ensure that the message has be processed by the receive so that the message will be totally removed from the queue.
We have done the very basic implementation to use AWS SQS in Node JS usgin AWS SDK for Node JS. But, to get the data from the queue we need to call it again and again. This process is something called as Polling. The polling strategy has to be implemented from developer side.
But there is a Node JS library that can do these things for us.
For Consumer: sqs-consumer
For Producer: sqs-producer
SQS-Consumer library is using a preferred strategy by AWS called long polling which the polling will wait for few seconds to hear from the queue whether any queues are there. This reduce the return of empty responses from queue.
