Giter Club home page Giter Club logo

wtsqs's Introduction

WTSQS

npm version Build Status Coverage Status Dependencies Dev Dependencies

Simplified SQS Wrapper and Async Worker manager.

Features:

  • Simple interface. ✅
  • Promise based. ✅
  • ES6. ✅
  • Optimized async worker. ✅

Install

# Using npm
$ npm install wtsqs --save

# Or using yarn
$ yarn add wtsqs

Classes

WTSQS

A simplified sqs wrapper with interface similar to a normal queue data structure.

WTSQSWorker

WTSQS worker job manager.

WTSQSWorker takes care of asynchronously fetching jobs from sqs while processing other jobs concurrently. It also takes care of deleting a job from the queue after successfully processing the message.

Typedefs

Message : Object

Received SQS Message

Job : Object

Worker Job

WTSQS

A simplified sqs wrapper with interface similar to a normal queue data structure.

Kind: global class


new WTSQS(options)

Constructs WTSQS object.

Param Type Default Description
options Object Options object.
options.url String SQS queue url.
[options.accessKeyId] String AWS access key id.
[options.secretAccessKey] String AWS secret access key.
[options.region] String us-east-1 AWS regions where queue exists.
[options.defaultMessageGroupId] String FIFO queues only. Default tag assigned to a message that specifies it belongs to a specific message group. If not provided random uuid is assigned to each message which doesn't guarantee order but allows parallelism.
[options.defaultVisibilityTimeout] Integer 60 Default duration (in seconds) that the received messages are hidden from subsequent retrieve requests.
[options.defaultPollWaitTime] Integer 10 Default duration (in seconds) for which read calls wait for a message to arrive in the queue before returning.
[options.sqsOptions] Object Additional options to extend/override the underlying SQS object creation.

Example

const { WTSQS } = require('wtsqs')

// The most simple way to construct a WTSQS object
const wtsqs = new WTSQS({
  url: '//queue-url',
  accessKeyId: 'AWS_ACCESS_KEY_ID',
  secretAccessKey: 'AWS_SECRET_ACCESS_KEY'
})

wtsqs.size() ⇒ Promise.<integer>

Get approximate total number of messages in the queue.

Kind: instance method of WTSQS
Example

const size = await wtsqs.size()
console.log(size) // output: 2

wtsqs.enqueueOne(payload, [options], [sqsOptions]) ⇒ Promise

Enqueue single payload in the queue.

Kind: instance method of WTSQS
See: SQS#sendMessage

Param Type Default Description
payload Object JSON serializable object.
[options] Object Options.
[options.messageGroupId] String Message group id to override default id.
[sqsOptions] Object {} Additional options to extend/override the underlying SQS sendMessage request.

Example

const myObj = { a: 1 }
await wtsqs.enqueueOne(myObj)

wtsqs.enqueueMany(payloads, [options], [sqsOptions]) ⇒ Promise

Enqueue batch of payloads in the queue.

Kind: instance method of WTSQS
See: SQS#sendMessageBatch

Param Type Default Description
payloads Array.<Object> Array of JSON serializable objects.
[options] Object Options object.
[options.messageGroupId] String Message group id to override default id.
[sqsOptions] Object {} Additional options to extend/override the underlying SQS sendMessageBatch request.

Example

const myObjList = [{ a: 1 }, { b: 3 }]
await wtsqs.enqueueMany(myObjList)

wtsqs.peekOne([options], [sqsOptions]) ⇒ Promise.<(Message|null)>

Retrieve single message without deleting it.

Kind: instance method of WTSQS
Returns: Promise.<(Message|null)> - Message object or null if queue is empty.

Param Type Default Description
[options] Object Options object.
[options.pollWaitTime] Integer Duration (in seconds) for which read call waits for a message to arrive in the queue before returning. If no messages are available and the wait time expires, the call returns successfully with an empty list of messages.
[options.visibilityTimeout] Integer Duration (in seconds) that the received messages are hidden from subsequent retrieve requests.
[sqsOptions] Object {} Additional options to extend/override the underlying SQS receiveMessage request.

Example

const myMessage = await wtsqs.peekOne()
console.log(myMessage)
// output:
{
  id: 'messageId',
  receiptHandle: 'messageReceiptHandle'
  md5: 'messageMD5',
  body: { a: 1 }
}

wtsqs.peekMany([maxNumberOfMessages], [options], [sqsOptions]) ⇒ Promise.<Array.<Message>>

Retrieve batch of messages without deleting them.

Kind: instance method of WTSQS
Returns: Promise.<Array.<Message>> - Array of retrieved messages.
See: SQS#receiveMessage

