summaryrefslogtreecommitdiffstats
path: root/ms/blueprintsprocessor/modules
diff options
context:
space:
mode:
authorBrinda Santh <bs2796@att.com>2020-01-02 11:59:29 -0500
committerBrinda Santh <bs2796@att.com>2020-01-02 11:59:29 -0500
commit4f4e2de08d3c6259da2497950a96d549d3e82f8a (patch)
tree4c5a8e7a97d96e9de73ecb7ed1b53597ee0ebdd5 /ms/blueprintsprocessor/modules
parentba75d2fad2b0111a510f4ee4cc87e658fb32ac4b (diff)
Message Prioritization message group lock.
Implementation to avoid concurrent procession of message group while prioritization. Sample message prioritization Kafka listener properties. Issue-ID: CCSDK-2011 Signed-off-by: Brinda Santh <bs2796@att.com> Change-Id: Ifbf39985b03c662b6ccf7740be711cfeb7bfbebb
Diffstat (limited to 'ms/blueprintsprocessor/modules')
-rw-r--r--ms/blueprintsprocessor/modules/commons/atomix-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/atomix/service/AtomixBluePrintClusterService.kt7
-rw-r--r--ms/blueprintsprocessor/modules/commons/processor-core/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/core/service/BluePrintClusterService.kt1
2 files changed, 8 insertions, 0 deletions
diff --git a/ms/blueprintsprocessor/modules/commons/atomix-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/atomix/service/AtomixBluePrintClusterService.kt b/ms/blueprintsprocessor/modules/commons/atomix-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/atomix/service/AtomixBluePrintClusterService.kt
index 0690eb89d..214a14310 100644
--- a/ms/blueprintsprocessor/modules/commons/atomix-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/atomix/service/AtomixBluePrintClusterService.kt
+++ b/ms/blueprintsprocessor/modules/commons/atomix-lib/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/atomix/service/AtomixBluePrintClusterService.kt
@@ -148,12 +148,18 @@ open class AtomixBluePrintClusterService : BluePrintClusterService {
}
open class ClusterLockImpl(private val atomix: Atomix, private val name: String) : ClusterLock {
+ val log = logger(ClusterLockImpl::class)
lateinit var distributedLock: DistributedLock
+ override fun name(): String {
+ return distributedLock.name()
+ }
+
override suspend fun lock() {
distributedLock = AtomixLibUtils.distributedLock(atomix, name)
distributedLock.lock()
+ log.debug("Cluster lock($name) created..")
}
override suspend fun tryLock(timeout: Long): Boolean {
@@ -163,6 +169,7 @@ open class ClusterLockImpl(private val atomix: Atomix, private val name: String)
override suspend fun unLock() {
distributedLock.unlock()
+ log.debug("Cluster unlock(${name()}) successfully..")
}
override fun isLocked(): Boolean {
diff --git a/ms/blueprintsprocessor/modules/commons/processor-core/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/core/service/BluePrintClusterService.kt b/ms/blueprintsprocessor/modules/commons/processor-core/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/core/service/BluePrintClusterService.kt
index 21fcfc509..f994628a2 100644
--- a/ms/blueprintsprocessor/modules/commons/processor-core/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/core/service/BluePrintClusterService.kt
+++ b/ms/blueprintsprocessor/modules/commons/processor-core/src/main/kotlin/org/onap/ccsdk/cds/blueprintsprocessor/core/service/BluePrintClusterService.kt
@@ -57,6 +57,7 @@ data class ClusterInfo(
data class ClusterMember(val id: String, val memberAddress: String?, val state: String? = null)
interface ClusterLock {
+ fun name(): String
suspend fun lock()
suspend fun tryLock(timeout: Long): Boolean
suspend fun unLock()