Add Synchronization topic in acm runtime

New sync topic for acm-ppnt synchronization
Added publisher for the sync topic
Refactor MessageDispatcherActivator for processing more than one topic
parameter.

Issue-ID: POLICY-5030
Change-Id: Id765b433beaf3f51fad9a9c66403a93d21c33797
Signed-off-by: zrrmmua <ramesh.murugan.iyer@est.tech>

5 files changed, 72 insertions, 19 deletions

diff --git a/runtime-acm/src/test/java/org/onap/policy/clamp/acm/runtime/config/messaging/MessageDispatcherActivatorTest.java b/runtime-acm/src/test/java/org/onap/policy/clamp/acm/runtime/config/messaging/MessageDispatcherActivatorTest.java
index 899e35f33..66595c89a 100644
--- a/runtime-acm/src/test/java/org/onap/policy/clamp/acm/runtime/config/messaging/MessageDispatcherActivatorTest.java
+++ b/runtime-acm/src/test/java/org/onap/policy/clamp/acm/runtime/config/messaging/MessageDispatcherActivatorTest.java
@@ -78,8 +78,8 @@ class MessageDispatcherActivatorTest {
             // repeat start - should throw an exception
             assertThatIllegalStateException().isThrownBy(activator::start);
             assertTrue(activator.isAlive());
-            verify(publisherFirst, times(1)).active(anyList());
-            verify(publisherSecond, times(1)).active(anyList());
+            verify(publisherFirst, times(1)).active(any());
+            verify(publisherSecond, times(1)).active(any());
 
             var sco = CODER.decode("{messageType:" + TOPIC_FIRST + "}", StandardCoderObject.class);
             activator.getMsgDispatcher().onTopicEvent(null, "msg", sco);
diff --git a/runtime-acm/src/test/java/org/onap/policy/clamp/acm/runtime/supervision/comm/SupervisionMessagesTest.java b/runtime-acm/src/test/java/org/onap/policy/clamp/acm/runtime/supervision/comm/SupervisionMessagesTest.java
index 295d2d781..31cd659b3 100644
--- a/runtime-acm/src/test/java/org/onap/policy/clamp/acm/runtime/supervision/comm/SupervisionMessagesTest.java
+++ b/runtime-acm/src/test/java/org/onap/policy/clamp/acm/runtime/supervision/comm/SupervisionMessagesTest.java
@@ -36,6 +36,7 @@ import java.util.UUID;
 import org.junit.jupiter.api.Test;
 import org.onap.policy.clamp.acm.runtime.instantiation.InstantiationUtils;
 import org.onap.policy.clamp.acm.runtime.main.parameters.AcRuntimeParameterGroup;
+import org.onap.policy.clamp.acm.runtime.main.parameters.Topics;
 import org.onap.policy.clamp.acm.runtime.participants.AcmParticipantProvider;
 import org.onap.policy.clamp.acm.runtime.supervision.SupervisionAcHandler;
 import org.onap.policy.clamp.acm.runtime.supervision.SupervisionHandler;
@@ -74,7 +75,7 @@ class SupervisionMessagesTest {
     void testSendParticipantRegisterAck() {
         var acRegisterAckPublisher = new ParticipantRegisterAckPublisher();
         var topicSink = mock(TopicSink.class);
-        acRegisterAckPublisher.active(List.of(topicSink));
+        acRegisterAckPublisher.active(topicSink);
         acRegisterAckPublisher.send(new ParticipantRegisterAck());
         verify(topicSink).send(anyString());
         acRegisterAckPublisher.stop();
@@ -100,7 +101,7 @@ class SupervisionMessagesTest {
     void testSendParticipantDeregisterAck() {
         var acDeregisterAckPublisher = new ParticipantDeregisterAckPublisher();
         var topicSink = mock(TopicSink.class);
-        acDeregisterAckPublisher.active(Collections.singletonList(topicSink));
+        acDeregisterAckPublisher.active(topicSink);
         acDeregisterAckPublisher.send(new ParticipantDeregisterAck());
         verify(topicSink).send(anyString());
         acDeregisterAckPublisher.stop();
@@ -140,7 +141,7 @@ class SupervisionMessagesTest {
     void testSendAutomationCompositionStateChangePublisher() {
         var publisher = new AutomationCompositionStateChangePublisher();
         var topicSink = mock(TopicSink.class);
-        publisher.active(List.of(topicSink));
+        publisher.active(topicSink);
         publisher.send(getAutomationComposition(), 0, true);
         verify(topicSink).send(anyString());
         publisher.stop();
@@ -151,7 +152,7 @@ class SupervisionMessagesTest {
         var publisher = new ParticipantPrimePublisher(mock(ParticipantProvider.class),
                 mock(AcmParticipantProvider.class), mock(AcRuntimeParameterGroup.class));
         var topicSink = mock(TopicSink.class);
-        publisher.active(List.of(topicSink));
+        publisher.active(topicSink);
         publisher.sendDepriming(UUID.randomUUID());
         verify(topicSink).send(anyString());
     }
@@ -173,7 +174,7 @@ class SupervisionMessagesTest {
         var publisher = new ParticipantPrimePublisher(participantProvider, mock(AcmParticipantProvider.class),
                 CommonTestData.getTestParamaterGroup());
         var topicSink = mock(TopicSink.class);
-        publisher.active(List.of(topicSink));
+        publisher.active(topicSink);
         var serviceTemplate = InstantiationUtils.getToscaServiceTemplate(TOSCA_SERVICE_TEMPLATE_YAML);
         serviceTemplate.setName("Name");
         serviceTemplate.setVersion("1.0.0");
@@ -192,7 +193,7 @@ class SupervisionMessagesTest {
     void testParticipantStatusReqPublisher() {
         var publisher = new ParticipantStatusReqPublisher();
         var topicSink = mock(TopicSink.class);
-        publisher.active(List.of(topicSink));
+        publisher.active(topicSink);
         publisher.send(CommonTestData.getParticipantId());
         verify(topicSink).send(anyString());
     }
@@ -201,7 +202,7 @@ class SupervisionMessagesTest {
     void testParticipantRegisterAckPublisher() {
         var publisher = new ParticipantRegisterAckPublisher();
         var topicSink = mock(TopicSink.class);
-        publisher.active(List.of(topicSink));
+        publisher.active(topicSink);
         publisher.send(UUID.randomUUID(), CommonTestData.getParticipantId());
         verify(topicSink).send(anyString());
     }
@@ -210,7 +211,7 @@ class SupervisionMessagesTest {
     void testParticipantDeregisterAckPublisher() {
         var publisher = new ParticipantDeregisterAckPublisher();
         var topicSink = mock(TopicSink.class);
-        publisher.active(List.of(topicSink));
+        publisher.active(topicSink);
         publisher.send(UUID.randomUUID());
         verify(topicSink).send(anyString());
     }
@@ -219,7 +220,7 @@ class SupervisionMessagesTest {
     void testAcElementPropertiesPublisher() {
         var publisher = new AcElementPropertiesPublisher();
         var topicSink = mock(TopicSink.class);
-        publisher.active(List.of(topicSink));
+        publisher.active(topicSink);
         var automationComposition =
                 InstantiationUtils.getAutomationCompositionFromResource(AC_INSTANTIATION_UPDATE_JSON, "Crud");
         publisher.send(automationComposition);
@@ -230,7 +231,7 @@ class SupervisionMessagesTest {
     void testAutomationCompositionMigrationPublisher() {
         var publisher = new AutomationCompositionMigrationPublisher();
         var topicSink = mock(TopicSink.class);
-        publisher.active(List.of(topicSink));
+        publisher.active(topicSink);
         var automationComposition =
                 InstantiationUtils.getAutomationCompositionFromResource(AC_INSTANTIATION_UPDATE_JSON, "Crud");
         publisher.send(automationComposition, UUID.randomUUID());
@@ -241,7 +242,31 @@ class SupervisionMessagesTest {
     void testParticipantRestartPublisher() {
         var publisher = new ParticipantRestartPublisher(CommonTestData.getTestParamaterGroup());
         var topicSink = mock(TopicSink.class);
-        publisher.active(List.of(topicSink));
+        publisher.active(topicSink);
+
+        var serviceTemplate = InstantiationUtils.getToscaServiceTemplate(TOSCA_SERVICE_TEMPLATE_YAML);
+        var acmDefinition = new AutomationCompositionDefinition();
+        acmDefinition.setCompositionId(UUID.randomUUID());
+        acmDefinition.setServiceTemplate(serviceTemplate);
+        var acElements = AcmUtils
+                .extractAcElementsFromServiceTemplate(serviceTemplate, "");
+        acmDefinition.setElementStateMap(AcmUtils.createElementStateMap(acElements, AcTypeState.PRIMED));
+
+        var automationComposition =
+                InstantiationUtils.getAutomationCompositionFromResource(AC_INSTANTIATION_UPDATE_JSON, "Crud");
+
+        var participantId = automationComposition.getElements().values().iterator().next().getParticipantId();
+        acmDefinition.getElementStateMap().values().iterator().next().setParticipantId(participantId);
+
+        publisher.send(participantId, acmDefinition, List.of(automationComposition));
+        verify(topicSink).send(anyString());
+    }
+
+    @Test
+    void testParticipantSyncPublisher() {
+        var publisher = new ParticipantSyncPublisher(CommonTestData.getTestParamaterGroup());
+        var topicSink = mock(TopicSink.class);
+        publisher.active(topicSink);
 
         var serviceTemplate = InstantiationUtils.getToscaServiceTemplate(TOSCA_SERVICE_TEMPLATE_YAML);
         var acmDefinition = new AutomationCompositionDefinition();
diff --git a/runtime-acm/src/test/resources/application-prometheus-noauth.yaml b/runtime-acm/src/test/resources/application-prometheus-noauth.yaml
index 1c71252f3..620e7534d 100644
--- a/runtime-acm/src/test/resources/application-prometheus-noauth.yaml
+++ b/runtime-acm/src/test/resources/application-prometheus-noauth.yaml
@@ -19,13 +19,16 @@ server:
     context-path: /onap/policy/clamp/acm
 
 runtime:
+  topics:
+    operationTopic: policy-acruntime-participant
+    syncTopic: acm-ppnt-sync
   participantParameters:
     updateParameters:
       maxRetryCount: 3
   topicParameterGroup:
     topicSources:
       -
-        topic: POLICY-ACRUNTIME-PARTICIPANT
+        topic: ${runtime.topics.operationTopic}
         servers:
           - localhost
         topicCommInfrastructure: noop
@@ -35,7 +38,12 @@ runtime:
         topicCommInfrastructure: noop
         servers:
           - localhost
-        topic: POLICY-ACRUNTIME-PARTICIPANT
+        topic: ${runtime.topics.operationTopic}
+
+      - topic: ${runtime.topics.syncTopic}
+        servers:
+          - ${topicServer:kafka:9092}
+        topicCommInfrastructure: noop
 
 tracing:
   enabled: true
diff --git a/runtime-acm/src/test/resources/application-test.yaml b/runtime-acm/src/test/resources/application-test.yaml
index 13b1f788a..5d616d529 100644
--- a/runtime-acm/src/test/resources/application-test.yaml
+++ b/runtime-acm/src/test/resources/application-test.yaml
@@ -19,13 +19,16 @@ server:
     context-path: /onap/policy/clamp/acm
 
 runtime:
+  topics:
+    operationTopic: policy-acruntime-participant
+    syncTopic: acm-ppnt-sync
   participantParameters:
     updateParameters:
       maxRetryCount: 3
   topicParameterGroup:
     topicSources:
       -
-        topic: policy-acruntime-participant
+        topic: ${runtime.topics.operationTopic}
         servers:
           - kafka:9092
         topicCommInfrastructure: NOOP
@@ -35,7 +38,12 @@ runtime:
         topicCommInfrastructure: NOOP
         servers:
           - kafka:9092
-        topic: policy-acruntime-participant
+        topic: ${runtime.topics.operationTopic}
+      -
+        topic: ${runtime.topics.syncTopic}
+        servers:
+            - ${topicServer:kafka:9092}
+        topicCommInfrastructure: NOOP
   acmParameters:
     acElementName: org.onap.policy.clamp.acm.AutomationCompositionElement
     acNodeType: org.onap.policy.clamp.acm.AutomationComposition
diff --git a/runtime-acm/src/test/resources/parameters/TestParameters.json b/runtime-acm/src/test/resources/parameters/TestParameters.json
index 8192b7214..1558abc2e 100644
--- a/runtime-acm/src/test/resources/parameters/TestParameters.json
+++ b/runtime-acm/src/test/resources/parameters/TestParameters.json
@@ -20,10 +20,15 @@
         "databasePassword": "P01icY",
         "persistenceUnit": "InstantiationTests"
     },
+    "topics":{
+        "operationTopic": "policy-acruntime-participant",
+        "syncTopic": "acm-ppnt-sync"
+    },
     "topicParameterGroup": {
+
         "topicSources": [
             {
-                "topic": "POLICY-ACRUNTIME-PARTICIPANT",
+                "topic": "${topics.operationTopic}",
                 "servers": [
                     "localhost"
                 ],
@@ -33,7 +38,14 @@
         ],
         "topicSinks": [
             {
-                "topic": "POLICY-ACRUNTIME-PARTICIPANT",
+                "topic": "${topics.operationTopic}",
+                "servers": [
+                    "localhost"
+                ],
+                "topicCommInfrastructure": "NOOP"
+            },
+            {
+                "topic": "${topics.syncTopic}",
                 "servers": [
                     "localhost"
                 ],