#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

Prerequisites

Express app
MongoDB Server

Installation

npm install sqns --save

Queue Management

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.

Queue Manager

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.
    },
});

Queue Worker

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]
});

Manager Scheduler

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 * * * * *');

Processing Scheduler

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

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.

Notification Manager

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 },
});

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

sqnsClient.createQueue({
  QueueName: 'queueName',
  Attributes: { attribute1: 'value1' }, // optional
  tags: { tag1: 'value2'}, // optional
}).then(queue => {});

CreateQueue Attributes

maxReceiveCount: Maximum number of time any event of the queue will be retried.

client.createQueue({ ..., Attributes: { maxReceiveCount: '2' } });

sendMessage

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',
});

sendMessageBatch

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',
    }, 
  ],
});

findMessageById

client.findMessageById({
  QueueUrl: queue.QueueUrl,
  MessageId: 'messageId',
}).then(response => {
  const MessageId = response.Messages[0].MessageId;
});

findMessageByDeduplicationId

client.findMessageByDeduplicationId({
  QueueUrl: queue.QueueUrl,
  MessageDeduplicationId: 'messageId',
}).then(response => {
  const MessageId = response.Messages[0].MessageId;
});

receiveMessage

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;
});

updateMessageById

client.updateMessageById({
  QueueUrl: queue.QueueUrl,
  MessageId: 'messageId',
}).then(response => {
  const MessageId = response.Messages[0].MessageId;
});

updateMessageByDeduplicationId

client.updateMessageByDeduplicationId({
  QueueUrl: queue.QueueUrl,
  MessageDeduplicationId: 'messageDeduplicationId',
}).then(response => {
  const MessageId = response.Messages[0].MessageId;
});

listQueues

client.listQueues({
  QueueNamePrefix: 'queuePrefix', // optional: queueName prefix to find list of queues
  NextToken: 'nextQuestToken', // optional: token from previous listQueue request.
})

deleteQueue

client.deleteQueue({ QueueUrl: queue.QueueUrl });

getQueueUrl

client.getQueueUrl({ QueueName: 'queueName' });

markEventSuccess

client.markEventSuccess(MessageId, queue.QueueUrl, 'success message');

markEventFailure

client.markEventFailure(MessageId, queue.QueueUrl, 'success message');

createTopic

client.createTopic({
  Name: 'Topic1', // topic name
  Attributes: { DisplayName: 'Topic One' }, // optional
  Tags: [{ Key: 'tag1', Value: 'value1' }], // optional
}).then(topic => {});

listTopics

client.listTopics({
  NextToken: 'nextToken' // optinal
})

getTopicAttributes

client.getTopicAttributes({ TopicArn: topic.TopicArn })

setTopicAttributes

client.setTopicAttributes({
  TopicArn: topic.TopicArn,
  AttributeName: 'DisplayName',
  AttributeValue: 'Updated Topic One',
})

deleteTopic

client.deleteTopic({ TopicArn: topic.TopicArn });

publish

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;
})

listSubscriptions

client.listSubscriptions({
  NextToken: 'NextToken' // optional
})

listSubscriptionsByTopic

client.listSubscriptionsByTopic({
 TopicArn: topic.TopicArn,
 NextToken: 'NextToken' // optional
});

confirmSubscription

client.confirmSubscription({
 TopicArn: 'topicArn',
 Token: 'verificationToken',
});

unsubscribe

client.unsubscribe({ SubscriptionArn: 'subscriptionArn' });

getPublish

client.getPublish({ MessageId: 'MessageId' })

getSubscription

client.getSubscription({ SubscriptionArn: 'subscriptionArn' })

markPublished

client.markPublished({ MessageId: 'MessageId' })

sqns