fix: Stopping a queue should delete all pending messages immediately #…

…146 (#147)
PruvoNet · Aug 6, 2023 · 16b93ab · 16b93ab
1 parent 0e511a4
commit 16b93ab
Show file tree

Hide file tree

Showing 4 changed files with 916 additions and 940 deletions.
diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
   "name": "squiss-ts",
-  "version": "5.2.1",
+  "version": "5.2.2",
   "description": "High-volume SQS poller",
   "main": "dist/index.js",
   "types": "dist/index.d.ts",
@@ -48,9 +48,9 @@
   },
   "homepage": "https://github.com/PruvoNet/squiss-ts#readme",
   "dependencies": {
-    "@aws-sdk/client-s3": "^3.362.0",
-    "@aws-sdk/client-sqs": "^3.360.0",
-    "@aws-sdk/types": "^3.369.0",
+    "@aws-sdk/client-s3": "^3.385.0",
+    "@aws-sdk/client-sqs": "^3.385.0",
+    "@aws-sdk/types": "^3.378.0",
     "linked-list": "2.1.0",
     "ts-type-guards": "^0.7.0",
     "uuid": "^9.0.0"
@@ -60,30 +60,31 @@
     "@types/chai": "^4.3.5",
     "@types/chai-as-promised": "^7.1.5",
     "@types/mocha": "^10.0.1",
-    "@types/node": "^14.18.48",
+    "@types/node": "^16.18.39",
     "@types/proxyquire": "^1.3.28",
-    "@types/uuid": "^9.0.1",
+    "@types/uuid": "^9.0.2",
     "chai": "^4.3.7",
     "chai-as-promised": "^7.1.1",
     "delay": "5.0.0",
     "dirty-chai": "^2.0.1",
     "mocha": "^10.2.0",
     "nyc": "^15.1.0",
     "proxyquire": "^2.1.3",
-    "sinon": "^15.1.0",
+    "sinon": "^15.2.0",
     "sinon-chai": "^3.7.0",
     "source-map-support": "^0.5.21",
     "ts-node": "^10.9.1",
     "tslint": "^6.1.3",
-    "typescript": "^5.1.3"
+    "typescript": "^5.1.6"
   },
   "nyc": {
     "extends": "@istanbuljs/nyc-config-typescript",
     "check-coverage": true,
     "all": true,
     "exclude": [
       "src/test/**",
-      "e2e**"
+      "e2e**",
+      "build**"
     ],
     "reporter": [
       "html",

diff --git a/src/Squiss.ts b/src/Squiss.ts
@@ -110,9 +110,9 @@ export class Squiss extends (EventEmitter as new() => SquissEmitter) {
         });
     }
 
