sqns

#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.

1. Queue Management
2. Notification Management

This project is inspired by AWS SQS and AWS SNS. This is an extension of the existing AWS SQS and AWS SNS with some more functionality add that I felt were missing.

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

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

2. Register routes with Express Server.

   sqns.registerExpressRoutes(app);
   

Queue Worker

1. Initialize Worker

   import { SQNSClient } from 'sqns';
   
   const sqnsClient = new SQNSClient({
    endpoint: 'http://your.server.url/api',
    accessKeyId: 'yourKey',
    secretAccessKey: 'yourSecretKey',
   });
   

2. Create Queue

   sqnsClient.createQueue({ QueueName: 'queueName' }).then(queue => {});
   

3. Send a Message to the queue

   sqnsClient.sendMessage({ QueueUrl: queue.QueueUrl, MessageBody: '123' });
   

4. 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.

1. SQNSClient options.
2. Queue Name and Initial Params json to which events will be added.
3. 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.
4. 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

1. SimpleQueueServerClient options.
2. Array of Queue Name to which events will be added.
3. Listener function that will be called with EventItem to be process.
4. 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

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

2. Register routes with Express Server

   sqns.registerExpressRoutes(app);
   

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

4. Create Topic

   client.createTopic({ Name: 'Topic1' })
   .then(topic => {})
   

5. 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

1. 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
})

subscribe

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