gerrit.onap Code Review - policy/clamp.git/blob

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2024 Nordix Foundation.
   4  *  Modifications Copyright (C) 2021 AT&T Intellectual Property. All rights reserved.
   5  * ================================================================================
   6  * Licensed under the Apache License, Version 2.0 (the "License");
   7  * you may not use this file except in compliance with the License.
   8  * You may obtain a copy of the License at
   9  *
  10  *      http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  *
  18  * SPDX-License-Identifier: Apache-2.0
  19  * ============LICENSE_END=========================================================
  20  */
  21
  22 package org.onap.policy.clamp.acm.participant.intermediary.handler;
  23
  24 import java.util.ArrayList;
  25 import java.util.List;
  26 import java.util.UUID;
  27 import java.util.stream.Collectors;
  28 import lombok.RequiredArgsConstructor;
  29 import org.onap.policy.clamp.acm.participant.intermediary.api.CompositionDto;
  30 import org.onap.policy.clamp.acm.participant.intermediary.comm.ParticipantMessagePublisher;
  31 import org.onap.policy.clamp.models.acm.concepts.AcTypeState;
  32 import org.onap.policy.clamp.models.acm.concepts.AutomationCompositionElementDefinition;
  33 import org.onap.policy.clamp.models.acm.concepts.ParticipantState;
  34 import org.onap.policy.clamp.models.acm.concepts.StateChangeResult;
  35 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantPrime;
  36 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantPrimeAck;
  37 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantSync;
  38 import org.springframework.stereotype.Component;
  39
  40 @Component
  41 @RequiredArgsConstructor
  42 public class AcDefinitionHandler {
  43
  44     private final CacheProvider cacheProvider;
  45     private final ParticipantMessagePublisher publisher;
  46     private final ThreadHandler listener;
  47
  48     /**
  49      * Handle a participant Prime message.
  50      *
  51      * @param participantPrimeMsg the ParticipantPrime message
  52      */
  53     public void handlePrime(ParticipantPrime participantPrimeMsg) {
  54         if (!participantPrimeMsg.getParticipantDefinitionUpdates().isEmpty()) {
  55             // prime
  56             List<AutomationCompositionElementDefinition> list = new ArrayList<>();
  57             for (var participantDefinition : participantPrimeMsg.getParticipantDefinitionUpdates()) {
  58                 if (participantDefinition.getParticipantId().equals(cacheProvider.getParticipantId())) {
  59                     list.addAll(participantDefinition.getAutomationCompositionElementDefinitionList());
  60                 }
  61             }
  62             if (!list.isEmpty()) {
  63                 cacheProvider.addElementDefinition(participantPrimeMsg.getCompositionId(), list);
  64                 prime(participantPrimeMsg.getMessageId(), participantPrimeMsg.getCompositionId(), list);
  65             }
  66         } else {
  67             // deprime
  68             deprime(participantPrimeMsg.getMessageId(), participantPrimeMsg.getCompositionId());
  69         }
  70     }
  71
  72     private void prime(UUID messageId, UUID compositionId, List<AutomationCompositionElementDefinition> list) {
  73         var inPropertiesMap = list.stream().collect(Collectors.toMap(
  74                 AutomationCompositionElementDefinition::getAcElementDefinitionId,
  75                 el -> el.getAutomationCompositionElementToscaNodeTemplate().getProperties()));
  76         var outPropertiesMap = list.stream().collect(Collectors.toMap(
  77                 AutomationCompositionElementDefinition::getAcElementDefinitionId,
  78                 AutomationCompositionElementDefinition::getOutProperties));
  79         listener.prime(messageId, new CompositionDto(compositionId, inPropertiesMap, outPropertiesMap));
  80     }
  81
  82     private void deprime(UUID messageId, UUID compositionId) {
  83         var acElementsDefinitions = cacheProvider.getAcElementsDefinitions().get(compositionId);
  84         if (acElementsDefinitions == null) {
  85             // this participant does not handle this composition
  86             var participantPrimeAck = new ParticipantPrimeAck();
  87             participantPrimeAck.setCompositionId(compositionId);
  88             participantPrimeAck.setMessage("Already deprimed or never primed");
  89             participantPrimeAck.setResult(true);
  90             participantPrimeAck.setResponseTo(messageId);
  91             participantPrimeAck.setCompositionState(AcTypeState.COMMISSIONED);
  92             participantPrimeAck.setStateChangeResult(StateChangeResult.NO_ERROR);
  93             participantPrimeAck.setParticipantId(cacheProvider.getParticipantId());
  94             participantPrimeAck.setReplicaId(cacheProvider.getReplicaId());
  95             participantPrimeAck.setState(ParticipantState.ON_LINE);
  96             publisher.sendParticipantPrimeAck(participantPrimeAck);
  97             return;
  98         }
  99         var list = new ArrayList<>(acElementsDefinitions.values());
 100         var inPropertiesMap = list.stream().collect(Collectors.toMap(
 101                 AutomationCompositionElementDefinition::getAcElementDefinitionId,
 102                 el -> el.getAutomationCompositionElementToscaNodeTemplate().getProperties()));
 103         var outPropertiesMap = list.stream().collect(Collectors.toMap(
 104                 AutomationCompositionElementDefinition::getAcElementDefinitionId,
 105                 AutomationCompositionElementDefinition::getOutProperties));
 106         listener.deprime(messageId, new CompositionDto(compositionId, inPropertiesMap, outPropertiesMap));
 107     }
 108
 109     /**
 110      * Handle a Participant Sync message.
 111      *
 112      * @param participantSyncMsg the participantRestart message
 113      */
 114     public void handleParticipantSync(ParticipantSync participantSyncMsg) {
 115
 116         if (participantSyncMsg.isDelete()) {
 117             deleteScenario(participantSyncMsg);
 118             return;
 119         }
 120
 121         if (!participantSyncMsg.getParticipantDefinitionUpdates().isEmpty()) {
 122             if (StateChangeResult.TIMEOUT.equals(participantSyncMsg.getStateChangeResult())) {
 123                 listener.cleanExecution(participantSyncMsg.getCompositionId(), participantSyncMsg.getMessageId());
 124             }
 125
 126             List<AutomationCompositionElementDefinition> list = new ArrayList<>();
 127             for (var participantDefinition : participantSyncMsg.getParticipantDefinitionUpdates()) {
 128                 list.addAll(participantDefinition.getAutomationCompositionElementDefinitionList());
 129             }
 130             cacheProvider.addElementDefinition(participantSyncMsg.getCompositionId(), list);
 131         }
 132
 133         for (var automationcomposition : participantSyncMsg.getAutomationcompositionList()) {
 134             cacheProvider
 135                     .initializeAutomationComposition(participantSyncMsg.getCompositionId(), automationcomposition);
 136             if (StateChangeResult.TIMEOUT.equals(automationcomposition.getStateChangeResult())) {
 137                 for (var element : automationcomposition.getAcElementList()) {
 138                     listener.cleanExecution(element.getId(), participantSyncMsg.getMessageId());
 139                 }
 140             }
 141         }
 142     }
 143
 144     private void deleteScenario(ParticipantSync participantSyncMsg) {
 145         if (AcTypeState.COMMISSIONED.equals(participantSyncMsg.getState())) {
 146             cacheProvider.removeElementDefinition(participantSyncMsg.getCompositionId());
 147         }
 148         for (var automationcomposition : participantSyncMsg.getAutomationcompositionList()) {
 149             cacheProvider.removeAutomationComposition(automationcomposition.getAutomationCompositionId());
 150         }
 151     }
 152 }