Rest endpoint for message Prioritization

Refactored to support both Kafka and Rest endpoint prioritization.

Simplified number for Kafka processors.

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

2 files changed, 56 insertions, 21 deletions

diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConsumerTest.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConsumerTest.kt
index f9e23e826..ec0515c42 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConsumerTest.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/MessagePrioritizationConsumerTest.kt
@@ -27,12 +27,13 @@ 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.functions.message.prioritization.db.PrioritizationMessageRepository
-import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service.MessagePrioritizationStateService
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.kafka.MessagePrioritizationConsumer
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.utils.MessagePrioritizationSample
 import org.onap.ccsdk.cds.blueprintsprocessor.message.BluePrintMessageLibConfiguration
 import org.onap.ccsdk.cds.blueprintsprocessor.message.service.BluePrintMessageLibPropertyService
 import org.onap.ccsdk.cds.blueprintsprocessor.message.service.KafkaBasicAuthMessageProducerService
 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonString
+import org.onap.ccsdk.cds.controllerblueprints.core.logger
 import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.boot.test.autoconfigure.orm.jpa.DataJpaTest
@@ -43,6 +44,7 @@ import org.springframework.test.context.TestPropertySource
 import org.springframework.test.context.junit4.SpringRunner
 import kotlin.test.Test
 import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
 
 @RunWith(SpringRunner::class)
 @DataJpaTest
@@ -72,6 +74,8 @@ import kotlin.test.assertNotNull
 )
 open class MessagePrioritizationConsumerTest {
 
+    private val log = logger(MessagePrioritizationConsumerTest::class)
+
     @Autowired
     lateinit var applicationContext: ApplicationContext
 
@@ -82,6 +86,9 @@ open class MessagePrioritizationConsumerTest {
     lateinit var bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService
 
     @Autowired
+    lateinit var messagePrioritizationService: MessagePrioritizationService
+
+    @Autowired
     lateinit var messagePrioritizationConsumer: MessagePrioritizationConsumer
 
     @Before
@@ -107,6 +114,38 @@ open class MessagePrioritizationConsumerTest {
     }
 
     @Test
+    fun testMessagePrioritizationService() {
+        runBlocking {
+            assertTrue(
+                ::messagePrioritizationService.isInitialized,
+                "failed to initialize messagePrioritizationService"
+            )
+
+            log.info("****************  without Correlation **************")
+            /** Checking without correlation */
+            MessagePrioritizationSample.sampleMessages(MessageState.NEW.name, 2).forEach {
+                messagePrioritizationService.prioritize(it)
+            }
+            log.info("****************  Same Group , with Correlation **************")
+            /** checking same group with correlation */
+            MessagePrioritizationSample
+                .sampleMessageWithSameCorrelation("same-group", MessageState.NEW.name, 2)
+                .forEach {
+                    delay(10)
+                    messagePrioritizationService.prioritize(it)
+                }
+            log.info("****************  Different Type , with Correlation **************")
+            /** checking different type, with correlation */
+            MessagePrioritizationSample
+                .sampleMessageWithDifferentTypeSameCorrelation("group-typed", MessageState.NEW.name, 3)
+                .forEach {
+                    delay(10)
+                    messagePrioritizationService.prioritize(it)
+                }
+        }
+    }
+
+    @Test
     fun testStartConsuming() {
         runBlocking {
             val configuration = MessagePrioritizationSample.samplePrioritizationConfiguration()
@@ -118,7 +157,9 @@ open class MessagePrioritizationConsumerTest {
             val spyStreamingConsumerService = spyk(streamingConsumerService)
             coEvery { spyStreamingConsumerService.consume(any(), any()) } returns Unit
             coEvery { spyStreamingConsumerService.shutDown() } returns Unit
-            val messagePrioritizationConsumer = MessagePrioritizationConsumer(bluePrintMessageLibPropertyService)
+            val messagePrioritizationConsumer = MessagePrioritizationConsumer(
+                bluePrintMessageLibPropertyService
+            )
             val spyMessagePrioritizationConsumer = spyk(messagePrioritizationConsumer)
 
             // Test Topology
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/TestConfiguration.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/TestConfiguration.kt
index 3d3d0c6f5..0285079ad 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/TestConfiguration.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/TestConfiguration.kt
@@ -17,10 +17,9 @@
 package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization
 
 import org.onap.ccsdk.cds.blueprintsprocessor.db.PrimaryDBLibGenericService
-import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
-import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.topology.MessageAggregateProcessor
-import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.topology.MessageOutputProcessor
-import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.topology.MessagePrioritizeProcessor
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.kafka.DefaultMessagePrioritizeProcessor
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.kafka.MessagePrioritizationConsumer
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service.SampleMessagePrioritizationService
 import org.onap.ccsdk.cds.blueprintsprocessor.message.service.BluePrintMessageLibPropertyService
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration
 import org.springframework.context.annotation.Bean
@@ -65,22 +64,17 @@ open class SamplePrioritizationListeners(private val defaultMessagePrioritizatio
  */
 
 @Service
-open class SampleMessagePrioritizationConsumer(
+open class TestMessagePrioritizationService(messagePrioritizationStateService: MessagePrioritizationStateService) :
+    SampleMessagePrioritizationService(messagePrioritizationStateService)
+
+/** For Kafka Consumer  **/
+@Service
+open class TestMessagePrioritizationConsumer(
     bluePrintMessageLibPropertyService: BluePrintMessageLibPropertyService
 ) : MessagePrioritizationConsumer(bluePrintMessageLibPropertyService)
 
 @Service(MessagePrioritizationConstants.PROCESSOR_PRIORITIZE)
-open class SampleMessagePrioritizeProcessor : MessagePrioritizeProcessor() {
-    override fun getGroupCorrelationTypes(messagePrioritization: MessagePrioritization): List<String>? {
-        return when (messagePrioritization.group) {
-            "group-typed" -> arrayListOf("type-0", "type-1", "type-2")
-            else -> null
-        }
-    }
-}
-
-@Service(MessagePrioritizationConstants.PROCESSOR_AGGREGATE)
-open class SampleMessageAggregateProcessor() : MessageAggregateProcessor()
-
-@Service(MessagePrioritizationConstants.PROCESSOR_OUTPUT)
-open class SampleMessageOutputProcessor : MessageOutputProcessor()
+open class TestMessagePrioritizeProcessor(
+    messagePrioritizationStateService: MessagePrioritizationStateService,
+    messagePrioritizationService: MessagePrioritizationService
+) : DefaultMessagePrioritizeProcessor(messagePrioritizationStateService, messagePrioritizationService)