Skip to content

A complete SQS poller implementation that makes message processing easier, written on top of AWS SDK V3

Notifications You must be signed in to change notification settings

0xkalvin/sqs-poller

Repository files navigation

sqs-poller

A complete SQS poller implementation to make message processing easier, written on top of Amazon SDK V3.

Installation

npm i --save sqs-poller

Usage

const { SQS } = require('@aws-sdk/client-sqs')
const { Poller } = require('sqs-poller')

const sqs = new SQS({
  endpoint: 'your sqs endpoint',
  region: 'your sqs region',
})

const poller = new Poller({
  queueUrl: 'your sqs endpoint + queue name',
  sqsClient: sqs,
})

poller.start({
  eachMessage: async (message) => {
    await doSomethingWithMessage(message)
  },
  // or
  eachBatch: async function (messages) {
    await doSomethingWithBatch(messages)
  }
})

poller.on('error', console.error)

API Reference

new Poller ([options])

  • options <object>
    • queueUrl <string> The complete URL for the queue, consisting of your AWS SQS endpoint + queue name. Required.
    • sqsClient <sqsClient> Instance of the sqsClient class provided by AWS SDK v3. Required.
    • messageAttributeNames <string | QueueAttributeName > The names of the message attributes. Optional.
    • maxNumberOfMessages <number> The maximum number of messages to return. Optional. Default: 10
    • visibilityTimeout <number> The duration (in seconds) that the received messages are hidden from subsequent retrieve requests after being retrieved by a ReceiveMessage request. Optional. Default: 20
    • waitTimeSeconds <number> The duration (in seconds) for which the call waits for a message to arrive in the queue before returning. Optional. Default: 10
    • pollingTimeout <number> The interval (in miliseconds) between a finished poll call and the next one. Optional. Default: 1
    • shutdownTimeout <number> The duration (in miliseconds) for which the poller should wait for in flight messages before stopping it Optional. Default: 5000

Create a new Poller class instance.

poller.start (options)

  • options <object>
    • eachMessage <Function> (message) => Promise<any> The handler to process one message at a time. Either eachMessage or eachBatch must be passed. If both are available, the eachMessage handler will be used. Conditionally Required
    • eachBatch <Function> (message) => Promise<any> The handler to process a message batch (up to ten messages) at a time. It is required if eachMessage is not passed. Conditionally Required
    • beforePoll <Function> (message) => Promise<any> The optional handler executed before performing a receiveMessage call. It can be used in scenarios where a business rule needs to be verified before polling new messages, such as rate limiting. Optional

Start message processing by passing a handler for each message.

poller.stop () : Promise

Stop message processing, waiting for in-fligth messages to be finished before resolving the promise. It can be used for proper graceful shutdown as following.

process.once('SIGTERM', async (signal) => {
  await poller.stop()

  console.log('Poller has stopped')

  process.exit(0)
})

Diagnostics Channels

The sqs-poller supports diagnostics channels (feature currently available only on Node.js v16+). It is the preferred way to instrument this package. The data is only published to these channels in case there are subscribers listening to them. The available channels are the following:

  • sqspoller:poller:eachMessage:start

    • Before a SQS message gets processed by the eachMessage handler, the SQS message gets published to this channel.
    const diagnosticsChannel = require('diagnostics_channel')
    
    diagnosticsChannel.channel('sqspoller:poller:eachMessage:start').subscribe(({ message }) => {
    console.log('body', message.Body)
    console.log('ReceiptHandle', message.ReceiptHandle)
    })
  • sqspoller:poller:eachMessage:end

    • When the eachMessage handler has been either resolved or rejected, the SQS message gets published to this channel.
    diagnosticsChannel.channel('sqspoller:poller:eachMessage:end').subscribe(({ message }) => {
    console.log('body', message.Body)
    console.log('ReceiptHandle', message.ReceiptHandle)
    })
  • sqspoller:poller:eachMessage:error

    • When the eachMessage handler rejects, the SQS message and the error get published to this channel.
    diagnosticsChannel.channel('sqspoller:poller:eachMessage:error').subscribe(({ message, error }) => {
    console.log('body', message.Body)
    console.log('ReceiptHandle', message.ReceiptHandle)
    console.log('error', error)
    })
  • sqspoller:poller:eachBatch:start

    • Before a SQS message batch gets processed by the eachBatch handler, the batch (messages) gets published to this channel.
  • sqspoller:poller:eachBatch:end

    • When the eachBatch resolves or rejects, the message batch (messages) gets published to this channel.
  • sqspoller:poller:eachBatch:error

    • When the eachBatch rejects, the message batch (messages) and error get published to this channel.
  • sqspoller:poller:deleteMessage:start

    • Before deleting a message from SQS, the message gets published to this channel.
  • sqspoller:poller:deleteMessage:end

    • When the deleteMessage call resolves or rejects, the message gets published to this channel.
  • sqspoller:poller:deleteMessage:error

    • When the deleteMessage call rejects, the message and error get published to this channel.
  • sqspoller:poller:deleteBatch:start

    • Before deleting a message batch from SQS, the batch (messages) gets published to this channel.
  • sqspoller:poller:deleteBatch:end

    • When the deleteBatch call resolves or rejects, the batch (messages) gets published to this channel.
  • sqspoller:poller:deleteBatch:error

    • When the deleteBatch call rejects, the batch (messages) and error get published to this channel.

About

A complete SQS poller implementation that makes message processing easier, written on top of AWS SDK V3

Topics

Resources

Stars

Watchers

Forks

Packages

No packages published