1 files changed, 137 insertions, 0 deletions
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AcDefinitionHandler.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AcDefinitionHandler.java
new file mode 100644
index 000000000..e1d4b0959
--- /dev/null
+++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/acm/participant/intermediary/handler/AcDefinitionHandler.java
@@ -0,0 +1,137 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2024 Nordix Foundation.
+ *  Modifications Copyright (C) 2021 AT&T Intellectual Property. All rights reserved.
+ * ================================================================================
+ * 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.handler;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import lombok.RequiredArgsConstructor;
+import org.onap.policy.clamp.acm.participant.intermediary.api.CompositionDto;
+import org.onap.policy.clamp.acm.participant.intermediary.comm.ParticipantMessagePublisher;
+import org.onap.policy.clamp.models.acm.concepts.AcTypeState;
+import org.onap.policy.clamp.models.acm.concepts.AutomationCompositionElementDefinition;
+import org.onap.policy.clamp.models.acm.concepts.ParticipantState;
+import org.onap.policy.clamp.models.acm.concepts.StateChangeResult;
+import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantPrime;
+import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantPrimeAck;
+import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantRestart;
+import org.springframework.stereotype.Component;
+
+@Component
+@RequiredArgsConstructor
+public class AcDefinitionHandler {
+
+    private final CacheProvider cacheProvider;
+    private final ParticipantMessagePublisher publisher;
+    private final ThreadHandler listener;
+
+    /**
+     * Handle a participant Prime message.
+     *
+     * @param participantPrimeMsg the ParticipantPrime message
+     */
+    public void handlePrime(ParticipantPrime participantPrimeMsg) {
+        if (!participantPrimeMsg.getParticipantDefinitionUpdates().isEmpty()) {
+            // prime
+            List<AutomationCompositionElementDefinition> list = new ArrayList<>();
+            for (var participantDefinition : participantPrimeMsg.getParticipantDefinitionUpdates()) {
+                if (participantDefinition.getParticipantId().equals(cacheProvider.getParticipantId())) {
+                    list.addAll(participantDefinition.getAutomationCompositionElementDefinitionList());
+                }
+            }
+            if (!list.isEmpty()) {
+                cacheProvider.addElementDefinition(participantPrimeMsg.getCompositionId(), list);
+                prime(participantPrimeMsg.getMessageId(), participantPrimeMsg.getCompositionId(), list);
+            }
+        } else {
+            // deprime
+            deprime(participantPrimeMsg.getMessageId(), participantPrimeMsg.getCompositionId());
+        }
+    }
+
+    private void prime(UUID messageId, UUID compositionId, List<AutomationCompositionElementDefinition> list) {
+        var inPropertiesMap = list.stream().collect(Collectors.toMap(
+                AutomationCompositionElementDefinition::getAcElementDefinitionId,
+                el -> el.getAutomationCompositionElementToscaNodeTemplate().getProperties()));
+        var outPropertiesMap = list.stream().collect(Collectors.toMap(
+                AutomationCompositionElementDefinition::getAcElementDefinitionId,
+                AutomationCompositionElementDefinition::getOutProperties));
+        listener.prime(messageId, new CompositionDto(compositionId, inPropertiesMap, outPropertiesMap));
+    }
+
+    private void deprime(UUID messageId, UUID compositionId) {
+        var acElementsDefinitions = cacheProvider.getAcElementsDefinitions().get(compositionId);
+        if (acElementsDefinitions == null) {
+            // this participant does not handle this composition
+            var participantPrimeAck = new ParticipantPrimeAck();
+            participantPrimeAck.setCompositionId(compositionId);
+            participantPrimeAck.setMessage("Already deprimed or never primed");
+            participantPrimeAck.setResult(true);
+            participantPrimeAck.setResponseTo(messageId);
+            participantPrimeAck.setCompositionState(AcTypeState.COMMISSIONED);
+            participantPrimeAck.setStateChangeResult(StateChangeResult.NO_ERROR);
+            participantPrimeAck.setParticipantId(cacheProvider.getParticipantId());
+            participantPrimeAck.setState(ParticipantState.ON_LINE);
+            publisher.sendParticipantPrimeAck(participantPrimeAck);
+            return;
+        }
+        var list = new ArrayList<>(acElementsDefinitions.values());
+        var inPropertiesMap = list.stream().collect(Collectors.toMap(
+                AutomationCompositionElementDefinition::getAcElementDefinitionId,
+                el -> el.getAutomationCompositionElementToscaNodeTemplate().getProperties()));
+        var outPropertiesMap = list.stream().collect(Collectors.toMap(
+                AutomationCompositionElementDefinition::getAcElementDefinitionId,
+                AutomationCompositionElementDefinition::getOutProperties));
+        listener.deprime(messageId, new CompositionDto(compositionId, inPropertiesMap, outPropertiesMap));
+    }
+
+    /**
+     * Handle a ParticipantRestart message.
+     *
+     * @param participantRestartMsg the participantRestart message
+     */
+    public void handleParticipantRestart(ParticipantRestart participantRestartMsg) {
+        List<AutomationCompositionElementDefinition> list = new ArrayList<>();
+        for (var participantDefinition : participantRestartMsg.getParticipantDefinitionUpdates()) {
+            list.addAll(participantDefinition.getAutomationCompositionElementDefinitionList());
+        }
+        if (!AcTypeState.COMMISSIONED.equals(participantRestartMsg.getState())) {
+            cacheProvider.addElementDefinition(participantRestartMsg.getCompositionId(), list);
+        }
+
+        for (var automationcomposition : participantRestartMsg.getAutomationcompositionList()) {
+            cacheProvider
+                    .initializeAutomationComposition(participantRestartMsg.getCompositionId(), automationcomposition);
+        }
+        var inPropertiesMap = list.stream().collect(Collectors.toMap(
+                AutomationCompositionElementDefinition::getAcElementDefinitionId,
+                el -> el.getAutomationCompositionElementToscaNodeTemplate().getProperties()));
+        var outPropertiesMap = list.stream().collect(Collectors.toMap(
+                AutomationCompositionElementDefinition::getAcElementDefinitionId,
+                AutomationCompositionElementDefinition::getOutProperties));
+        var composition =
+                new CompositionDto(participantRestartMsg.getCompositionId(), inPropertiesMap, outPropertiesMap);
+        listener.restarted(participantRestartMsg.getMessageId(), composition, participantRestartMsg.getState(),
+                participantRestartMsg.getAutomationcompositionList());
+    }
+}