sqs-messenger

sqs-messenger

This library makes message sending/receiving in SQS/SNS easy.

Simple usage

const AWS = require('aws-sdk')
const SqsMessenger = require('sqs-messenger')

const sqs = new AWS.SQS({
  region: 'cn-north-1',
  sqs: '2012-11-05',
})

const sns = new AWS.SNS({
  region: 'cn-north-1',
  sns: '2010-03-31',
})

const sqsMessenger = new SqsMessenger({ sqs, sns }, {
  snsArnPrefix: 'arn:aws-cn:sns:cn-north-1:123456789012:',
  sqsArnPrefix: 'arn:aws-cn:sqs:cn-north-1:123456789012:',
  queueUrlPrefix: 'http://sqs.cn-north-1.amazonaws.com.cn/123456789012/',
  resourceNamePrefix: 'test_',
})

const myTopic = sqsMessenger.createTopic('myTopic')
const myQueue = sqsMessenger.createQueue('myQueue', {
  bindTopic: myTopic,
})


// register consumer on queue
sqsMessenger.on('myQueue', (message, done) => {
  // do something
  console.log(message)
  done()
})

// send message to topic
sqsMessenger.sendTopicMessage('myTopic', { text: 'a simple message send to topic' })

// send message to queue
sqsMessenger.sendQueueMessage('myQueue', { text: 'a simple message send directly to queue' })

Advanced usage

const myQueue = sqsMessenger.createQueue('myQueue', {
  bindTopics: [myTopic],
  withDeadLetter: true,
  maxReceiveCount: 3,
  delaySeconds: 30,
})

// batchSize
sqsMessenger.on('myQueue', (message, done) => {
  // do something
  console.log(message)
  done()
}, {
  batchSize: 10
})

// batch handling
sqsMessenger.on('myQueue', (messages, done) => {
  console.log(messages.length) // 10
  done()
}, {
  batchSize: 10,
  batchHandle: true,
})

myQueue.deadLetterQueue.onMessage((messsage, done)=> {
  // do something
  done()
})

// Error handling
sqsMessenger.onError(err => {
  console.log('Error handled')
  console.error(err.stack)
})

// Start multiple consumers for a queue
sqsMessenger.on('myQueue', (message, done) => {
  // do something
  done()
}, {
  consumers: 5
})

Graceful shutdown

shutdown queue with queue.shutdown(timeout):

const myQueue = sqsMessenger.createQueue('myQueue')
process.once('SIGTERM', () => {
  myQueue.shutdown(5000).then(() => {
    process.exit(0)
  })
})

or shutdown all queues with messenger.shutdown(timeout), each queue will have at most timeout time to cleanup:

process.once('SIGTERM', () => {
  sqsMessenger.shutdown(5000).then(() => {
    process.exit(0)
  })
})

Features

• Automatically create SNS topic, SQS queue and subscription
• Dead letter support
• Automatically acknowledge message on consumer finished
• Graceful shutdown support
• Batch sending(TODO)
• Message schema validation(TODO)