Param Type Default Description
[maxNumberOfMessages] Number 10 Maximum number of messages to retrieve. Must be between 1 and 10.
[options] Object Options object.
[options.pollWaitTime] Integer Duration (in seconds) for which read call waits for a message to arrive in the queue before returning. If no messages are available and the wait time expires, the call returns successfully with an empty list of messages.
[options.visibilityTimeout] Integer Duration (in seconds) that the received messages are hidden from subsequent retrieve requests.
[sqsOptions] Object {} Additional options to extend/override the underlying SQS receiveMessage request.

Example

const myMessageList = await wtsqs.peekMany(2)
console.log(myMessageList)
// output:
[
 {
   id: 'messageId',
   receiptHandle: 'messageReceiptHandle'
   md5: 'messageMD5',
   body: { a: 1 }
 },
 {
   id: 'messageId',
   receiptHandle: 'messageReceiptHandle'
   md5: 'messageMD5',
   body: { b: 3 }
 }
]

wtsqs.deleteOne(message) ⇒ Promise

Delete single message from queue.

Kind: instance method of WTSQS
See: SQS#deleteMessage

Param Type Description
message Message Message to be deleted

Example

const myMessage = await wtsqs.peekOne()
await wtsqs.deleteOne(myMessage)

wtsqs.deleteMany(messages) ⇒ Promise

Delete batch of messages from queue.

Kind: instance method of WTSQS
See: SQS#deleteMessageBatch

Param Type Description
messages Array.<Message> Messages to be deleted

Example

const myMessageList = await wtsqs.peekMany(2)
await wtsqs.deleteMany(myMessageList)

wtsqs.deleteAll() ⇒ Promise

Delete ALL messages in the queue.

NOTE: Can only be called once every 60 seconds.

Kind: instance method of WTSQS
See: SQS#purgeQueue
Example

await wtsqs.deleteAll()

wtsqs.popOne([options], [sqsOptions]) ⇒ Promise.<(Message|null)>

Retrieve single message and immediately delete it.

Kind: instance method of WTSQS
Returns: Promise.<(Message|null)> - Message object or null if queue is empty.

Param Type Default Description
[options] Object Options object.
[options.pollWaitTime] Integer Duration (in seconds) for which read call waits for a message to arrive in the queue before returning. If no messages are available and the wait time expires, the call returns successfully with an empty list of messages.
[options.visibilityTimeout] Integer Duration (in seconds) that the received messages are hidden from subsequent retrieve requests.
[sqsOptions] Object {} Additional options to extend/override the underlying SQS receiveMessage request.

Example

const myMessage = await wtsqs.popOne()
// The message no longer exists in queue
console.log(myMessage)
// output:
{
  id: 'messageId',
  receiptHandle: 'messageReceiptHandle'
  md5: 'messageMD5',
  body: { a: 1 }
}

wtsqs.popMany([maxNumberOfMessages], [options], [sqsOptions]) ⇒ Promise.<Array.<Message>>

Retrieve batch of messages and immediately delete them.

Kind: instance method of WTSQS
Returns: Promise.<Array.<Message>> - Array of retrieved messages.

Param Type Default Description
[maxNumberOfMessages] Number 10 Maximum number of messages to retrieve. Must be between 1 and 10.
[options] Object Options object.
[options.pollWaitTime] Integer Duration (in seconds) for which read call waits for a message to arrive in the queue before returning. If no messages are available and the wait time expires, the call returns successfully with an empty list of messages.
[options.visibilityTimeout] Integer Duration (in seconds) that the received messages are hidden from subsequent retrieve requests.
[sqsOptions] Object {} Additional options to extend/override the underlying SQS receiveMessage request.

Example

const myMessageList = await wtsqs.popMany(2)
// Messages no longer exist in queue
console.log(myMessageList)
// output:
[
 {
   id: 'messageId',
   receiptHandle: 'messageReceiptHandle'
   md5: 'messageMD5',
   body: { a: 1 }
 },
 {
   id: 'messageId',
   receiptHandle: 'messageReceiptHandle'
   md5: 'messageMD5',
   body: { b: 3 }
 }
]

WTSQSWorker

WTSQS worker job manager.

WTSQSWorker takes care of asynchronously fetching jobs from sqs while processing other jobs concurrently. It also takes care of deleting a job from the queue after successfully processing the message.

Kind: global class


new WTSQSWorker(options)

Constructs WTSQSWorker object.

Param Type Default Description
options Object Options object.
options.wtsqs WTSQS WTSQS instance to use for connecting to sqs.
[options.maxConcurrency] Integer 20 Maximum number of concurrent jobs.
[options.pollWaitTime] Integer 5 Duration (in seconds) for which read calls wait for a job to arrive in the queue before returning.
[options.visibilityTimeout] Integer 30 Duration (in seconds) that the received jobs are hidden from subsequent retrieve requests.
[options.logger] Object | String Object with debug, info, warn, error methods to use for logging. Or a string with log level to use default internal logger.

