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>

11 files changed, 189 insertions, 14 deletions

diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/comm/ParticipantAckListener.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/comm/ParticipantAckListener.java
index ec14e6cd7..4ed89dce8 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/comm/ParticipantAckListener.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/comm/ParticipantAckListener.java
@@ -64,4 +64,9 @@ public abstract class ParticipantAckListener<T extends ParticipantAckMessage> ex
     public ScoListener<T> getScoListener() {
         return this;
     }
+
+    @Override
+    public boolean isDefaultTopic() {
+        return true;
+    }
 }
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/comm/ParticipantListener.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/comm/ParticipantListener.java
index bf2892aa8..87eb5c8a1 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/comm/ParticipantListener.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/comm/ParticipantListener.java
@@ -61,4 +61,9 @@ public abstract class ParticipantListener<T extends ParticipantMessage> extends
     public ScoListener<T> getScoListener() {
         return this;
     }
+
+    @Override
+    public boolean isDefaultTopic() {
+        return true;
+    }
 }
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/comm/ParticipantSyncListener.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/comm/ParticipantSyncListener.java
new file mode 100644
index 000000000..0b359f98a
--- /dev/null
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/comm/ParticipantSyncListener.java
@@ -0,0 +1,49 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2024 Nordix Foundation.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.clamp.acm.participant.intermediary.comm;
+
+import org.onap.policy.clamp.acm.participant.intermediary.handler.ParticipantHandler;
+import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantMessageType;
+import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantSync;
+import org.springframework.stereotype.Component;
+
+@Component
+public class ParticipantSyncListener extends ParticipantListener<ParticipantSync> {
+
+    /**
+     * Constructs the object.
+     *
+     * @param participantHandler the handler for managing the state of the participant
+     */
+    public ParticipantSyncListener(ParticipantHandler participantHandler) {
+        super(ParticipantSync.class, participantHandler, participantHandler::handleParticipantSync);
+    }
+
+    @Override
+    public String getType() {
+        return ParticipantMessageType.PARTICIPANT_SYNC_MSG.name();
+    }
+
+    @Override
+    public boolean isDefaultTopic() {
+        return false;
+    }
+}
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/IntermediaryActivator.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/IntermediaryActivator.java
index a77d5242a..2c54a22dd 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/IntermediaryActivator.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/IntermediaryActivator.java
@@ -1,6 +1,6 @@
 /*-
  * ============LICENSE_START=======================================================
- *  Copyright (C) 2021 Nordix Foundation.
+ *  Copyright (C) 2021,2024 Nordix Foundation.
  *  Modifications Copyright (C) 2021 AT&T Intellectual Property. All rights reserved.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -28,6 +28,7 @@ import java.util.Timer;
 import java.util.TimerTask;
 import lombok.Getter;
 import org.onap.policy.clamp.acm.participant.intermediary.parameters.ParticipantParameters;
+import org.onap.policy.clamp.acm.participant.intermediary.parameters.Topics;
 import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager;
 import org.onap.policy.common.endpoints.event.comm.TopicSink;
 import org.onap.policy.common.endpoints.event.comm.TopicSource;
@@ -55,6 +56,9 @@ public class IntermediaryActivator extends ServiceManagerContainer implements Cl
     @Getter
     private final MessageTypeDispatcher msgDispatcher;
 
+    @Getter
+    private final MessageTypeDispatcher syncMsgDispatcher;
+
     /**
      * Instantiate the activator for participant.
      *
@@ -75,22 +79,32 @@ public class IntermediaryActivator extends ServiceManagerContainer implements Cl
 
         msgDispatcher = new MessageTypeDispatcher(MSG_TYPE_NAMES);
 
+        syncMsgDispatcher = new MessageTypeDispatcher(MSG_TYPE_NAMES);
+
         // @formatter:off
         addAction("Topic endpoint management",
             () -> TopicEndpointManager.getManager().start(),
             () -> TopicEndpointManager.getManager().shutdown());
 
-        listeners.forEach(listener ->
-                addAction("Listener " + listener.getClass().getSimpleName(),
+        listeners.stream().filter(Listener::isDefaultTopic)
+                .forEach(listener -> addAction("Listener " + listener.getClass().getSimpleName(),
                         () -> msgDispatcher.register(listener.getType(), listener.getScoListener()),
                         () -> msgDispatcher.unregister(listener.getType())));
 
+        listeners.stream().filter(l -> ! l.isDefaultTopic())
+                .forEach(listener -> addAction("Listener " + listener.getClass().getSimpleName(),
+                        () -> syncMsgDispatcher.register(listener.getType(), listener.getScoListener()),
+                        () -> syncMsgDispatcher.unregister(listener.getType())));
+
         publishers.forEach(publisher ->
             addAction("Publisher " + publisher.getClass().getSimpleName(),
                 () -> publisher.active(topicSinks),
                 publisher::stop));
 
-        addAction("Topic Message Dispatcher", this::registerMsgDispatcher, this::unregisterMsgDispatcher);
+        var topics = parameters.getIntermediaryParameters().getTopics();
+
+        addAction("Topic Message Dispatcher", () -> this.registerMsgDispatcher(topics),
+                () -> this.unregisterMsgDispatcher(topics));
         // @formatter:on
     }
 
@@ -133,18 +147,26 @@ public class IntermediaryActivator extends ServiceManagerContainer implements Cl
     /**
      * Registers the dispatcher with the topic source(s).
      */
