sqs-queue-parallel
v0.1.6
Published
Create a poll of Amazon SQS queue watchers and each one can receive 1+ messages
Downloads
5,641
Readme
sqs-queue-parallel.js
sqs-queue-parallel is a node.js library build on top of Amazon AWS SQS with concurrency and parallel message poll support.
You can create a poll of SQS queue watchers, each one can receive 1 or more messages from Amazon SQS.
With sqs-queue-parallel you need just to configure your AWS private keys, setup your one o more message
event callbacks and wait for new messages to be processed.
Example
var SqsQueueParallel = require('sqs-queue-parallel');
// Simple configuration:
// - 2 concurrency listeners
// - each listener can receive up to 4 messages
// With this configuration you could receive and parse 8 `message` events in parallel
var queue = new SqsQueueParallel({
name: "sqs-test",
maxNumberOfMessages: 4,
concurrency: 2
});
queue.on('message', function (e)
{
console.log('New message: ', e.metadata, e.data.MessageId)
e.deleteMessage(function(err, data) {
e.next();
});
});
queue.on('error', function (err)
{
console.log('There was an error: ', err);
});
Download
You can download and install this library using Node Package Manager (npm):
npm install sqs-queue-parallel --save
Summary
- new SqsQueueParallel(options = {})
- sendMessage(message = {}, callback)
- deleteMessage(receiptHandle, callback)
- changeMessageVisibility(receiptHandle, timeout, callback)
- client
- url
- connection
- connect
- message
- error
Global env:
- AWS_REGION
- AWS_ACCESS_KEY
- AWS_SECREY_KEY
Constructor
new SqsQueueParallel(options = {})
First you need to initialize a new object instance with a configuration.
Examples:
Constructing an object
var queue = new SqsQueueParallel({ name: 'sqs-test' });
Options Hash (options):
- name (String) — Required: name of the remote queue to be watched
- region (String) — the region to send/read service requests. Default is
process.env.AWS_REGION
- accessKeyId (String) — your AWS access key ID. Default is
process.env.AWS_ACCESS_KEY
- secretAccessKey (String) — your AWS secret access key. Default is
process.env.AWS_SECRET_KEY
- visibilityTimeout (Integer) — duration (in seconds) that the received messages are hidden from subsequent retrieve requests after being retrieved by a ReceiveMessage request.
- waitTimeSeconds (Integer) — duration (in seconds) for which the call will wait for a message to arrive in the queue before returning. If a message is available, the call will return sooner than WaitTimeSeconds. Default is 20
- maxNumberOfMessages (Integer) — maximum number of messages to return. Amazon SQS never returns more messages than this value but may return fewer. Default is 1
- concurrency (Integer) — number of concurrency fetcher to start. Default is 1
- debug (Boolean) — enable debug mode. Default is false
Important:
Each concurrency
queue can read maxNumberOfMessages
messages from Amazon SQS.
For example, 2 concurrency
queue with 5 maxNumberOfMessages
can trigger a max of 5 * 2 = 10 message
events; so it's very important to be carefull, expecially if you're working with I/O streams.
Properties
queue.client
Returns the SQS client object used by the queue.
queue.url
Url of the connected queue.
Methods
queue.sendMessage(params = {}, callback)
Build on the top of SQS.sendMessage()
allow you to easly push a message to the connected queue.
Parameters:
- params (Object)
body (Any type) — default to {}
An arbitrary message, could be a string, a number or a object.
delay (Integer)
The number of seconds (0 to 900 - 15 minutes) to delay a specific message. Messages with a positive DelaySeconds value become available for processing after the delay time is finished. If you don't specify a value, the default value for the queue applies
Callback (callback):
function(err, data) {}
For more information take checkout the official AWS documentation.
Esample:
var SqsQueueParallel = require('src/sqs-queue-parallel');
var queue = new SqsQueueParallel({ name: "sqs-test" });
queue.sendMessage({
body: 'my message',
delay: 10
});
queue.sendMessage({
body: [1, 2, 3]
}, function (err, data)
{
if (err)
console.log('There was a problem: ', err);
else
console.log('Item pushed', data);
});
queue.changeMessageVisibility(receiptHandle, timeout, callback)
Build on the top of SQS.changeMessageVisibility()
allow you to easly delay a message from the connected queue.
Parameters:
receipHandler (String)
The receipt handle associated with the message to delay.
timeout (Integer)
The new value (in seconds - from 0 to 43200 - maximum 12 hours) for the message's visibility timeout.
Callback (callback):
function(err, data) {}
For more information take checkout the official AWS documentation.
Esample:
var SqsQueueParallel = require('src/sqs-queue-parallel');
var queue = new SqsQueueParallel({ name: "sqs-test" });
queue.changeMessageVisibility('receipt-handle-to-delay-1', 30);
queue.on('message', function (job)
{
if (myTest is true)
job.deleteMessage();
else
job.changeMessageVisibility(30);
job.next();
});
queue.deleteMessage(receiptHandle, callback)
Build on the top of SQS.deleteMessage()
allow you to easly delete a message from the connected queue.
Parameters:
receipHandler (String)
The receipt handle associated with the message to delete.
Callback (callback):
function(err, data) {}
For more information take checkout the official AWS documentation.
Example:
var SqsQueueParallel = require('src/sqs-queue-parallel');
var queue = new SqsQueueParallel({ name: "sqs-test" });
queue.deleteMessage('receipt-handle-to-delete-1');
queue.on('message', function (job)
{
if (myTest is true)
job.deleteMessage(function(err, data) {
job.next();
});
});
Events
connection
function(urls) { }
Triggered when a connection is established with the remote server.
- urls (Array): list of all remotes urls
connect
function(url) { }
Triggered when the required queue name
is found in the remote list of queues.
- url (Object): url of the connected queue
message
function(message) { }
Event triggered each time a new message has been received from the remote queue.
- message (Object)
type (String): default is "Message"
data (Unknown): JSON.parsed message.Body or a string (if could not be parsed)
message (Object): reference to the received message
metadata (Object): reference to the metadata of the received message
name (String): name of the remote queue
url (String): url of the connected queue
deleteMessage(callback) (Function):
Helper to deleteMessage (or
SQS.deleteMessage()
) when the job is completed;callback
is the same of the publicdeleteMessage()
methodchangeMessageVisibility(timeout, callback) (Function):
Helper to changeMessageVisibility (or
SQS.changeMessageVisibility()
) when the job is completed;callback
is the same of the publicchangeMessageVisibility()
methoddelay(timeout, callback) (Function):
Helper to changeMessageVisibility (or
SQS.changeMessageVisibility()
) without completing the job;callback
is the same of the publicchangeMessageVisibility()
methodsendMessage(params = {}, callback) (Function): send a new message in the queue
next() (Function): call this method when you've completed your jobs in the event callback.
error
function(error) { }
License
(The MIT License)
Copyright (c) 2014 Luca Bigon
Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.