Example

const { WTSQS, WTSQSWorker } = require('wtsqs')

const wtsqs = new WTSQS({
  url: '//queue-url',
  accessKeyId: 'AWS_ACCESS_KEY_ID',
  secretAccessKey: 'AWS_SECRET_ACCESS_KEY'
})

const worker = new WTSQSWorker({ wtsqs })

worker.run(async (job) => {
 await someAsyncFunction(job.body)
 console.log(job)
})

worker.run(handler)

Start fetching and processing jobs.

Kind: instance method of WTSQSWorker

Param Type Description
handler runHandler Async function to process a single job.

worker.shutdown() ⇒ Promise

Shutsdown the worker and drain active jobs.

Kind: instance method of WTSQSWorker
Returns: Promise - Resolves when all active jobs have been drained.


WTSQSWorker~runHandler ⇒ Promise

Async callback function to process single job.

Kind: inner typedef of WTSQSWorker

Param Type Description
job Job A single job to process

Message : Object

Received SQS Message

Kind: global typedef
Properties

Name Type Description
id String Message id.
receiptHandle String Message receipt handle.
md5 String Message body md5 hash sum.
body Object Message body containing original payload.

Job : Object

Worker Job

Kind: global typedef
Properties

Name Type Description
id String Job id.
receiptHandle String Job receipt handle.
md5 String Job body md5 hash sum.
body Object Job body containing original payload.

wtsqs's People

Contributors

ali-essam avatar dependabot[bot] avatar

Stargazers

 avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar

Watchers

 avatar  avatar  avatar  avatar

wtsqs's Issues

Jobs with the same group id don't get processed sequentially

Hi Ali

when we have maxConcurrency > 1
i noticed that the worker process jobs with the same group id concurrently
but to process them in order they need be processed sequentially instead

i think that happen because when the worker receive multiple messages as a batch
the code uses promise.all to process all without considering messages with the same group id

Support extending job visibility timeout

When fetching a message from SQS, visibilityTimeout (Duration (in seconds) that the received messages are hidden from subsequent retrieve requests.) is determined at request time

In some cases, the timeout might not be clear during the request and the processing job might need more time to process the message. Currently, if the visibilityTimeout is less than the job processing time, that will cause the message to be back in the queue and visible to other workers to fetch again, causing double processing of the same message.

It's possible through SQS API to extend the timeout after fetching the message through ChangeMessageVisibility.

TODO: Implement this functionality in WTSQS.

enqueueMany fails if message list is empty

{ MissingParameter: The request must contain the parameter SendMessageBatchRequestEntry.1.Id.
    at Request.extractError 
....
  message: 'The request must contain the parameter SendMessageBatchRequestEntry.1.Id.',
  code: 'MissingParameter',
  time: 2019-06-19T09:58:56.933Z,
  requestId: 'XXXXXXX-XXXX-XXXX-XXXX-XXXXXXXX,
  statusCode: 400,
  retryable: false,
  retryDelay: 8.720774804845522 }

Job router worker handler

Utility function that facilitates routing different jobs to different handlers.

For example

const handler = new JobRouter({
  key: 'type',
  handlers: {
    'type_1': handler1Foo,
    'type_2': handler2Foo,
  },
  defaultHandler: defaultHandlerFoo
})

Recommend Projects

  • React photo React

    A declarative, efficient, and flexible JavaScript library for building user interfaces.

  • Vue.js photo Vue.js

    🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.

  • Typescript photo Typescript

    TypeScript is a superset of JavaScript that compiles to clean JavaScript output.

  • TensorFlow photo TensorFlow

    An Open Source Machine Learning Framework for Everyone

  • Django photo Django

    The Web framework for perfectionists with deadlines.

  • D3 photo D3

    Bring data to life with SVG, Canvas and HTML. 📊📈🎉

Recommend Topics

  • javascript

    JavaScript (JS) is a lightweight interpreted programming language with first-class functions.

  • web

    Some thing interesting about web. New door for the world.

  • server

    A server is a program made to process requests and deliver data to clients.

  • Machine learning

    Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.

  • Game

    Some thing interesting about game, make everyone happy.

Recommend Org

  • Facebook photo Facebook

    We are working to build community through open source technology. NB: members must have two-factor auth.

  • Microsoft photo Microsoft

    Open source projects and samples from Microsoft.

  • Google photo Google

    Google ❤️ Open Source for everyone.

  • D3 photo D3

    Data-Driven Documents codes.