Add Sync topic for participant Intermediary

Add new sync topic config for Intermediary
Add sync topic listener
Refactor IntermediaryActivator for processing multiple topic source

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

2 files changed, 30 insertions, 6 deletions

diff --git a/participant/participant-impl/participant-impl-simulator/src/main/resources/config/application.yaml b/participant/participant-impl/participant-impl-simulator/src/main/resources/config/application.yaml
index f3731d7a8..77a3ef443 100644
--- a/participant/participant-impl/participant-impl-simulator/src/main/resources/config/application.yaml
+++ b/participant/participant-impl/participant-impl-simulator/src/main/resources/config/application.yaml
@@ -14,18 +14,26 @@ security:
   enable-csrf: false
 participant:
   intermediaryParameters:
+    topics:
+      operationTopic: policy-acruntime-participant
+      syncTopic: acm-ppnt-sync
     reportingTimeIntervalMs: 120000
     description: Participant Description
     participantId: 101c62b3-8918-41b9-a747-d21eb79c6c90
     clampAutomationCompositionTopics:
       topicSources:
-        - topic: policy-acruntime-participant
+        - topic: ${participant.intermediaryparameters.topics.operationTopic}
+          servers:
+            - ${topicServer:kafka:9092}
+          topicCommInfrastructure: NOOP
+          fetchTimeout: 15000
+        - topic: ${participant.intermediaryparameters.topics.syncTopic}
           servers:
             - ${topicServer:kafka:9092}
           topicCommInfrastructure: NOOP
           fetchTimeout: 15000
       topicSinks:
-        - topic: policy-acruntime-participant
+        - topic: ${participant.intermediaryparameters.topics.operationTopic}
           servers:
             - ${topicServer:kafka:9092}
           topicCommInfrastructure: NOOP
diff --git a/participant/participant-impl/participant-impl-simulator/src/test/java/org/onap/policy/clamp/acm/participant/sim/comm/CommonTestData.java b/participant/participant-impl/participant-impl-simulator/src/test/java/org/onap/policy/clamp/acm/participant/sim/comm/CommonTestData.java
index 54f193039..5499931a2 100644
--- a/participant/participant-impl/participant-impl-simulator/src/test/java/org/onap/policy/clamp/acm/participant/sim/comm/CommonTestData.java
+++ b/participant/participant-impl/participant-impl-simulator/src/test/java/org/onap/policy/clamp/acm/participant/sim/comm/CommonTestData.java
@@ -25,6 +25,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
 import java.util.UUID;
+import org.onap.policy.clamp.acm.participant.intermediary.parameters.Topics;
 import org.onap.policy.clamp.acm.participant.sim.parameters.ParticipantSimParameters;
 import org.onap.policy.clamp.models.acm.concepts.AutomationComposition;
 import org.onap.policy.clamp.models.acm.concepts.AutomationCompositionElement;
@@ -37,7 +38,8 @@ public class CommonTestData {
     public static final Coder CODER = new StandardCoder();
     public static final String DESCRIPTION = "Participant description";
     public static final long TIME_INTERVAL = 2000;
-    public static final List<TopicParameters> TOPIC_PARAMS = List.of(getTopicParams());
+    public static final List<TopicParameters> SINK_TOPIC_PARAMS = List.of(getSinkTopicParams());
+    public static final List<TopicParameters> SOURCE_TOPIC_PARAMS = List.of(getSinkTopicParams(), getSyncTopicParams());
 
     /**
      * Get ParticipantSimParameters.
@@ -77,6 +79,7 @@ public class CommonTestData {
         map.put("participantId", getParticipantId());
         map.put("clampAutomationCompositionTopics", getTopicParametersMap());
         map.put("participantSupportedElementTypes", new ArrayList<>());
+        map.put("topics", new Topics("policy-acruntime-participant", "acm-ppnt-sync"));
 
         return map;
     }
@@ -88,8 +91,8 @@ public class CommonTestData {
      */
     private static Map<String, Object> getTopicParametersMap() {
         final Map<String, Object> map = new TreeMap<>();
-        map.put("topicSources", TOPIC_PARAMS);
-        map.put("topicSinks", TOPIC_PARAMS);
+        map.put("topicSources", SOURCE_TOPIC_PARAMS);
+        map.put("topicSinks", SINK_TOPIC_PARAMS);
         return map;
     }
 
@@ -98,7 +101,7 @@ public class CommonTestData {
      *
      * @return topic parameters
      */
-    private static TopicParameters getTopicParams() {
+    private static TopicParameters getSinkTopicParams() {
         final TopicParameters topicParams = new TopicParameters();
         topicParams.setTopic("policy-acruntime-participant");
         topicParams.setTopicCommInfrastructure("NOOP");
@@ -107,6 +110,19 @@ public class CommonTestData {
     }
 
     /**
+     * Returns sync topic parameters for test cases.
+     *
+     * @return topic parameters
+     */
+    private static TopicParameters getSyncTopicParams() {
+        final TopicParameters topicParams = new TopicParameters();
+        topicParams.setTopic("acm-ppnt-sync");
+        topicParams.setTopicCommInfrastructure("NOOP");
+        topicParams.setServers(List.of("localhost"));
+        return topicParams;
+    }
+
+    /**
      * Returns participantId for test cases.
      *
      * @return participant Id