fix: prevent executing bulk when instance is destroied (#81)

.github/workflows/ci.yml

@@ -10,6 +10,7 @@ jobs:
   test:
     runs-on: ubuntu-latest
     strategy:
+      fail-fast: false
       matrix:
         node-version: [14.x, 16.x, 18.x]
 

package.json

@@ -5,11 +5,11 @@
   "main": "persistence.js",
   "scripts": {
     "lint": "standard --verbose | snazzy",
-    "unit": "tape test.js | faucet",
+    "unit": "node test.js | faucet",
     "test": "npm run lint && npm run unit",
-    "coverage": "nyc --reporter=lcov tape test.js",
+    "coverage": "nyc --reporter=lcov node test.js",
     "test:ci": "npm run lint && npm run coverage",
-    "license-checker": "license-checker --production --onlyAllow='MIT;ISC;BSD-3-Clause;BSD-2-Clause;Apache-2.0;Apache*'",
+    "license-checker": "license-checker --production --onlyAllow='MIT;ISC;BSD-3-Clause;BSD-2-Clause;0BSD;Apache-2.0;Apache*'",
     "release": "read -p 'GITHUB_TOKEN: ' GITHUB_TOKEN && export GITHUB_TOKEN=$GITHUB_TOKEN && release-it --disable-metrics"
   },
   "release-it": {
@@ -52,24 +52,24 @@
   "homepage": "https://github.com/moscajs/aedes-persistence-mongodb#readme",
   "devDependencies": {
     "concat-stream": "^2.0.0",
-    "faucet": "0.0.1",
+    "faucet": "0.0.3",
     "license-checker": "^25.0.1",
     "mqemitter-mongodb": "^8.1.0",
     "nyc": "^15.1.0",
     "pre-commit": "^1.2.2",
-    "release-it": "^15.0.0",
+    "release-it": "^15.5.1",
     "snazzy": "^9.0.0",
     "standard": "^17.0.0",
-    "tape": "^4.15.1"
+    "tape": "^5.6.1"
   },
   "dependencies": {
     "aedes-cached-persistence": "^9.0.0",
     "escape-string-regexp": "^4.0.0",
     "fastparallel": "^2.4.1",
-    "mongodb": "^4.6.0",
+    "mongodb": "^4.13.0",
     "native-url": "^0.3.4",
     "pump": "^3.0.0",
-    "qlobber": "^7.0.0",
+    "qlobber": "^7.0.1",
     "through2": "^4.0.2"
   }
 }

persistence.js

@@ -620,7 +620,7 @@ class MongoPersistence extends CachedPersistence {
   }
 
   _executeBulk () {
-    if (!this.executing && this.packetsQueue.length > 0) {
+    if (!this.executing && !this._destroyed && this.packetsQueue.length > 0) {
       this.executing = true
       const bulk = this._cl.retained.initializeOrderedBulkOp()
       const onEnd = []

test.js

@@ -768,4 +768,39 @@ function runTest (client, db) {
       })
     })
   })
+
+  test('prevent executing bulk when instance is destroied', function (t) {
+    clean(function (err) {
+      t.error(err)
+
+      const emitter = mqemitterMongo(dbopts)
+
+      emitter.status.on('stream', function () {
+        t.pass('mqemitter ready')
+        const instance = persistence(dbopts)
+        instance.broker = toBroker('2', emitter)
+
+        instance.on('ready', function () {
+          t.pass('instance ready')
+
+          const packet = {
+            cmd: 'publish',
+            id: instance.broker.id,
+            topic: 'hello/world',
+            payload: Buffer.from('muahah'),
+            qos: 0,
+            retain: true
+          }
+
+          instance.packetsQueue.push({ packet, cb: () => { } })
+
+          instance.destroy(function () {
+            t.pass('Instance dies')
+            instance._executeBulk() // should not throw
+            emitter.close(t.end.bind(t))
+          })
+        })
+      })
+    })
+  })
 }