2 * Copyright © 2018-2019 AT&T Intellectual Property.
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
8 * http://www.apache.org/licenses/LICENSE-2.0
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
17 package org.onap.ccsdk.cds.blueprintsprocessor.atomix
19 import com.fasterxml.jackson.databind.JsonNode
20 import io.atomix.core.Atomix
21 import kotlinx.coroutines.Dispatchers
22 import kotlinx.coroutines.async
23 import kotlinx.coroutines.awaitAll
24 import kotlinx.coroutines.delay
25 import kotlinx.coroutines.runBlocking
26 import kotlinx.coroutines.withContext
27 import org.junit.Before
29 import org.onap.ccsdk.cds.blueprintsprocessor.atomix.service.AtomixBluePrintClusterService
30 import org.onap.ccsdk.cds.blueprintsprocessor.atomix.utils.AtomixLibUtils
31 import org.onap.ccsdk.cds.blueprintsprocessor.core.service.ClusterInfo
32 import org.onap.ccsdk.cds.controllerblueprints.core.asJsonPrimitive
33 import org.onap.ccsdk.cds.controllerblueprints.core.deleteNBDir
34 import org.onap.ccsdk.cds.controllerblueprints.core.logger
35 import kotlin.test.assertNotNull
36 import kotlin.test.assertTrue
38 class AtomixBluePrintClusterServiceTest {
39 val log = logger(AtomixBluePrintClusterServiceTest::class)
44 deleteNBDir("target/cluster")
48 /** Testing two cluster with distributed map store creation, This is time consuming test casetake around 10s **/
50 fun testClusterJoin() {
52 val members = arrayListOf("node-5679", "node-5680")
53 val deferred = arrayListOf(5679, 5680).map { port ->
54 async(Dispatchers.IO) {
55 val nodeId = "node-$port"
56 log.info("********** Starting node($nodeId) on port($port)")
57 val clusterInfo = ClusterInfo(
58 id = "test-cluster", nodeId = nodeId,
59 clusterMembers = members, nodeAddress = "localhost:$port", storagePath = "target/cluster"
61 val atomixClusterService = AtomixBluePrintClusterService()
62 atomixClusterService.startCluster(clusterInfo)
63 atomixClusterService.atomix
66 val atomixs = deferred.awaitAll()
67 testDistributedStore(atomixs)
68 testDistributedLock(atomixs)
72 private suspend fun testDistributedStore(atomix: List<Atomix>) {
73 /** Test Distributed store creation */
74 repeat(2) { storeId ->
75 val store = AtomixLibUtils.distributedMapStore<JsonNode>(atomix.get(0), "blueprint-runtime-$storeId")
76 assertNotNull(store, "failed to get store")
77 val store1 = AtomixLibUtils.distributedMapStore<JsonNode>(atomix.get(1), "blueprint-runtime-$storeId")
79 log.info("Received map event : $it")
82 store["key-$storeId-$it"] = "value-$it".asJsonPrimitive()
89 private suspend fun testDistributedLock(atomix: List<Atomix>) {
90 val lockName = "sample-lock"
91 withContext(Dispatchers.IO) {
92 val deferred = async {
93 executeLock(atomix.get(0), "first", lockName)
95 val deferred2 = async {
96 executeLock(atomix.get(1), "second", lockName)
98 val deferred3 = async {
99 executeLock(atomix.get(1), "third", lockName)
107 private suspend fun executeLock(atomix: Atomix, lockId: String, lockName: String) {
108 log.info("initialising $lockId lock...")
109 val distributedLock = AtomixLibUtils.distributedLock(atomix, lockName)
110 assertNotNull(distributedLock, "failed to create distributed $lockId lock")
111 distributedLock.lock()
112 assertTrue(distributedLock.isLocked, "failed to lock $lockId")
114 log.info("locked $lockId process for 5mSec")
117 distributedLock.unlock()
118 log.info("$lockId lock released")
120 distributedLock.close()