Prioritization Optional NATS consumer support

Add prioritization NATS consumer service and configuration data beans.

Optimizing message prioritization service interface.

Added Integration testing for NATS simulation.

Updated sample docker compose for NATS support

Issue-ID: CCSDK-1917
Signed-off-by: Brinda Santh <bs2796@att.com>
Change-Id: Icd21e5e2ab7b64d6e6e4b0610599ca947555ee15

7 files changed, 114 insertions, 8 deletions

diff --git a/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/BluePrintNatsLibConfiguration.kt b/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/BluePrintNatsLibConfiguration.kt
index 147d360ba..8d5d846d7 100644
--- a/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/BluePrintNatsLibConfiguration.kt
+++ b/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/BluePrintNatsLibConfiguration.kt
@@ -41,6 +41,7 @@ class NatsLibConstants {
         const val SERVICE_BLUEPRINT_NATS_LIB_PROPERTY = "blueprint-nats-lib-property-service"
         const val DEFULT_NATS_SELECTOR = "cds-controller"
         const val PROPERTY_NATS_PREFIX = "blueprintsprocessor.nats."
+        const val PROPERTY_NATS_CLUSTER_ID = "NATS_CLUSTER_ID"
         const val TYPE_TOKEN_AUTH = "token-auth"
         const val TYPE_TLS_AUTH = "tls-auth"
     }
diff --git a/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/BluePrintNatsLibData.kt b/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/BluePrintNatsLibData.kt
index 9767ac29d..74897f322 100644
--- a/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/BluePrintNatsLibData.kt
+++ b/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/BluePrintNatsLibData.kt
@@ -16,11 +16,12 @@
 
 package org.onap.ccsdk.cds.blueprintsprocessor.nats
 
+import org.onap.ccsdk.cds.blueprintsprocessor.nats.utils.NatsClusterUtils
 import org.onap.ccsdk.cds.controllerblueprints.core.utils.ClusterUtils
 
 open class NatsConnectionProperties {
     lateinit var type: String
-    var clusterId: String = ClusterUtils.clusterId()
+    var clusterId: String = NatsClusterUtils.clusterId()
     var clientId: String = ClusterUtils.clusterNodeId()
     lateinit var host: String
     /** Rest endpoint selector to access Monitoring API */
diff --git a/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/service/BluePrintNatsLibPropertyService.kt b/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/service/BluePrintNatsLibPropertyService.kt
index faf171528..18d0639f1 100644
--- a/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/service/BluePrintNatsLibPropertyService.kt
+++ b/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/service/BluePrintNatsLibPropertyService.kt
@@ -50,15 +50,13 @@ open class BluePrintNatsLibPropertyService(private var bluePrintPropertiesServic
                 JacksonUtils.readValue(jsonNode, TLSAuthNatsConnectionProperties::class.java)!!
             }
             else -> {
-                throw BluePrintProcessorException("Nats type($type) not supported")
+                throw BluePrintProcessorException("NATS type($type) not supported")
             }
         }
     }
 
     fun natsConnectionProperties(prefix: String): NatsConnectionProperties {
-        val type = bluePrintPropertiesService.propertyBeanType(
-            "$prefix.type", String::class.java
-        )
+        val type = bluePrintPropertiesService.propertyBeanType("$prefix.type", String::class.java)
         return when (type) {
             NatsLibConstants.TYPE_TOKEN_AUTH -> {
                 tokenAuthNatsConnectionProperties(prefix)
@@ -67,7 +65,7 @@ open class BluePrintNatsLibPropertyService(private var bluePrintPropertiesServic
                 tlsAuthNatsConnectionProperties(prefix)
             }
             else -> {
-                throw BluePrintProcessorException("Grpc type($type) not supported")
+                throw BluePrintProcessorException("NATS type($type) not supported")
             }
         }
     }
@@ -90,7 +88,7 @@ open class BluePrintNatsLibPropertyService(private var bluePrintPropertiesServic
                 TLSAuthNatsService(natsConnectionProperties)
             }
             else -> {
-                throw BluePrintProcessorException("couldn't get nats service for properties $natsConnectionProperties")
+                throw BluePrintProcessorException("couldn't get NATS service for properties $natsConnectionProperties")
             }
         }
     }
