fix: cannot build ml-api-adapter with current version of central-serv…

…ices-stream (#141)

.circleci/config.yml

@@ -77,7 +77,7 @@ defaults_configure_nvm: &defaults_configure_nvm
 
     if [ -f "$NVM_DIR" ]; then
       echo "==> $NVM_DIR exists. Skipping steps 2-4!"
-    else 
+    else
       echo "==> $NVM_DIR does not exists. Executing steps 2-4!"
 
       echo "2. Installing NVM"
@@ -448,6 +448,57 @@ jobs:
       - slack/notify:
           event: fail
           template: SLACK_TEMP_RELEASE_FAILURE
+  publish-prerelease:
+    executor: default-machine
+    steps:
+      - run:
+          name: Install git
+          command: |
+            sudo apt-get update && sudo apt-get install -y git
+      - gh/install
+      - checkout
+      - restore_cache:
+          key: dependency-cache-{{ .Environment.CIRCLE_SHA1 }}
+      - run:
+          <<: *defaults_configure_git
+      - run:
+          name: Setup for pre-release
+          command: |
+            if [[ $CIRCLE_BRANCH =~ ^(major|minor|patch)/(.*)$ ]]; then
+              echo "export RELEASE_TAG=${BASH_REMATCH[1]}-${BASH_REMATCH[2]}" >> $BASH_ENV
+              echo "RELEASE_TAG=${BASH_REMATCH[1]}-${BASH_REMATCH[2]}"
+              echo "Pre-releasing ${BASH_REMATCH[1]} version ${BASH_REMATCH[2]}"
+              npx standard-version --prerelease "${BASH_REMATCH[2]}" --release-as "${BASH_REMATCH[1]}" --message "chore(release): [ci skip] %s" --skip.changelog --no-verify
+
+              PACKAGE_VERSION=$(cat package-lock.json | jq -r .version)
+              echo "export PACKAGE_VERSION=${PACKAGE_VERSION}" >> $BASH_ENV
+              echo "PACKAGE_VERSION=${PACKAGE_VERSION}"
+            else
+              echo "unsupported branch $CIRCLE_BRANCH for pre-release"
+              exit 1
+            fi
+      - run:
+          name: Push the pre-release
+          command: git push --follow-tags origin ${CIRCLE_BRANCH}
+      - run:
+          name: Setup Slack config
+          command: |
+            echo "export SLACK_PROJECT_NAME=${CIRCLE_PROJECT_REPONAME}" >> $BASH_ENV
+            echo "export SLACK_RELEASE_TYPE='NPM pre-release'" >> $BASH_ENV
+            echo "export SLACK_RELEASE_TAG=v${CIRCLE_TAG:1}" >> $BASH_ENV
+            echo "export SLACK_RELEASE_URL=https://www.npmjs.com/package/@mojaloop/${CIRCLE_PROJECT_REPONAME}/v/${CIRCLE_TAG:1}" >> $BASH_ENV
+            echo "export SLACK_BUILD_ID=${CIRCLE_BUILD_NUM}" >> $BASH_ENV
+            echo "export SLACK_CI_URL=${CIRCLE_BUILD_URL}" >> $BASH_ENV
+      - run:
+          <<: *defaults_npm_auth
+      - run:
+          <<: *defaults_npm_publish_release
+      - slack/notify:
+          event: pass
+          template: SLACK_TEMP_RELEASE_SUCCESS
+      - slack/notify:
+          event: fail
+          template: SLACK_TEMP_RELEASE_FAILURE
 
 workflows:
   build_and_test:
@@ -585,3 +636,16 @@ workflows:
             branches:
               ignore:
                 - /.*/
+      - publish-prerelease:
+          context: org-global
+          requires:
+            - pr-tools/pr-title-check
+            - test-lint
+            - test-unit
+            - test-coverage
+            - vulnerability-check
+            - audit-licenses
+          filters:
+            branches:
+              only:
+                - /(major|minor|patch)/.*/

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@mojaloop/central-services-stream",
-  "version": "11.2.5",
+  "version": "11.2.6-140.14",
   "description": "Streaming library code for central services.",
   "license": "Apache-2.0",
   "bugs": {
@@ -57,12 +57,12 @@
   },
   "devDependencies": {
     "audit-ci": "^6.6.1",
-    "npm-check-updates": "16.14.18",
+    "npm-check-updates": "16.14.20",
     "nyc": "15.1.0",
     "pre-commit": "1.2.2",
     "replace": "^1.2.2",
     "rewire": "7.0.0",
-    "sinon": "17.0.1",
+    "sinon": "18.0.0",
     "standard": "17.1.0",
     "standard-version": "^9.5.0",
     "tap-spec": "^5.0.0",

src/util/producer.js

@@ -189,11 +189,48 @@ const isConnected = async (topicName = undefined) => {
   return producer.isConnected()
 }
 
+/**
+ * @function getMetadataPromise
+ *
+ * @param {object} producer - the producer class
+ * @param {string} topic - the topic name of the producer to check
+ *
+ * @description Use this to determine whether or not we are connected to the broker. Internally, it calls `getMetadata` to determine
+ * if the broker client is connected.
+ *
+ * @returns object - resolve metadata object
+ * @throws {Error} - if Producer can't be found or the producer is not connected
+ */
+const getMetadataPromise = async (producer, topic) => {
+  return new Promise((resolve, reject) => {
+    const cb = async (err, metadata) => {
+      if (err) {
+        return reject(new Error(`Error connecting to producer: ${err.message}`))
+      }
+      return resolve(metadata)
+    }
+    producer.getMetadata({ topic, timeout: 6000 }, cb)
+  })
+}
+
+const allConnected = async () => {
+  for (const [key, value] of Object.entries(listOfProducers)) {
+    const metadata = await getMetadataPromise(value._producer, key)
+    const foundTopics = metadata.topics.map(topic => topic.name)
+    if (foundTopics.indexOf(key) === -1) {
+      Logger.isDebugEnabled && Logger.debug(`Connected to producer, but ${key} not found.`)
+      throw ErrorHandler.Factory.createInternalServerFSPIOPError(`Connected to producer, but ${key} not found.`)
+    }
+  }
+  return stateList.OK
+}
+
 module.exports = {
   getProducer,
   produceMessage,
   disconnect,
   isConnected,
+  allConnected,
   stateList,
   connectAll
 }

test/unit/util/producer.test.js

@@ -432,6 +432,39 @@ Test('Producer', producerTest => {
       test.end()
     })
 
+    isConnectedTest.test('pass if the topicName is not supplied', async test => {
+      // Arrange
+      const ProducerProxy = rewire(`${src}/util/producer`)
+      const metadata = {
+        orig_broker_id: 0,
+        orig_broker_name: 'kafka:9092/0',
+        topics: [
+          { name: 'admin', partitions: [] }
+        ],
+        brokers: [{ id: 0, host: 'kafka', port: 9092 }]
+      }
+      ProducerProxy.__set__('listOfProducers', {
+        admin: {
+          _producer: {
+            // Callback with error
+            getMetadata: (options, cb) => cb(null, metadata)
+          }
+        }
+      })
+
+      // Act
+      let result
+      try {
+        result = await ProducerProxy.allConnected()
+      } catch (err) {
+        test.fail(err.message)
+      }
+
+      // Assert
+      test.equal(result, Producer.stateList.OK, 'isConnected should return true')
+      test.end()
+    })
+
     isConnectedTest.end()
   })