Message Prioritization message group lock.

[ccsdk/cds.git] / ms / blueprintsprocessor / functions / message-prioritizaion / src / main / kotlin / org / onap / ccsdk / cds / blueprintsprocessor / functions / message / prioritization / utils / MessageProcessorUtils.kt

diff --git a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageProcessorUtils.kt b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageProcessorUtils.kt
index 7e5862c..d1f38f4 100644 (file)
--- a/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageProcessorUtils.kt
+++ b/ms/blueprintsprocessor/functions/message-prioritizaion/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/functions/message/prioritization/utils/MessageProcessorUtils.kt
@@ -17,14 +17,40 @@
 package org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.utils
 
 import org.apache.kafka.streams.processor.ProcessorSupplier
+import org.onap.ccsdk.cds.blueprintsprocessor.core.service.BluePrintClusterService
+import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterLock
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.AbstractMessagePrioritizeProcessor
 import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.PrioritizationConfiguration
+import org.onap.ccsdk.cds.blueprintsprocessor.functions.message.prioritization.db.MessagePrioritization
+import org.onap.ccsdk.cds.controllerblueprints.core.BluePrintProcessorException
 import org.onap.ccsdk.cds.controllerblueprints.core.service.BluePrintDependencyService
 
 object MessageProcessorUtils {
 
+    /** Utility to create the cluster lock for message [messagePrioritization] */
+    suspend fun prioritizationGrouplock(
+        clusterService: BluePrintClusterService?,
+        messagePrioritization: MessagePrioritization
+    ): ClusterLock? {
+        return if (clusterService != null && clusterService.clusterJoined()) {
+            val lockName = "prioritization-${messagePrioritization.group}"
+            val clusterLock = clusterService.clusterLock(lockName)
+            clusterLock.lock()
+            if (!clusterLock.isLocked()) throw BluePrintProcessorException("failed to lock($lockName)")
+            clusterLock
+        } else null
+    }
+
+    /** Utility used to cluster unlock for message [messagePrioritization] */
+    suspend fun prioritizationGroupUnLock(clusterService: BluePrintClusterService?, clusterLock: ClusterLock?) {
+        if (clusterService != null && clusterService.clusterJoined() && clusterLock != null) {
+            clusterLock.unLock()
+            clusterLock.close()
+        }
+    }
+
     fun <K, V> bluePrintProcessorSupplier(name: String, prioritizationConfiguration: PrioritizationConfiguration):
-            ProcessorSupplier<K, V> {
+        ProcessorSupplier<K, V> {
         return ProcessorSupplier<K, V> {
             // Dynamically resolve the Prioritization Processor
             val processorInstance = BluePrintDependencyService.instance<AbstractMessagePrioritizeProcessor<K, V>>(name)