diff --git a/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/service/TokenAuthNatsService.kt b/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/service/TokenAuthNatsService.kt
index 60b7934ba..43a43bc03 100644
--- a/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/service/TokenAuthNatsService.kt
+++ b/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/service/TokenAuthNatsService.kt
@@ -21,15 +21,23 @@ import io.nats.client.Options
 import io.nats.streaming.NatsStreaming
 import io.nats.streaming.StreamingConnection
 import org.onap.ccsdk.cds.blueprintsprocessor.nats.TokenAuthNatsConnectionProperties
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
 import org.onap.ccsdk.cds.controllerblueprints.core.splitCommaAsList
 
 open class TokenAuthNatsService(private val natsConnectionProperties: TokenAuthNatsConnectionProperties) :
     BluePrintNatsService {
 
+    private val log = logger(TokenAuthNatsService::class)
+
     lateinit var streamingConnection: StreamingConnection
 
     override suspend fun connection(): StreamingConnection {
         if (!::streamingConnection.isInitialized) {
+            log.info(
+                "NATS connection requesting for cluster(${natsConnectionProperties.clusterId}) with" +
+                    "clientId($natsConnectionProperties.clientId)"
+            )
+
             val serverList = natsConnectionProperties.host.splitCommaAsList()
 
             val options = Options.Builder()
diff --git a/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/utils/NatsClusterUtils.kt b/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/utils/NatsClusterUtils.kt
new file mode 100644
index 000000000..a7726a14b
--- /dev/null
+++ b/ms/blueprintsprocessor/modules/commons/nats-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/utils/NatsClusterUtils.kt
@@ -0,0 +1,45 @@
+/*
+ * Copyright © 2018-2019 AT&T Intellectual Property.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onap.ccsdk.cds.blueprintsprocessor.nats.utils
+
+import org.onap.ccsdk.cds.blueprintsprocessor.nats.NatsLibConstants
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintConstants
+import org.onap.ccsdk.cds.controllerblueprints.core.utils.ClusterUtils
+
+object NatsClusterUtils {
+
+    fun clusterId(): String {
+        return System.getenv(NatsLibConstants.PROPERTY_NATS_CLUSTER_ID)
+            ?: ClusterUtils.clusterId()
+    }
+
+    fun applicationSubject(appName: String, subject: String): String {
+        return "$appName.$subject"
+    }
+
+    fun currentApplicationSubject(subject: String): String {
+        return "${BluePrintConstants.APP_NAME}.$subject"
+    }
+
+    fun currentNodeDurable(subject: String): String {
+        return "${ClusterUtils.clusterNodeId()}-$subject"
+    }
+
+    fun applicationLoadBalanceGroup(): String {
+        return "${BluePrintConstants.APP_NAME}"
+    }
+}
diff --git a/ms/blueprintsprocessor/modules/commons/nats-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/service/BluePrintNatsLibPropertyServiceTest.kt b/ms/blueprintsprocessor/modules/commons/nats-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/service/BluePrintNatsLibPropertyServiceTest.kt
new file mode 100644
index 000000000..ba993d96a
--- /dev/null
+++ b/ms/blueprintsprocessor/modules/commons/nats-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/service/BluePrintNatsLibPropertyServiceTest.kt
@@ -0,0 +1,53 @@
+/*
+ * Copyright © 2018-2019 AT&T Intellectual Property.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onap.ccsdk.cds.blueprintsprocessor.nats.service
+
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertiesService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.BluePrintPropertyConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.nats.BluePrintNatsLibConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.nats.NatsLibConstants
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.test.context.ContextConfiguration
+import org.springframework.test.context.TestPropertySource
+import org.springframework.test.context.junit4.SpringRunner
+import kotlin.test.assertTrue
+
+@RunWith(SpringRunner::class)
+@ContextConfiguration(
+    classes = [BluePrintNatsLibConfiguration::class,
+        BluePrintPropertyConfiguration::class, BluePrintPropertiesService::class]
+)
+@TestPropertySource(
+    properties =
+    ["blueprintsprocessor.nats.cds-controller.type=token-auth",
+        "blueprintsprocessor.nats.cds-controller.host=nats://localhost:4222",
+        "blueprintsprocessor.nats.cds-controller.token=tokenAuth"
+    ]
+)
+class BluePrintNatsLibPropertyServiceTest {
+
+    @Autowired
+    lateinit var bluePrintNatsLibPropertyService: BluePrintNatsLibPropertyService
+
+    @Test
+    fun testNatsProperties() {
+        assertTrue(::bluePrintNatsLibPropertyService.isInitialized)
+        bluePrintNatsLibPropertyService.bluePrintNatsService(NatsLibConstants.DEFULT_NATS_SELECTOR)
+    }
+}
diff --git a/ms/blueprintsprocessor/modules/commons/nats-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/service/BluePrintNatsServiceTest.kt b/ms/blueprintsprocessor/modules/commons/nats-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/service/BluePrintNatsServiceTest.kt
index 549be6481..721828ac9 100644
--- a/ms/blueprintsprocessor/modules/commons/nats-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/service/BluePrintNatsServiceTest.kt
+++ b/ms/blueprintsprocessor/modules/commons/nats-lib/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/nats/service/BluePrintNatsServiceTest.kt
@@ -84,7 +84,7 @@ class BluePrintNatsServiceTest {
      * Start the Server with : nats-streaming-server -cid cds-cluster --auth tokenAuth -m 8222 -V
      */
     // @Test
-    fun localTntegrationTest() {
+    fun localIntegrationTest() {
         runBlocking {
 
             val connectionProperties = TokenAuthNatsConnectionProperties().apply {