fix(mojaloop/#3067): removed disconnected producers from listOfProducers map

.ncurc.js

@@ -1,6 +1,7 @@
 module.exports = {
   reject: [
     // TODO: Upgrading tape to v5+ causes tests to fail due to assert.end() being called multiple times. Will need to address this! Perhaps even move to Jest?
-    "tape"
+    "tape",
+    "node-rdkafka" // updating to the next major v3.0.0 should be done in a separate task
   ]
 }

audit-ci.jsonc

@@ -8,6 +8,7 @@
         "GHSA-8cf7-32gw-wr33",
         "GHSA-hjrf-2m68-5959",
         "GHSA-qwph-4952-7xr6",
-        "GHSA-w5p7-h5w8-2hfq"
+        "GHSA-w5p7-h5w8-2hfq",
+        "GHSA-f5x3-32g6-xq36" // https://github.com/advisories/GHSA-f5x3-32g6-xq36
     ]
 }

package.json

@@ -28,6 +28,7 @@
   "pre-commit": [
     "lint",
     "dep:check",
+    "audit:check",
     "test"
   ],
   "scripts": {
@@ -56,7 +57,7 @@
   },
   "devDependencies": {
     "audit-ci": "^6.6.1",
-    "npm-check-updates": "16.14.17",
+    "npm-check-updates": "16.14.18",
     "nyc": "15.1.0",
     "pre-commit": "1.2.2",
     "replace": "^1.2.2",

src/util/producer.js

@@ -108,6 +108,11 @@ const connectAll = async (configs) => {
   }
 }
 
+const disconnectAndRemoveProducer = async (topicName) => {
+  await getProducer(topicName).disconnect()
+  delete listOfProducers[topicName]
+}
+
 /**
  * @function Disconnect
  *
@@ -120,7 +125,7 @@ const connectAll = async (configs) => {
 const disconnect = async (topicName = null) => {
   if (topicName && typeof topicName === 'string') {
     try {
-      await getProducer(topicName).disconnect()
+      await disconnectAndRemoveProducer(topicName)
     } catch (err) {
       Logger.isErrorEnabled && Logger.error(err)
       throw ErrorHandler.Factory.reformatFSPIOPError(err)
@@ -131,7 +136,7 @@ const disconnect = async (topicName = null) => {
     let tpName
     for (tpName in listOfProducers) {
       try {
-        await getProducer(tpName).disconnect()
+        await disconnectAndRemoveProducer(tpName)
       } catch (e) {
         isError = true
         errorTopicList.push({ topic: tpName, error: e.toString() })
@@ -160,6 +165,7 @@ const getProducer = (topicName) => {
     return listOfProducers[topicName]
   } else {
     throw ErrorHandler.Factory.createInternalServerFSPIOPError(`No producer found for topic ${topicName}`)
+    // clarify, why we throw an error here and not just return null?
   }
 }
 

test/unit/util/producer.test.js

@@ -94,6 +94,15 @@ const topicConf = {
   opaqueKey: 0
 }
 
+const getProducerWithoutThrowError = (topicName) => {
+  try {
+    return Producer.getProducer(topicName)
+  } catch (err) {
+    Logger.warn(`getProducer error: ${err?.message}`)
+    return null
+  }
+}
+
 Test('Producer', producerTest => {
   let sandbox
   const config = {}
@@ -205,6 +214,7 @@ Test('Producer', producerTest => {
       sandbox.restore()
       t.end()
     })
+
     disconnectTest.test('disconnect from kafka', async test => {
       await Producer.produceMessage({}, { topicName: 'test' }, {})
       test.ok(Producer.disconnect('test'))
@@ -217,9 +227,11 @@ Test('Producer', producerTest => {
         test.ok(await Producer.produceMessage({}, { topicName }, {}))
         await Producer.disconnect(topicName)
         test.pass('Disconnect specific topic successfully')
+        const producer = getProducerWithoutThrowError(topicName)
+        test.equal(producer, null, 'No disconnected producer')
         test.end()
       } catch (e) {
-        test.fail('Error thrown')
+        test.fail(`Error thrown: ${e.message}`)
         test.end()
       }
     })
@@ -233,9 +245,11 @@ Test('Producer', producerTest => {
         test.ok(await Producer.produceMessage({}, { topicName }, {}))
         await Producer.disconnect()
         test.pass('Disconnected all topics successfully')
+        const producer = getProducerWithoutThrowError(topicName)
+        test.equal(producer, null, 'No disconnected producer')
         test.end()
       } catch (e) {
-        test.fail('Error thrown')
+        test.fail(`Error thrown: ${e.message}`)
         test.end()
       }
     })
@@ -530,5 +544,6 @@ Test('Producer', producerTest => {
 
     connectAllTest.end()
   })
+
   producerTest.end()
 })