#Introduction SQNS is a Simple Queue and Notification Service. It manages processing of queues and notification due to various events and subscriptions in distributed system. This service can be divided into following sections.
- Queue Management
- Notification Management
- Express app
- MongoDB Server
- npm install sqns --save
Queue management is a distributed multiple priority queue processing system. This is divided into two part Manager and Workers. Manager collects all the events sent via different channels. Worker requests events from Manager to process them. There can be only one Manager, but many Workers.
- Initialize queue manager
import { SQNS } from 'sqns'; const dataBaseConnectionConfig = {}; const sqns = new SQNS({ endpoint: 'http://your.server.url/api', adminSecretKeys: [{ accessKey: 'yourKey', secretAccessKey: 'yourSecretKey' }], db: { uri: 'DatabaseUri', config: dataBaseConnectionConfig }, sns: { // optional disable: true // disable SNS service initialization. }, });
- Register routes with Express Server.
sqns.registerExpressRoutes(app);
- Initialize Worker
import { SQNSClient } from 'sqns'; const sqnsClient = new SQNSClient({ endpoint: 'http://your.server.url/api', accessKeyId: 'yourKey', secretAccessKey: 'yourSecretKey', });
- Create Queue
sqnsClient.createQueue({ QueueName: 'queueName' }).then(queue => {});
- Send a Message to the queue
sqnsClient.sendMessage({ QueueUrl: queue.QueueUrl, MessageBody: '123' });
- Receive a Message form queue
client.receiveMessage({ QueueUrl: queue.QueueUrl, MaxNumberOfMessages: 1 }) .then(response => { const message = response.Messages[0] });
Either you can use the SQNSClient support to add the event in the queue or use ManagerEventScheduler to fetch events and add them into the queue periodically. ManagerEventScheduler constructor requires below parameters.
- SQNSClient options.
- Queue Name and Initial Params json to which events will be added.
- Listener function to fetch request related to Events that need to be added in the queue. This function returns array of two elements. First is pagination params for next call and second is items that need to be added in the queue. If items that need to be added in the queue length is zero then next time listener function will be called wih initial pagination params.
- Cron Interval (optional).
import { ManagerEventScheduler } from 'sqns';
...
new ManagerEventScheduler(
{
endpoint: 'http://xyz.abz/api', // Master Server address
accessKeyId: 'accessKey',
secretAccessKey: 'secretKey',
},
{ 'queueName': { page: 0 } },
(params) => {
const items = []// request events to be added in the queue
const eventItems = items.map(each => {
const requestItem = {
MessageBody: each.message,
DelaySeconds: 10, // optional
MessageAttributes: { attribute1: { StringValue: 'value1', DataType: 'String' } }, // optional
MessageSystemAttributes: { attribute1: { StringValue: 'value1', DataType: 'String' } }, // optional
MessageDeduplicationId: each.duplicationId, // optional
};
return requestItem;
});
return [{ "page": params.page + 1 }, eventItems];
},
'*/10 * * * * *');
Either you can use SimpleQueueServerClient support to fetch the event from Manager Server or use WorkerEventScheduler to fetch events and process them periodically. WorkerEventScheduler constructor requires below parameters
- SimpleQueueServerClient options.
- Array of Queue Name to which events will be added.
- Listener function that will be called with EventItem to be process.
- Cron Interval (optional).
import { WorkerEventScheduler } from 'sqns';
...
new WorkerEventScheduler(
{
endpoint: 'http://xyz.abz/api', // Master Server address
accessKeyId: 'accessKey',
secretAccessKey: 'secretKey',
},
["queueName"],
(queueName, item) => {
// process eventItem
},
'0 * * * *');
Notification Management deals with passing one published event to many subscribed links. This uses the Queue Management module for passing published events to its subscribers.
- Initialize queue manager
import { SQNS } from 'sqns'; const dataBaseConnectionConfig = {}; const sqns = new SQNS({ endpoint: 'http://your.server.url/api', adminSecretKeys: [{ accessKey: 'yourKey', secretAccessKey: 'yourSecretKey' }], db: { uri: 'DatabaseUri', config: dataBaseConnectionConfig }, });
- Register routes with Express Server
sqns.registerExpressRoutes(app);
- Notification Scheduler
import { WorkerEventScheduler } from 'sqns'; ... new WorkerEventScheduler( { endpoint: 'http://xyz.abz/api', // Master Server address accessKeyId: 'accessKey', secretAccessKey: 'secretKey', }, ["sqns"], // SNS queue name (queueName, item) => { // process eventItem }, '0 * * * *');
- Create Topic
client.createTopic({ Name: 'Topic1' }) .then(topic => {})
- Publish Message
client.publish({ Message: 'This is message' })
sqnsClient.createQueue({
QueueName: 'queueName',
Attributes: { attribute1: 'value1' }, // optional
tags: { tag1: 'value2'}, // optional
}).then(queue => {});
- maxReceiveCount: Maximum number of time any event of the queue will be retried.
client.createQueue({ ..., Attributes: { maxReceiveCount: '2' } });
client.sendMessage({
QueueUrl: queue.QueueUrl,
MessageAttributes: { // optional
attribute1: {
StringValue: 'value1',
DataType: 'String'
}
},
MessageSystemAttributes: { // optional
attribute1: {
StringValue: 'attributeValue',
DataType: 'String'
}
},
MessageDeduplicationId: 'uniqueId1', // optional: message unique Id to avoid duplication of message.
MessageBody: 'This is message body',
});
client.sendMessageBatch({
QueueUrl: queue.QueueUrl,
Entries: [
{
Id: '123', // Entry id for the request. Should be unique in one call.
MessageAttributes: { // optional
attribute1: {
StringValue: 'value1',
DataType: 'String'
}
},
MessageSystemAttributes: { // optional
attribute1: {
StringValue: 'attributeValue',
DataType: 'String'
}
},
MessageDeduplicationId: 'uniqueId1', // optional: message unique Id to avoid duplication of message.
MessageBody: 'This is message body',
},
],
});
client.findMessageById({
QueueUrl: queue.QueueUrl,
MessageId: 'messageId',
}).then(response => {
const MessageId = response.Messages[0].MessageId;
});
client.findMessageByDeduplicationId({
QueueUrl: queue.QueueUrl,
MessageDeduplicationId: 'messageId',
}).then(response => {
const MessageId = response.Messages[0].MessageId;
});
client.receiveMessage({
QueueUrl: queue.QueueUrl,
AttributeNames: ['attributeValue'], // optional: Array of attribute name to include. ALL to add all attributes
MessageSystemAttributes: ['ALL'], // optional: Array of attribute name to include. ALL to add all attributes
MaxNumberOfMessages: 10, // optional: number of messages to fetch in one call
VisibilityTimeout: 30, // optional: message not available for next 30 seconds to retry.
}).then(response => {
const MessageId = response.Messages[0].MessageId;
});
client.updateMessageById({
QueueUrl: queue.QueueUrl,
MessageId: 'messageId',
}).then(response => {
const MessageId = response.Messages[0].MessageId;
});
client.updateMessageByDeduplicationId({
QueueUrl: queue.QueueUrl,
MessageDeduplicationId: 'messageDeduplicationId',
}).then(response => {
const MessageId = response.Messages[0].MessageId;
});
client.listQueues({
QueueNamePrefix: 'queuePrefix', // optional: queueName prefix to find list of queues
NextToken: 'nextQuestToken', // optional: token from previous listQueue request.
})
client.deleteQueue({ QueueUrl: queue.QueueUrl });
client.getQueueUrl({ QueueName: 'queueName' });
client.markEventSuccess(MessageId, queue.QueueUrl, 'success message');
client.markEventFailure(MessageId, queue.QueueUrl, 'success message');
client.createTopic({
Name: 'Topic1', // topic name
Attributes: { DisplayName: 'Topic One' }, // optional
Tags: [{ Key: 'tag1', Value: 'value1' }], // optional
}).then(topic => {});
client.listTopics({
NextToken: 'nextToken' // optinal
})
client.getTopicAttributes({ TopicArn: topic.TopicArn })
client.setTopicAttributes({
TopicArn: topic.TopicArn,
AttributeName: 'DisplayName',
AttributeValue: 'Updated Topic One',
})
client.deleteTopic({ TopicArn: topic.TopicArn });
client.publish({
Message: 'This is message',
TopicArn: topic.TopicArn, // TopicArn Or TargetArn is required
TargetArn: topic.TopicArn, // TopicArn Or TargetArn is required
MessageAttributes: { key1: { DataType: 'String', StringValue: 'value' } }, // optional
})
client.subscribe({
TopicArn: topic.TopicArn,
Attributes: { key: 'value' }, // optional
Endpoint: 'http://your.server.subscription/url', // subscription url
Protocol: 'http', // http or https
ReturnSubscriptionArn: true, // optional
}).then(result => {
const SubscriptionArn = result.SubscriptionArn;
})
client.listSubscriptions({
NextToken: 'NextToken' // optional
})
client.listSubscriptionsByTopic({
TopicArn: topic.TopicArn,
NextToken: 'NextToken' // optional
});
client.unsubscribe({ SubscriptionArn: 'subscriptionArn' });
client.getPublish({ MessageId: 'MessageId' })
client.getSubscription({ SubscriptionArn: 'subscriptionArn' })
client.markPublished({ MessageId: 'MessageId' })