-    public deleteMessage(msg: Message): Promise<void> {
+    public async deleteMessage(msg: Message): Promise<void> {
         if (!msg.raw) {
-            return Promise.reject(new Error('Squiss.deleteMessage requires a Message object'));
+            throw new Error('Squiss.deleteMessage requires a Message object');
         }
         const promise = new Promise<void>((resolve, reject) => {
             this._delQueue.set(msg.raw.MessageId!,
@@ -126,11 +126,11 @@ export class Squiss extends (EventEmitter as new() => SquissEmitter) {
                 clearTimeout(this._delTimer);
                 this._delTimer = undefined;
             }
-            this._deleteXMessages(this._opts.deleteBatchSize);
+            await this._deleteXMessages(this._opts.deleteBatchSize);
         } else if (!this._delTimer) {
-            this._delTimer = setTimeout(() => {
+            this._delTimer = setTimeout(async () => {
                 this._delTimer = undefined;
-                this._deleteXMessages();
+                await this._deleteXMessages();
             }, this._opts.deleteWaitMs);
         }
         return promise;
@@ -312,18 +312,19 @@ export class Squiss extends (EventEmitter as new() => SquissEmitter) {
         return this._startPoller();
     }
 
-    public stop(soft?: boolean, timeout?: number): Promise<boolean> {
+    public async stop(soft?: boolean, timeout?: number): Promise<boolean> {
         if (!soft && this._activeReq) {
             this._activeReq.abort();
         }
         this._running = this._paused = false;
         if (!this._inFlight) {
-            return Promise.resolve(true);
+            await this._drainDeleteQueue();
+            return true;
         }
         let resolved = false;
         let timer: any;
-        return new Promise((resolve) => {
-            this.on('drained', () => {
+        const result = await new Promise<boolean>(async (resolve) => {
+            this.on('drained',() => {
                 if (!resolved) {
                     resolved = true;
                     if (timer) {
@@ -338,6 +339,8 @@ export class Squiss extends (EventEmitter as new() => SquissEmitter) {
                 resolve(false);
             }, timeout) : undefined;
         });
+        await this._drainDeleteQueue();
+        return result;
     }
 
     public getS3(): S3 {
@@ -347,6 +350,14 @@ export class Squiss extends (EventEmitter as new() => SquissEmitter) {
         return this._s3;
     }
 
+    private async _drainDeleteQueue(): Promise<void> {
+        if (this._delTimer) {
+            clearTimeout(this._delTimer);
+            this._delTimer = undefined;
+        }
+        await this._deleteXMessages();
+    }
+
     private _initS3() {
         if (this._opts.S3) {
             if (typeof this._opts.S3 === 'function') {
@@ -500,15 +511,15 @@ export class Squiss extends (EventEmitter as new() => SquissEmitter) {
             })
     }
 
-    private _deleteXMessages(x?: number) {
+    private async _deleteXMessages(x?: number): Promise<void> {
         const delQueue = this._delQueue;
         const iterator = delQueue.entries();
         const delBatch = Array.from({length: x || delQueue.size}, function(this: typeof iterator) {
             const element = this.next().value;
             delQueue.delete(element[0]);
             return element[1];
         }, iterator);
-        this._deleteMessages(delBatch);
+        await this._deleteMessages(delBatch);
     }
 
     private _isLargeMessage(message: ISendMessageRequest, minSize?: number): Promise<boolean> {

diff --git a/src/test/src/index.spec.ts b/src/test/src/index.spec.ts
@@ -387,22 +387,31 @@ describe('index', () => {
     });
     it('should resolve when queue already drained', () => {
       const spy = sinon.spy();
+      const deleteSpy = sinon.spy();
       inst = new SquissPatched({queueUrl: 'foo'} as ISquissOptions);
-      inst!.sqs = new SQSStub(0, 1000) as any as SQS;
+      inst!.sqs = new SQSStub(1, 1000) as any as SQS;
       inst!.on('aborted', spy);
+      inst!.on('deleted', deleteSpy);
+      inst!.on('message', (msg: Message) => {
+        msg.del();
+      });
       inst!.start();
       return wait().then(() => {
         spy.should.not.be.called();
+        deleteSpy.should.not.be.called();
         return inst!.stop(false, 1000);
       }).then((result: boolean) => {
+        deleteSpy.should.be.called();
         result.should.eq(true);
       });
     });
     it('should resolve when queue drained before timeout', () => {
       const spy = sinon.spy();
+      const deleteSpy = sinon.spy();
       inst = new SquissPatched({queueUrl: 'foo'} as ISquissOptions);
       inst!.sqs = new SQSStub(1, 1000) as any as SQS;
       inst!.on('aborted', spy);
+      inst!.on('deleted', deleteSpy);
       inst!.on('message', (msg: Message) => {
         setTimeout(() => {
           msg.del();
@@ -411,8 +420,10 @@ describe('index', () => {
       inst!.start();
       return wait().then(() => {
         spy.should.not.be.called();
+        deleteSpy.should.not.be.called();
         return inst!.stop(false, 10000);
       }).then((result: boolean) => {
+        deleteSpy.should.be.called();
         result.should.eq(true);
       });
     });