Merge pull request #175 from MoonActive/feature/new-polling-wait-time-ms

Feature/new polling wait time ms

README.md

@@ -106,6 +106,7 @@ Creates a new SQS consumer.
 * `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
 
 ### `consumer.start()`

src/consumer.ts

@@ -78,6 +78,7 @@ export interface ConsumerOptions {
   visibilityTimeout?: number;
   waitTimeSeconds?: number;
   authenticationErrorTimeout?: number;
+  pollingWaitTimeMs?: number;
   terminateVisibilityTimeout?: boolean;
   sqs?: SQS;
   region?: string;
@@ -98,6 +99,7 @@ export class Consumer extends EventEmitter {
   private visibilityTimeout: number;
   private waitTimeSeconds: number;
   private authenticationErrorTimeout: number;
+  private pollingWaitTimeMs: number;
   private terminateVisibilityTimeout: boolean;
   private sqs: SQS;
 
@@ -116,6 +118,7 @@ export class Consumer extends EventEmitter {
     this.terminateVisibilityTimeout = options.terminateVisibilityTimeout || false;
     this.waitTimeSeconds = options.waitTimeSeconds || 20;
     this.authenticationErrorTimeout = options.authenticationErrorTimeout || 10000;
+    this.pollingWaitTimeMs = options.pollingWaitTimeMs || 0;
 
     this.sqs = options.sqs || new SQS({
       region: options.region || process.env.AWS_REGION || 'eu-west-1'
@@ -272,18 +275,18 @@ export class Consumer extends EventEmitter {
       VisibilityTimeout: this.visibilityTimeout
     };
 
-    let pollingTimeout = 0;
+    let currentPollingTimeout = this.pollingWaitTimeMs;
     this.receiveMessage(receiveParams)
       .then(this.handleSqsResponse)
       .catch((err) => {
         this.emit('error', err);
         if (isConnectionError(err)) {
           debug('There was an authentication error. Pausing before retrying.');
-          pollingTimeout = this.authenticationErrorTimeout;
+          currentPollingTimeout = this.authenticationErrorTimeout;
         }
         return;
       }).then(() => {
-        setTimeout(this.poll, pollingTimeout);
+      setTimeout(this.poll, currentPollingTimeout);
       }).catch((err) => {
         this.emit('error', err);
       });

test/consumer.ts

@@ -7,6 +7,7 @@ import { Consumer } from '../src/index';
 const sandbox = sinon.createSandbox();
 
 const AUTHENTICATION_ERROR_TIMEOUT = 20;
+const POLLING_TIMEOUT = 100;
 
 function stubResolve(value?: any): any {
   return sandbox
@@ -314,6 +315,31 @@ describe('Consumer', () => {
       });
     });
 
+    it('waits before repolling when a polling timeout is set', async () => {
+      consumer = new Consumer({
+        queueUrl: 'some-queue-url',
+        region: 'some-region',
+        handleMessage,
+        sqs,
+        authenticationErrorTimeout: 20,
+        pollingWaitTimeMs: 100
+      });
+      return new Promise((resolve) => {
+        const timings = [];
+        const timeListener = sandbox.stub().callsFake(() => timings.push(new Date()));
+
+        timeListener.onThirdCall().callsFake(() => {
+          consumer.stop();
+          sandbox.assert.calledThrice(sqs.receiveMessage);
+          assert.isAtLeast(timings[1] - timings[0], POLLING_TIMEOUT);
+          resolve();
+        });
+
+        consumer.on('message_received', timeListener);
+        consumer.start();
+      });
+    });
+
     it('fires a message_received event when a message is received', async () => {
       consumer.start();
       const message = await pEvent(consumer, 'message_received');