-    private void registerMsgDispatcher() {
-        for (final TopicSource source : topicSources) {
-            source.register(msgDispatcher);
+    private void registerMsgDispatcher(Topics topics) {
+        for (final var source : topicSources) {
+            if (source.getTopic().equals(topics.getOperationTopic())) {
+                source.register(msgDispatcher);
+            } else if (source.getTopic().equals(topics.getSyncTopic())) {
+                source.register(syncMsgDispatcher);
+            }
         }
     }
 
     /**
      * Unregisters the dispatcher from the topic source(s).
      */
-    private void unregisterMsgDispatcher() {
-        for (final TopicSource source : topicSources) {
-            source.unregister(msgDispatcher);
+    private void unregisterMsgDispatcher(Topics topics) {
+        for (final var source : topicSources) {
+            if (source.getTopic().equals(topics.getOperationTopic())) {
+                source.unregister(msgDispatcher);
+            } else if (source.getTopic().equals(topics.getSyncTopic())) {
+                source.unregister(syncMsgDispatcher);
+            }
         }
     }
 
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/Listener.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/Listener.java
index 56ed55441..27585cf20 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/Listener.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/Listener.java
@@ -1,6 +1,6 @@
 /*-
  * ============LICENSE_START=======================================================
- *  Copyright (C) 2021 Nordix Foundation.
+ *  Copyright (C) 2021,2024 Nordix Foundation.
  *  Modifications Copyright (C) 2021 AT&T Intellectual Property. All rights reserved.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
@@ -38,4 +38,10 @@ public interface Listener<T> {
      * @return listener to register
      */
     ScoListener<T> getScoListener();
+
+    /**
+     * Check if default topic.
+     * @return true if default topic
+     */
+    boolean isDefaultTopic();
 }
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ParticipantHandler.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ParticipantHandler.java
index 54a05912a..7ac58ae6c 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ParticipantHandler.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/ParticipantHandler.java
@@ -39,6 +39,7 @@ import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantRe
 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantRestart;
 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantStatus;
 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantStatusReq;
+import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantSync;
 import org.onap.policy.clamp.models.acm.messages.kafka.participant.PropertiesUpdate;
 import org.onap.policy.clamp.models.acm.messages.rest.instantiation.DeployOrder;
 import org.slf4j.Logger;
@@ -209,6 +210,18 @@ public class ParticipantHandler {
     }
 
     /**
+     * Handle a ParticipantSync message.
+     *
+     * @param participantSyncMsg the participantSync message
+     */
+    @Timed(value = "listener.participant_sync_msg", description = "PARTICIPANT_SYNC messages received")
+    public void handleParticipantSync(ParticipantSync participantSyncMsg) {
+        LOGGER.debug("ParticipantSync message received for participantId {}",
+                participantSyncMsg.getParticipantId());
+        acDefinitionHandler.handleParticipantRestart(participantSyncMsg);
+    }
+
+    /**
      * Dispatch a heartbeat for this participant.
      */
     public void sendHeartbeat() {
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/parameters/ParticipantIntermediaryParameters.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/parameters/ParticipantIntermediaryParameters.java
index 54774044b..1c36ad17f 100644
--- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/parameters/ParticipantIntermediaryParameters.java
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/parameters/ParticipantIntermediaryParameters.java
@@ -1,6 +1,6 @@
 /*-
  * ============LICENSE_START=======================================================
- *  Copyright (C) 2021-2023 Nordix Foundation.
+ *  Copyright (C) 2021-2024 Nordix Foundation.
  * ================================================================================
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -60,4 +60,8 @@ public class ParticipantIntermediaryParameters {
     @Valid
     private List<ParticipantSupportedElementType> participantSupportedElementTypes;
 
+    @NotNull
+    @Valid
+    private Topics topics;
+
 }
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/parameters/Topics.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/parameters/Topics.java
new file mode 100644
index 000000000..ddf72052f
--- /dev/null
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/parameters/Topics.java
@@ -0,0 +1,44 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2024 Nordix Foundation.
+ * ================================================================================
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.clamp.acm.participant.intermediary.parameters;
+
+import jakarta.validation.Valid;
+import jakarta.validation.constraints.NotNull;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * Class to hold topic names for operation and synchronization.
+ */
+@Getter
+@Setter
+@AllArgsConstructor
+public class Topics {
+
+    @NotNull
+    @Valid
+    private String operationTopic;
+
+    @NotNull
+    @Valid
+    private String syncTopic;
+}
diff --git a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/comm/ParticipantCommTest.java b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/comm/ParticipantCommTest.java
index 10f9d4586..d7c97bd00 100644
--- a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/comm/ParticipantCommTest.java
+++ b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/comm/ParticipantCommTest.java
@@ -82,6 +82,10 @@ class ParticipantCommTest {
         assertEquals(ParticipantMessageType.PARTICIPANT_RESTART.name(),
                 participantRestartListener.getType());
 
+        var participantSyncListener = new ParticipantSyncListener(participantHandler);
+        assertEquals(ParticipantMessageType.PARTICIPANT_SYNC_MSG.name(),
+                participantSyncListener.getType());
+
         var acMigrationListener = new AutomationCompositionMigrationListener(participantHandler);
         assertEquals(ParticipantMessageType.AUTOMATION_COMPOSITION_MIGRATION.name(), acMigrationListener.getType());
     }
diff --git a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/IntermediaryActivatorTest.java b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/IntermediaryActivatorTest.java
index ac9cbfad7..8868c733a 100644
--- a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/IntermediaryActivatorTest.java
+++ b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/handler/IntermediaryActivatorTest.java
@@ -34,8 +34,10 @@ import static org.mockito.Mockito.when;
 import java.util.List;
 import org.junit.jupiter.api.Test;
 import org.onap.policy.clamp.acm.participant.intermediary.comm.ParticipantStatusReqListener;
+import org.onap.policy.clamp.acm.participant.intermediary.comm.ParticipantSyncListener;
 import org.onap.policy.clamp.acm.participant.intermediary.main.parameters.CommonTestData;
 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantStatusReq;
+import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantSync;
 import org.onap.policy.common.utils.coder.Coder;
 import org.onap.policy.common.utils.coder.StandardCoder;
 import org.onap.policy.common.utils.coder.StandardCoderObject;
@@ -59,10 +61,12 @@ class IntermediaryActivatorTest {
         var listenerFirst = mock(ParticipantStatusReqListener.class);
         when(listenerFirst.getType()).thenReturn(TOPIC_FIRST);
         when(listenerFirst.getScoListener()).thenReturn(listenerFirst);
+        when(listenerFirst.isDefaultTopic()).thenReturn(true);
 
         var listenerSecond = mock(ParticipantStatusReqListener.class);
         when(listenerSecond.getType()).thenReturn(TOPIC_SECOND);
         when(listenerSecond.getScoListener()).thenReturn(listenerSecond);
+        when(listenerSecond.isDefaultTopic()).thenReturn(false);
 
         List<Listener<ParticipantStatusReq>> listeners = List.of(listenerFirst, listenerSecond);
 
@@ -84,7 +88,7 @@ class IntermediaryActivatorTest {
             verify(listenerFirst, times(1)).onTopicEvent(any(), any(), any());
 
             sco = CODER.decode("{messageType:" + TOPIC_SECOND + "}", StandardCoderObject.class);
-            activator.getMsgDispatcher().onTopicEvent(null, "msg", sco);
+            activator.getSyncMsgDispatcher().onTopicEvent(null, "msg", sco);
             verify(listenerSecond, times(1)).onTopicEvent(any(), any(), any());
 
             activator.close();
diff --git a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/main/parameters/CommonTestData.java b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/main/parameters/CommonTestData.java
index 1536a0be0..e8cafa96f 100644
--- a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/main/parameters/CommonTestData.java
+++ b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/acm/participant/intermediary/main/parameters/CommonTestData.java
@@ -30,6 +30,7 @@ import java.util.TreeMap;
 import java.util.UUID;
 import org.onap.policy.clamp.acm.participant.intermediary.handler.DummyParticipantParameters;
 import org.onap.policy.clamp.acm.participant.intermediary.parameters.ParticipantIntermediaryParameters;
+import org.onap.policy.clamp.acm.participant.intermediary.parameters.Topics;
 import org.onap.policy.clamp.models.acm.concepts.AcElementDeploy;
 import org.onap.policy.clamp.models.acm.concepts.AcElementRestart;
 import org.onap.policy.clamp.models.acm.concepts.AutomationComposition;
@@ -58,6 +59,7 @@ public class CommonTestData {
     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> TOPIC_SOURCE_PARAMS = List.of(getTopicParams(), getSyncTopicParams());
     public static final Coder CODER = new StandardCoder();
     public static final UUID AC_ID_0 = UUID.randomUUID();
     public static final UUID AC_ID_1 = UUID.randomUUID();
@@ -116,6 +118,7 @@ public class CommonTestData {
         map.put("description", DESCRIPTION);
         map.put("reportingTimeIntervalMs", TIME_INTERVAL);
         map.put("clampAutomationCompositionTopics", getTopicParametersMap(false));
+        map.put("topics", getTopics());
         var supportedElementType = new ParticipantSupportedElementType();
         supportedElementType.setTypeName("org.onap.policy.clamp.acm.HttpAutomationCompositionElement");
         supportedElementType.setTypeVersion("1.0.0");
@@ -133,7 +136,7 @@ public class CommonTestData {
     public static Map<String, Object> getTopicParametersMap(final boolean isEmpty) {
         final Map<String, Object> map = new TreeMap<>();
         if (!isEmpty) {
-            map.put("topicSources", TOPIC_PARAMS);
+            map.put("topicSources", TOPIC_SOURCE_PARAMS);
             map.put("topicSinks", TOPIC_PARAMS);
         }
         return map;
@@ -153,6 +156,22 @@ public class CommonTestData {
     }
 
     /**
+     * Returns topic parameters for sync topic.
+     * @return topicparamaters
+     */
+    public static TopicParameters getSyncTopicParams() {
+        final var topicParams = new TopicParameters();
+        topicParams.setTopic("acm-ppnt-sync");
+        topicParams.setTopicCommInfrastructure("NOOP");
+        topicParams.setServers(List.of("localhost"));
+        return topicParams;
+    }
+
+    private static Topics getTopics() {
+        return new Topics("policy-acruntime-participant", "acm-ppnt-sync");
+    }
+
+    /**
      * Returns participantId for test cases.
      *
      * @return participant Id