Prioritization expiry and clean scheduler service

Add prioritization expiry and clean scheduler service implementation.

Optimizing message passing between processors.

Added message sorting utils.

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

2 files changed, 39 insertions, 5 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 ec0515c42..190f4e891 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
@@ -28,6 +28,7 @@ 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.kafka.MessagePrioritizationConsumer
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.service.MessagePrioritizationSchedulerService
 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
@@ -89,6 +90,9 @@ open class MessagePrioritizationConsumerTest {
     lateinit var messagePrioritizationService: MessagePrioritizationService
 
     @Autowired
+    lateinit var messagePrioritizationSchedulerService: MessagePrioritizationSchedulerService
+
+    @Autowired
     lateinit var messagePrioritizationConsumer: MessagePrioritizationConsumer
 
     @Before
@@ -151,7 +155,7 @@ open class MessagePrioritizationConsumerTest {
             val configuration = MessagePrioritizationSample.samplePrioritizationConfiguration()
 
             val streamingConsumerService = bluePrintMessageLibPropertyService
-                .blueprintMessageConsumerService(configuration.inputTopicSelector)
+                .blueprintMessageConsumerService(configuration.kafkaConfiguration!!.inputTopicSelector)
             assertNotNull(streamingConsumerService, "failed to get blueprintMessageConsumerService")
 
             val spyStreamingConsumerService = spyk(streamingConsumerService)
@@ -176,6 +180,25 @@ open class MessagePrioritizationConsumerTest {
         }
     }
 
+    @Test
+    fun testSchedulerService() {
+        runBlocking {
+            val configuration = MessagePrioritizationSample.sampleSchedulerPrioritizationConfiguration()
+            assertTrue(
+                ::messagePrioritizationSchedulerService.isInitialized,
+                "failed to initialize messagePrioritizationSchedulerService"
+            )
+            launch {
+                messagePrioritizationSchedulerService.startScheduling(configuration)
+            }
+            launch {
+                /** To debug increase the delay time */
+                delay(20)
+                messagePrioritizationSchedulerService.shutdownScheduling(configuration)
+            }
+        }
+    }
+
     /** Integration Kafka Testing, Enable and use this test case only for local desktop testing with real kafka broker */
     // @Test
     fun testMessagePrioritizationConsumer() {
diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageCorrelationUtilsTest.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageCorrelationUtilsTest.kt
index 3876cbba5..73d3738e5 100644
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageCorrelationUtilsTest.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/test/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageCorrelationUtilsTest.kt
@@ -19,6 +19,8 @@ package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.
 import org.junit.Test
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.MessageState
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.orderByHighestPriority
+import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
 
 class MessageCorrelationUtilsTest {
@@ -59,10 +61,11 @@ class MessageCorrelationUtilsTest {
 
         /* Assumption is Same group with different types and one missing expected types,
         In this case type-3 message is missing */
-        val differentTypesWithSameCorrelationMessagesResWithMissingType = MessageCorrelationUtils.correlatedMessagesWithTypes(
-            differentTypesWithSameCorrelationMessages,
-            arrayListOf("type-0", "type-1", "type-2", "type-3")
-        )
+        val differentTypesWithSameCorrelationMessagesResWithMissingType =
+            MessageCorrelationUtils.correlatedMessagesWithTypes(
+                differentTypesWithSameCorrelationMessages,
+                arrayListOf("type-0", "type-1", "type-2", "type-3")
+            )
         assertTrue(
             !differentTypesWithSameCorrelationMessagesResWithMissingType.correlated,
             "failed to correlate differentTypesWithSameCorrelationMessagesResWithMissingType"
@@ -118,4 +121,12 @@ class MessageCorrelationUtilsTest {
             "failed to correlate differentTypesWithDifferentCorrelationMessageResp"
         )
     }
+
+    @Test
+    fun testPrioritizationOrdering() {
+        val differentPriorityMessages = MessagePrioritizationSample
+            .sampleMessageWithSameCorrelation("sample-group", MessageState.NEW.name, 5)
+        val orderedPriorityMessages = differentPriorityMessages.orderByHighestPriority()
+        assertNotNull(orderedPriorityMessages, "failed to order the priority messages")
+    }
 }