SQS Queue Listener

1import * as AWS from 'aws-sdk';
2import * as Consumer from 'sqs-consumer';
3import logService from './log-service';
4import databaseWriter from './database-writer';
5
6AWS.config.update({
7  region: 'us-east-1',
8  accessKeyId: process.env.AWS_ACCESS_KEY_ID,
9  secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY,
10});
11const sqs = new AWS.SQS({ apiVersion: '2012-11-05' });
12
13class QueueSubscriber {
14  private consumer: Consumer;
15  constructor() {
16    const queueURL = process.env.QUEUE_URI;
17    this.consumer = Consumer.create({
18      queueUrl: queueURL,
19      handleMessage: (message, done) => {
20        const body = message.Body;
21        const parsedMessage = JSON.parse(body);
22        databaseWriter.writeToDb(parsedMessage)
23          .then(result => {
24            done();
25          })
26          .catch(err => {
27            done(err); // cause the message to be left on the queue
28          });
29      },
30      sqs: new AWS.SQS()
31    });
32
33    this.consumer.on('error', (err) => {
34      logService.error(err.message);
35    });
36
37    this.consumer.start();
38  }
39}
40
41export default QueueSubscriber;