Socket
Socket
Sign inDemoInstall

sqs-consumer

Package Overview
Dependencies
Maintainers
2
Versions
104
Alerts
File Explorer

Advanced tools

License
Socket logo

Install Socket

Detect and block malicious and high-risk dependencies

Install

sqs-consumer - npm Package Compare versions

Comparing version 5.7.0 to 5.8.0

.github/CODE_OF_CONDUCT.md
.github/SECURITY.md
.github/workflows/coverage.yml
.github/workflows/stale.yml
.github/workflows/test.yml
.prettierignore
.prettierrc.js
test/consumer.test.ts

26

.github/CONTRIBUTING.md
# Contributing
Thank you for your interest in contributing to the sqs-consumer.
Thank you for your interest in contributing to the sqs-consumer.
* If you're unsure if a feature would make a good addition, you can always [create an issue](https://github.com/bbc/sqs-consumer/issues/new) first. Raising an issue before creating a pull request is recommended.
* We aim for 100% test coverage. Please write tests for any new functionality or changes.
* Any API changes should be fully documented.
* Make sure your code meets our linting standards. Run `npm run lint` to check your code.
* Maintain the existing coding style. There are some settings in `.jsbeautifyrc` to help.
* Be mindful of others when making suggestions and/or code reviewing.
- If you're unsure if a feature would make a good addition, you can always [create an issue](https://github.com/bbc/sqs-consumer/issues/new) first. Raising an issue before creating a pull request is recommended.
- We aim for 100% test coverage. Please write tests for any new functionality or changes.
- Any API changes should be fully documented.
- Make sure your code meets our linting standards. Run `npm run lint` to check your code.
- Maintain the existing coding style. There are some settings in `.jsbeautifyrc` to help.
- Be mindful of others when making suggestions and/or code reviewing.
## Reporting Issues
## Reporting Issues
Before opening a new issue, first check that there is not already an [open issue or Pull Request](https://github.com/bbc/sqs-consumer/issues?utf8=%E2%9C%93&q=is%3Aopen) that addresses it.
If there is, make relevant comments and add your reaction. Use a reaction in place of a "+1" comment:
* 👍 - upvote
* 👎 - downvote
- 👍 - upvote
- 👎 - downvote
If you cannot find an existing issue that describes your bug or feature, create a new issue using the guidelines below.

@@ -28,4 +28,5 @@

## Contributing Code
If you do not have push access to the repository, please [fork it](https://help.github.com/en/articles/fork-a-repo). You should then work on your own `master` branch.
If you do not have push access to the repository, please [fork it](https://help.github.com/en/articles/fork-a-repo). You should then work on your own `main` branch.
Otherwise, you may clone this repository and create a working branch with a _kebab-case_ name reflecting what you are working on (e.g. `fix-the-thing`).

@@ -38,4 +39,5 @@

## Pull Request Process
1. Make sure you have opened an issue and it was approved by a project maintainer before working on a PR
2. Read and complete all relevant sections of the PR template
3. Wait for the PR get approved

4

.github/ISSUE_TEMPLATE/bug-report.md

@@ -7,3 +7,2 @@ ---

assignees: ''
---

@@ -16,2 +15,3 @@

Steps to reproduce the behaviour:
1. Go to '...'

@@ -29,2 +29,2 @@ 2. Select '....'

**Additional context**
Add any other context about the problem here, such as specific device information.
Add any other context about the problem here, such as specific device information.

1

.github/ISSUE_TEMPLATE/feature-request.md

@@ -7,3 +7,2 @@ ---

assignees: ''
---

@@ -10,0 +9,0 @@

1

.github/ISSUE_TEMPLATE/technical-question.md

@@ -7,3 +7,2 @@ ---

assignees: ''
---

@@ -10,0 +9,0 @@

31

.github/pull_request_template.md

@@ -1,15 +0,24 @@

<!--- Provide a general summary of your changes in the Title above -->
## Description
<!--- Describe your changes in detail -->
## Motivation and Context
<!--- Why is this change required? What problem does it solve? -->
<!--- If it fixes an open issue, please link to the issue here. -->
## Types of changes
<!--- What types of changes does your code introduce? Put an `x` in all the boxes that apply: -->
Resolves #NUMBER
**Description:**
_A very high-level summary of easily-reproducible changes that can be understood by non-devs._
**Type of change:**
- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)
## Checklist:
<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
<!--- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->
**Why is this change required?:**
_A simple explanation of what the problem is and how this PR solves it_
**Code changes:**
- _A bullet point list of key code changes that have been made._
- _When describing code changes, try to communicate **how** and **why** you implemented something a specific way, not just **what** has changed._
---
**Checklist:**
- [ ] My code follows the code style of this project.

@@ -16,0 +25,0 @@ - [ ] My change requires a change to the documentation.

20

dist/consumer.d.ts
/// <reference types="node" />
import * as SQS from 'aws-sdk/clients/sqs';
import { SQS } from 'aws-sdk';
import { EventEmitter } from 'events';
export declare type SQSMessage = SQS.Types.Message;
export type SQSMessage = SQS.Types.Message;
export interface ConsumerOptions {

@@ -25,10 +25,10 @@ queueUrl?: string;

interface Events {
'response_processed': [];
'empty': [];
'message_received': [SQSMessage];
'message_processed': [SQSMessage];
'error': [Error, void | SQSMessage | SQSMessage[]];
'timeout_error': [Error, SQSMessage];
'processing_error': [Error, SQSMessage];
'stopped': [];
response_processed: [];
empty: [];
message_received: [SQSMessage];
message_processed: [SQSMessage];
error: [Error, void | SQSMessage | SQSMessage[]];
timeout_error: [Error, SQSMessage];
processing_error: [Error, SQSMessage];
stopped: [];
}

@@ -35,0 +35,0 @@ export declare class Consumer extends EventEmitter {

56

dist/consumer.js
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.Consumer = void 0;
const SQS = require("aws-sdk/clients/sqs");
const Debug = require("debug");
const aws_sdk_1 = require("aws-sdk");
const debug_1 = require("debug");
const events_1 = require("events");
const bind_1 = require("./bind");
const errors_1 = require("./errors");
const debug = Debug('sqs-consumer');
const debug = (0, debug_1.default)('sqs-consumer');
const requiredOptions = [

@@ -34,3 +34,4 @@ 'queueUrl',

}
if (options.heartbeatInterval && !(options.heartbeatInterval < options.visibilityTimeout)) {
if (options.heartbeatInterval &&
!(options.heartbeatInterval < options.visibilityTimeout)) {
throw new Error('heartbeatInterval must be less than visibilityTimeout.');

@@ -41,3 +42,5 @@ }

if (err instanceof errors_1.SQSError) {
return (err.statusCode === 403 || err.code === 'CredentialsError' || err.code === 'UnknownEndpoint');
return (err.statusCode === 403 ||
err.code === 'CredentialsError' ||
err.code === 'UnknownEndpoint');
}

@@ -73,12 +76,16 @@ return false;

this.visibilityTimeout = options.visibilityTimeout;
this.terminateVisibilityTimeout = options.terminateVisibilityTimeout || false;
this.terminateVisibilityTimeout =
options.terminateVisibilityTimeout || false;
this.heartbeatInterval = options.heartbeatInterval;
this.waitTimeSeconds = (_a = options.waitTimeSeconds) !== null && _a !== void 0 ? _a : 20;
this.authenticationErrorTimeout = (_b = options.authenticationErrorTimeout) !== null && _b !== void 0 ? _b : 10000;
this.authenticationErrorTimeout =
(_b = options.authenticationErrorTimeout) !== null && _b !== void 0 ? _b : 10000;
this.pollingWaitTimeMs = (_c = options.pollingWaitTimeMs) !== null && _c !== void 0 ? _c : 0;
this.shouldDeleteMessages = (_d = options.shouldDeleteMessages) !== null && _d !== void 0 ? _d : true;
this.sqs = options.sqs || new SQS({
region: options.region || process.env.AWS_REGION || 'eu-west-1'
});
bind_1.autoBind(this);
this.sqs =
options.sqs ||
new aws_sdk_1.SQS({
region: options.region || process.env.AWS_REGION || 'eu-west-1'
});
(0, bind_1.autoBind)(this);
}

@@ -155,5 +162,3 @@ emit(event, ...args) {

try {
return await this.sqs
.receiveMessage(params)
.promise();
return await this.sqs.receiveMessage(params).promise();
}

@@ -175,5 +180,3 @@ catch (err) {

try {
await this.sqs
.deleteMessage(deleteParams)
.promise();
await this.sqs.deleteMessage(deleteParams).promise();
}

@@ -190,6 +193,3 @@ catch (err) {

[timeout, pending] = createTimeout(this.handleMessageTimeout);
await Promise.race([
this.handleMessage(message),
pending
]);
await Promise.race([this.handleMessage(message), pending]);
}

@@ -262,5 +262,7 @@ else {

return;
}).then(() => {
})
.then(() => {
setTimeout(this.poll, currentPollingTimeout);
}).catch((err) => {
})
.catch((err) => {
this.emit('error', err);

@@ -310,5 +312,3 @@ });

try {
await this.sqs
.deleteMessageBatch(deleteParams)
.promise();
await this.sqs.deleteMessageBatch(deleteParams).promise();
}

@@ -338,5 +338,3 @@ catch (err) {

try {
return await this.sqs
.changeMessageVisibilityBatch(params)
.promise();
return await this.sqs.changeMessageVisibilityBatch(params).promise();
}

@@ -343,0 +341,0 @@ catch (err) {

1

dist/index.js
"use strict";
Object.defineProperty(exports, "__esModule", { value: true });
exports.Consumer = void 0;
var consumer_1 = require("./consumer");
Object.defineProperty(exports, "Consumer", { enumerable: true, get: function () { return consumer_1.Consumer; } });

76

package.json
{
"name": "sqs-consumer",
"version": "5.7.0",
"version": "5.8.0",
"description": "Build SQS-based Node applications without the boilerplate",

@@ -14,7 +14,9 @@ "main": "dist/index.js",

"test": "mocha --recursive --full-trace --exit",
"coverage": "nyc mocha && nyc report --reporter=html && nyc report --reporter=json-summary",
"lcov": "nyc mocha && nyc report --reporter=lcov",
"lint": "eslint . --ext .ts",
"lint:fix": "eslint . --fix",
"coverage": "nyc mocha && nyc report --reporter=html && nyc report --reporter=json-summary",
"lcov": "nyc mocha && nyc report --reporter=lcov",
"posttest": "npm run lint"
"format": "prettier --loglevel warn --write \"**/*.{js,json,jsx,md,ts,tsx,html}\"",
"format:check": "prettier --check \"**/*.{js,json,jsx,md,ts,tsx,html}\"",
"posttest": "npm run lint && npm run format:check"
},

@@ -36,28 +38,30 @@ "repository": {

"devDependencies": {
"@types/chai": "^4.2.11",
"@types/debug": "^4.1.5",
"@types/mocha": "^7.0.2",
"@types/node": "^14.0.13",
"@types/sinon": "^9.0.4",
"@types/typescript": "^2.0.0",
"chai": "^4.2.0",
"codeclimate-test-reporter": "^0.5.1",
"eslint": "^7.2.0",
"eslint-config-iplayer-ts": "^2.0.0",
"mocha": "^8.0.1",
"@types/chai": "^4.3.4",
"@types/debug": "^4.1.7",
"@types/mocha": "^10.0.1",
"@types/node": "^16.18.7",
"@types/sinon": "^10.0.13",
"chai": "^4.3.7",
"eslint": "^8.29.0",
"eslint-config-iplayer-ts": "^4.1.0",
"eslint-config-prettier": "^4.3.0",
"mocha": "^10.1.0",
"nyc": "^15.1.0",
"p-event": "^4.2.0",
"sinon": "^9.0.2",
"ts-node": "^8.10.2",
"tslint-config-airbnb": "^5.11.2",
"tslint-microsoft-contrib": "^6.2.0",
"typescript": "^3.9.5",
"aws-sdk": "^2.1114.0"
"prettier": "^2.8.1",
"sinon": "^15.0.0",
"ts-node": "^10.9.1",
"typescript": "^4.9.4"
},
"dependencies": {
"debug": "^4.3.1"
"aws-sdk": "^2.1271.0",
"debug": "^4.3.4"
},
"peerDependencies": {
"aws-sdk": "^2.1114.0"
"aws-sdk": "^2.1271.0"
},
"mocha": {
"spec": "test/**/**/*.test.ts",
"require": "ts-node/register"
},
"nyc": {

@@ -77,12 +81,26 @@ "include": [

"eslintConfig": {
"extends": "iplayer-ts",
"extends": [
"iplayer-ts",
"prettier",
"prettier/react",
"prettier/@typescript-eslint"
],
"parserOptions": {
"ecmaVersion": 2017,
"sourceType": "module"
},
"rules": {
"@typescript-eslint/naming-convention": [
"error",
{
"selector": "variable",
"format": [
"camelCase",
"UPPER_CASE",
"PascalCase"
],
"leadingUnderscore": "allow"
}
]
}
},
"mocha": {
"spec": "test/**/**/*.ts",
"require": "ts-node/register"
}
}

80

README.md
# sqs-consumer
[![NPM downloads](https://img.shields.io/npm/dm/sqs-consumer.svg?style=flat)](https://npmjs.org/package/sqs-consumer)
[![Build Status](https://travis-ci.org/bbc/sqs-consumer.svg)](https://travis-ci.org/bbc/sqs-consumer)
[![Code Climate](https://codeclimate.com/github/BBC/sqs-consumer/badges/gpa.svg)](https://codeclimate.com/github/BBC/sqs-consumer)
[![Build Status](https://github.com/bbc/sqs-consumer/actions/workflows/test.yml/badge.svg?branch=main)](https://github.com/bbc/sqs-consumer/actions/workflows/test.yml)
[![Code Climate](https://codeclimate.com/github/BBC/sqs-consumer/badges/gpa.svg)](https://codeclimate.com/github/BBC/sqs-consumer)
[![Test Coverage](https://codeclimate.com/github/BBC/sqs-consumer/badges/coverage.svg)](https://codeclimate.com/github/BBC/sqs-consumer)

@@ -13,3 +13,3 @@

```bash
npm install sqs-consumer --save
npm install sqs-consumer --save-dev
```

@@ -40,7 +40,8 @@

* The queue is polled continuously for messages using [long polling](http://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-long-polling.html).
* Messages are deleted from the queue once the handler function has completed successfully.
* Throwing an error (or returning a rejected promise) from the handler function will cause the message to be left on the queue. An [SQS redrive policy](http://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/SQSDeadLetterQueue.html) can be used to move messages that cannot be processed to a dead letter queue.
* By default messages are processed one at a time – a new message won't be received until the first one has been processed. To process messages in parallel, use the `batchSize` option [detailed below](#options).
* By default, the default Node.js HTTP/HTTPS SQS agent creates a new TCP connection for every new request ([AWS SQS documentation](https://docs.aws.amazon.com/sdk-for-javascript/v2/developer-guide/node-reusing-connections.html)). To avoid the cost of establishing a new connection, you can reuse an existing connection by passing a new SQS instance with `keepAlive: true`.
- The queue is polled continuously for messages using [long polling](http://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-long-polling.html).
- Messages are deleted from the queue once the handler function has completed successfully.
- Throwing an error (or returning a rejected promise) from the handler function will cause the message to be left on the queue. An [SQS redrive policy](http://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/SQSDeadLetterQueue.html) can be used to move messages that cannot be processed to a dead letter queue.
- By default messages are processed one at a time – a new message won't be received until the first one has been processed. To process messages in parallel, use the `batchSize` option [detailed below](#options).
- By default, the default Node.js HTTP/HTTPS SQS agent creates a new TCP connection for every new request ([AWS SQS documentation](https://docs.aws.amazon.com/sdk-for-javascript/v2/developer-guide/node-reusing-connections.html)). To avoid the cost of establishing a new connection, you can reuse an existing connection by passing a new SQS instance with `keepAlive: true`.
```js

@@ -87,3 +88,2 @@ const { Consumer } = require('sqs-consumer');

```js

@@ -116,3 +116,3 @@ const { Consumer } = require('sqs-consumer');

app.on('timeout_error', (err) => {
console.error(err.message);
console.error(err.message);
});

@@ -131,18 +131,19 @@

* `queueUrl` - _String_ - The SQS queue URL
* `region` - _String_ - The AWS region (default `eu-west-1`)
* `handleMessage` - _Function_ - An `async` function (or function that returns a `Promise`) to be called whenever a message is received. Receives an SQS message object as it's first argument.
* `handleMessageBatch` - _Function_ - An `async` function (or function that returns a `Promise`) to be called whenever a batch of messages is received. Similar to `handleMessage` but will receive the list of messages, not each message individually. **If both are set, `handleMessageBatch` overrides `handleMessage`**.
* `handleMessageTimeout` - _Number_ - Time in ms to wait for `handleMessage` to process a message before timing out. Emits `timeout_error` on timeout. By default, if `handleMessage` times out, the unprocessed message returns to the end of the queue.
* `attributeNames` - _Array_ - List of queue attributes to retrieve (i.e. `['All', 'ApproximateFirstReceiveTimestamp', 'ApproximateReceiveCount']`).
* `messageAttributeNames` - _Array_ - List of message attributes to retrieve (i.e. `['name', 'address']`).
* `batchSize` - _Number_ - The number of messages to request from SQS when polling (default `1`). This cannot be higher than the AWS limit of 10.
* `visibilityTimeout` - _Number_ - The duration (in seconds) that the received messages are hidden from subsequent retrieve requests after being retrieved by a ReceiveMessage request.
* `heartbeatInterval` - _Number_ - The interval (in seconds) between requests to extend the message visibility timeout. On each heartbeat the visibility is extended by adding `visibilityTimeout` to the number of seconds since the start of the handler function. This value must less than `visibilityTimeout`.
* `terminateVisibilityTimeout` - _Boolean_ - If true, sets the message visibility timeout to 0 after a `processing_error` (defaults to `false`).
* `waitTimeSeconds` - _Number_ - The duration (in seconds) for which the call will wait for a message to arrive in the queue before returning.
* `authenticationErrorTimeout` - _Number_ - The duration (in milliseconds) to wait before retrying after an authentication error (defaults to `10000`).
* `pollingWaitTimeMs` - _Number_ - The duration (in milliseconds) to wait before repolling the queue (defaults to `0`).
* `sqs` - _Object_ - An optional [AWS SQS](http://docs.aws.amazon.com/AWSJavaScriptSDK/latest/AWS/SQS.html) object to use if you need to configure the client manually
* `shouldDeleteMessages` - _Boolean_ - Default to `true`, if you don't want the package to delete messages from sqs set this to `false`
- `queueUrl` - _String_ - The SQS queue URL
- `region` - _String_ - The AWS region (default `eu-west-1`)
- `handleMessage` - _Function_ - An `async` function (or function that returns a `Promise`) to be called whenever a message is received. Receives an SQS message object as it's first argument.
- `handleMessageBatch` - _Function_ - An `async` function (or function that returns a `Promise`) to be called whenever a batch of messages is received. Similar to `handleMessage` but will receive the list of messages, not each message individually. **If both are set, `handleMessageBatch` overrides `handleMessage`**.
- `handleMessageTimeout` - _Number_ - Time in ms to wait for `handleMessage` to process a message before timing out. Emits `timeout_error` on timeout. By default, if `handleMessage` times out, the unprocessed message returns to the end of the queue.
- `attributeNames` - _Array_ - List of queue attributes to retrieve (i.e. `['All', 'ApproximateFirstReceiveTimestamp', 'ApproximateReceiveCount']`).
- `messageAttributeNames` - _Array_ - List of message attributes to retrieve (i.e. `['name', 'address']`).
- `batchSize` - _Number_ - The number of messages to request from SQS when polling (default `1`). This cannot be higher than the AWS limit of 10.
- `visibilityTimeout` - _Number_ - The duration (in seconds) that the received messages are hidden from subsequent retrieve requests after being retrieved by a ReceiveMessage request.
- `heartbeatInterval` - _Number_ - The interval (in seconds) between requests to extend the message visibility timeout. On each heartbeat the visibility is extended by adding `visibilityTimeout` to the number of seconds since the start of the handler function. This value must less than `visibilityTimeout`.
- `terminateVisibilityTimeout` - _Boolean_ - If true, sets the message visibility timeout to 0 after a `processing_error` (defaults to `false`).
- `waitTimeSeconds` - _Number_ - The duration (in seconds) for which the call will wait for a message to arrive in the queue before returning (defaults to `20`).
- `authenticationErrorTimeout` - _Number_ - The duration (in milliseconds) to wait before retrying after an authentication error (defaults to `10000`).
- `pollingWaitTimeMs` - _Number_ - The duration (in milliseconds) to wait before repolling the queue (defaults to `0`).
- `sqs` - _Object_ - An optional [AWS SQS](http://docs.aws.amazon.com/AWSJavaScriptSDK/latest/AWS/SQS.html) object to use if you need to configure the client manually
- `shouldDeleteMessages` - _Boolean_ - Default to `true`, if you don't want the package to delete messages from sqs set this to `false`
### `consumer.start()`

@@ -156,3 +157,3 @@

### `consumer.isRunning`
### `consumer.isRunning`

@@ -165,12 +166,12 @@ Returns the current polling state of the consumer: `true` if it is actively polling, `false` if it is not.

|Event|Params|Description|
|-----|------|-----------|
|`error`|`err`, `[message]`|Fired when an error occurs interacting with the queue. If the error correlates to a message, that message is included in Params|
|`processing_error`|`err`, `message`|Fired when an error occurs processing the message.|
|`timeout_error`|`err`, `message`|Fired when `handleMessageTimeout` is supplied as an option and if `handleMessage` times out.|
|`message_received`|`message`|Fired when a message is received.|
|`message_processed`|`message`|Fired when a message is successfully processed and removed from the queue.|
|`response_processed`|None|Fired after one batch of items (up to `batchSize`) has been successfully processed.|
|`stopped`|None|Fired when the consumer finally stops its work.|
|`empty`|None|Fired when the queue is empty (All messages have been consumed).|
| Event | Params | Description |
| -------------------- | ------------------ | ------------------------------------------------------------------------------------------------------------------------------- |
| `error` | `err`, `[message]` | Fired when an error occurs interacting with the queue. If the error correlates to a message, that message is included in Params |
| `processing_error` | `err`, `message` | Fired when an error occurs processing the message. |
| `timeout_error` | `err`, `message` | Fired when `handleMessageTimeout` is supplied as an option and if `handleMessage` times out. |
| `message_received` | `message` | Fired when a message is received. |
| `message_processed` | `message` | Fired when a message is successfully processed and removed from the queue. |
| `response_processed` | None | Fired after one batch of items (up to `batchSize`) has been successfully processed. |
| `stopped` | None | Fired when the consumer finally stops its work. |
| `empty` | None | Fired when the queue is empty (All messages have been consumed). |

@@ -181,3 +182,4 @@ ### AWS IAM Permissions

### Contributing
See contributing [guidelines](https://github.com/bbc/sqs-consumer/blob/master/.github/CONTRIBUTING.md).
### Contributing
See contributing [guidelines](https://github.com/bbc/sqs-consumer/blob/main/.github/CONTRIBUTING.md).

140

src/consumer.ts

@@ -1,5 +0,4 @@

import { AWSError } from 'aws-sdk';
import * as SQS from 'aws-sdk/clients/sqs';
import { AWSError, SQS } from 'aws-sdk';
import { PromiseResult } from 'aws-sdk/lib/request';
import * as Debug from 'debug';
import Debug from 'debug';
import { EventEmitter } from 'events';

@@ -11,3 +10,6 @@ import { autoBind } from './bind';

type ReceieveMessageResponse = PromiseResult<SQS.Types.ReceiveMessageResult, AWSError>;
type ReceieveMessageResponse = PromiseResult<
SQS.Types.ReceiveMessageResult,
AWSError
>;
type ReceiveMessageRequest = SQS.Types.ReceiveMessageRequest;

@@ -41,3 +43,5 @@ export type SQSMessage = SQS.Types.Message;

if (!possibilities.find((p) => options[p])) {
throw new Error(`Missing SQS consumer option [ ${possibilities.join(' or ')} ].`);
throw new Error(
`Missing SQS consumer option [ ${possibilities.join(' or ')} ].`
);
}

@@ -50,3 +54,6 @@ });

if (options.heartbeatInterval && !(options.heartbeatInterval < options.visibilityTimeout)) {
if (
options.heartbeatInterval &&
!(options.heartbeatInterval < options.visibilityTimeout)
) {
throw new Error('heartbeatInterval must be less than visibilityTimeout.');

@@ -58,3 +65,7 @@ }

if (err instanceof SQSError) {
return (err.statusCode === 403 || err.code === 'CredentialsError' || err.code === 'UnknownEndpoint');
return (
err.statusCode === 403 ||
err.code === 'CredentialsError' ||
err.code === 'UnknownEndpoint'
);
}

@@ -101,10 +112,10 @@ return false;

interface Events {
'response_processed': [];
'empty': [];
'message_received': [SQSMessage];
'message_processed': [SQSMessage];
'error': [Error, void | SQSMessage | SQSMessage[]];
'timeout_error': [Error, SQSMessage];
'processing_error': [Error, SQSMessage];
'stopped': [];
response_processed: [];
empty: [];
message_received: [SQSMessage];
message_processed: [SQSMessage];
error: [Error, void | SQSMessage | SQSMessage[]];
timeout_error: [Error, SQSMessage];
processing_error: [Error, SQSMessage];
stopped: [];
}

@@ -142,12 +153,16 @@

this.visibilityTimeout = options.visibilityTimeout;
this.terminateVisibilityTimeout = options.terminateVisibilityTimeout || false;
this.terminateVisibilityTimeout =
options.terminateVisibilityTimeout || false;
this.heartbeatInterval = options.heartbeatInterval;
this.waitTimeSeconds = options.waitTimeSeconds ?? 20;
this.authenticationErrorTimeout = options.authenticationErrorTimeout ?? 10000;
this.authenticationErrorTimeout =
options.authenticationErrorTimeout ?? 10000;
this.pollingWaitTimeMs = options.pollingWaitTimeMs ?? 0;
this.shouldDeleteMessages = options.shouldDeleteMessages ?? true;
this.sqs = options.sqs || new SQS({
region: options.region || process.env.AWS_REGION || 'eu-west-1'
});
this.sqs =
options.sqs ||
new SQS({
region: options.region || process.env.AWS_REGION || 'eu-west-1'
});

@@ -161,7 +176,13 @@ autoBind(this);

on<T extends keyof Events>(event: T, listener: (...args: Events[T]) => void): this {
on<T extends keyof Events>(
event: T,
listener: (...args: Events[T]) => void
): this {
return super.on(event, listener);
}
once<T extends keyof Events>(event: T, listener: (...args: Events[T]) => void): this {
once<T extends keyof Events>(
event: T,
listener: (...args: Events[T]) => void
): this {
return super.once(event, listener);

@@ -191,3 +212,5 @@ }

private async handleSqsResponse(response: ReceieveMessageResponse): Promise<void> {
private async handleSqsResponse(
response: ReceieveMessageResponse
): Promise<void> {
debug('Received SQS response');

@@ -235,7 +258,7 @@ debug(response);

private async receiveMessage(params: ReceiveMessageRequest): Promise<ReceieveMessageResponse> {
private async receiveMessage(
params: ReceiveMessageRequest
): Promise<ReceieveMessageResponse> {
try {
return await this.sqs
.receiveMessage(params)
.promise();
return await this.sqs.receiveMessage(params).promise();
} catch (err) {

@@ -248,3 +271,5 @@ throw toSQSError(err, `SQS receive message failed: ${err.message}`);

if (!this.shouldDeleteMessages) {
debug('Skipping message delete since shouldDeleteMessages is set to false');
debug(
'Skipping message delete since shouldDeleteMessages is set to false'
);
return;

@@ -260,5 +285,3 @@ }

try {
await this.sqs
.deleteMessage(deleteParams)
.promise();
await this.sqs.deleteMessage(deleteParams).promise();
} catch (err) {

@@ -275,6 +298,3 @@ throw toSQSError(err, `SQS delete message failed: ${err.message}`);

[timeout, pending] = createTimeout(this.handleMessageTimeout);
await Promise.race([
this.handleMessage(message),
pending
]);
await Promise.race([this.handleMessage(message), pending]);
} else {

@@ -295,3 +315,6 @@ await this.handleMessage(message);

private async changeVisibilityTimeout(message: SQSMessage, timeout: number): Promise<PromiseResult<any, AWSError>> {
private async changeVisibilityTimeout(
message: SQSMessage,
timeout: number
): Promise<PromiseResult<any, AWSError>> {
try {

@@ -306,3 +329,7 @@ return await this.sqs

} catch (err) {
this.emit('error', toSQSError(err, `Error changing visibility timeout: ${err.message}`), message);
this.emit(
'error',
toSQSError(err, `Error changing visibility timeout: ${err.message}`),
message
);
}

@@ -347,5 +374,7 @@ }

return;
}).then(() => {
})
.then(() => {
setTimeout(this.poll, currentPollingTimeout);
}).catch((err) => {
})
.catch((err) => {
this.emit('error', err);

@@ -364,3 +393,6 @@ });

heartbeat = this.startHeartbeat(async () => {
return this.changeVisabilityTimeoutBatch(messages, this.visibilityTimeout);
return this.changeVisabilityTimeoutBatch(
messages,
this.visibilityTimeout
);
});

@@ -386,6 +418,11 @@ }

if (!this.shouldDeleteMessages) {
debug('Skipping message delete since shouldDeleteMessages is set to false');
debug(
'Skipping message delete since shouldDeleteMessages is set to false'
);
return;
}
debug('Deleting messages %s', messages.map((msg) => msg.MessageId).join(' ,'));
debug(
'Deleting messages %s',
messages.map((msg) => msg.MessageId).join(' ,')
);

@@ -401,5 +438,3 @@ const deleteParams = {

try {
await this.sqs
.deleteMessageBatch(deleteParams)
.promise();
await this.sqs.deleteMessageBatch(deleteParams).promise();
} catch (err) {

@@ -419,3 +454,6 @@ throw toSQSError(err, `SQS delete message failed: ${err.message}`);

private async changeVisabilityTimeoutBatch(messages: SQSMessage[], timeout: number): Promise<PromiseResult<any, AWSError>> {
private async changeVisabilityTimeoutBatch(
messages: SQSMessage[],
timeout: number
): Promise<PromiseResult<any, AWSError>> {
const params = {

@@ -430,7 +468,9 @@ QueueUrl: this.queueUrl,

try {
return await this.sqs
.changeMessageVisibilityBatch(params)
.promise();
return await this.sqs.changeMessageVisibilityBatch(params).promise();
} catch (err) {
this.emit('error', toSQSError(err, `Error changing visibility timeout: ${err.message}`), messages);
this.emit(
'error',
toSQSError(err, `Error changing visibility timeout: ${err.message}`),
messages
);
}

@@ -437,0 +477,0 @@ }

5

src/errors.ts

@@ -23,5 +23,2 @@ class SQSError extends Error {

export {
SQSError,
TimeoutError
};
export { SQSError, TimeoutError };

11

tsconfig.json

@@ -12,9 +12,4 @@ {

},
"include": [
"src/**/*"
],
"exclude": [
"node_modules",
"dist"
]
}
"include": ["src/**/*"],
"exclude": ["node_modules", "dist"]
}
.travis.yml
test/consumer.ts
.github/CODEOWNERS

Sorry, the diff of this file is not supported yet

SocketSocket SOC 2 Logo

Product

  • Package Alerts
  • Integrations
  • Docs
  • Pricing
  • FAQ
  • Roadmap
  • Changelog

About

Packages

npm

Go

Maven

PyPI

Rubygems

Stay in touch

Get open source security insights delivered straight into your inbox.

  • Terms
  • Privacy
  • Security

Made with ⚡️ by Socket Inc