Merge pull request #5 from blumamir/no-sleep

fix: no sleep between batch receive calls
aspecto-io · Jul 23, 2020 · 4dec99c · 4dec99c
2 parents 4eb3aa1 + 57bd33f
commit 4dec99c
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 23 deletions.
diff --git a/README.md b/README.md
@@ -43,7 +43,7 @@ const snsProducer = SnsProducer.create({
     region: 'us-east-1',
     // to enable sending large payloads (>256KiB) though S3
     largePayloadThoughS3: true,
-    s3Bucket: '...',
+    s3EndpointUrl: '...',
 });
 
 await snsProducer.sendJSON({

diff --git a/src/sqs-consumer.ts b/src/sqs-consumer.ts
@@ -46,7 +46,6 @@ export class SqsConsumer {
     private waitTimeSeconds: number;
     private started = false;
     private events = new EventEmitter();
-    private pollingInterval = 1000;
     private connErrorTimeout = 10000;
     private handleMessage?: (message: SqsMessage) => Promise<void>;
     private parsePayload?: (payload: any) => any;
@@ -101,33 +100,26 @@ export class SqsConsumer {
         this.events.on(event, handler);
     }
 
-    private poll(): void {
-        if (!this.started) return;
-        let currentPollingInterval = this.pollingInterval;
-
-        this.receiveMessages({
-            QueueUrl: this.queueUrl,
-            MaxNumberOfMessages: this.batchSize,
-            WaitTimeSeconds: this.waitTimeSeconds,
-        })
-            .then((response) => {
+    private async poll() {
+        while(this.started) {
+            try {
+                const response = await this.receiveMessages({
+                    QueueUrl: this.queueUrl,
+                    MaxNumberOfMessages: this.batchSize,
+                    WaitTimeSeconds: this.waitTimeSeconds,
+                });
                 if (!this.started) return;
-                return this.handleSqsResponse(response);
-            })
-            .catch((err) => {
+                await this.handleSqsResponse(response);    
+            }
+            catch (err) {
                 if (this.isConnError(err)) {
                     this.events.emit(SqsConsumerEvents.connectionError, err);
-                    currentPollingInterval = this.connErrorTimeout;
+                    await new Promise(resolve => setTimeout(resolve, this.connErrorTimeout));
                 } else {
                     this.events.emit(SqsConsumerEvents.error, err);
                 }
-            })
-            .then(() => {
-                setTimeout(this.poll.bind(this), currentPollingInterval);
-            })
-            .catch((err) => {
-                this.events.emit('error', err);
-            });
+            }
+        }
     }
 
     private isConnError(err: AWSError): Boolean {