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

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2023-2025 OpenInfra Foundation Europe. All rights reserved.
   4  * ================================================================================
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *      http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  * SPDX-License-Identifier: Apache-2.0
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.clamp.acm.participant.intermediary.handler;
  22
  23 import io.opentelemetry.context.Context;
  24 import java.util.concurrent.ExecutorService;
  25 import java.util.concurrent.Executors;
  26 import lombok.RequiredArgsConstructor;
  27 import org.onap.policy.clamp.acm.participant.intermediary.comm.ParticipantMessagePublisher;
  28 import org.onap.policy.clamp.acm.participant.intermediary.handler.cache.AutomationCompositionMsg;
  29 import org.onap.policy.clamp.acm.participant.intermediary.handler.cache.CacheProvider;
  30 import org.onap.policy.clamp.common.acm.utils.AcmThreadFactory;
  31 import org.onap.policy.clamp.models.acm.messages.kafka.participant.ParticipantReqSync;
  32 import org.slf4j.Logger;
  33 import org.slf4j.LoggerFactory;
  34 import org.springframework.stereotype.Component;
  35
  36 @Component
  37 @RequiredArgsConstructor
  38 public class MsgExecutor {
  39
  40     private final ExecutorService executor =
  41             Context.taskWrapping(Executors.newSingleThreadExecutor(new AcmThreadFactory()));
  42     private static final Logger LOGGER = LoggerFactory.getLogger(MsgExecutor.class);
  43
  44     private final CacheProvider cacheProvider;
  45     private final ParticipantMessagePublisher publisher;
  46
  47     /**
  48      * Execute the message if all data are present or put on Hold if something is missing.
  49      *
  50      * @param message the message
  51      */
  52     public void execute(AutomationCompositionMsg<?> message) {
  53         if (validExecution(message)) {
  54             message.execute();
  55         } else {
  56             cacheProvider.getMessagesOnHold().put(message.getKey(), message);
  57             var participantReqSync = new ParticipantReqSync();
  58             participantReqSync.setParticipantId(cacheProvider.getParticipantId());
  59             participantReqSync.setReplicaId(cacheProvider.getReplicaId());
  60             participantReqSync.setCompositionId(message.getCompositionId());
  61             participantReqSync.setAutomationCompositionId(message.getInstanceId());
  62             participantReqSync.setCompositionTargetId(message.getCompositionTargetId());
  63             publisher.sendParticipantReqSync(participantReqSync);
  64         }
  65     }
  66
  67     /**
  68      * Check if messages on hold can be executed.
  69      */
  70     public void check() {
  71         executor.submit(this::checkAndExecute);
  72     }
  73
  74     private void checkAndExecute() {
  75         var executable = cacheProvider.getMessagesOnHold().values().stream()
  76                 .filter(this::validExecution).toList();
  77         executable.forEach(AutomationCompositionMsg::execute);
  78         executable.forEach(msg -> cacheProvider.getMessagesOnHold().remove(msg.getKey()));
  79     }
  80
  81     private boolean validExecution(AutomationCompositionMsg<?> message) {
  82         var result = true;
  83         if (message.getCompositionId() != null) {
  84             var valid = cacheProvider.isCompositionDefinitionUpdated(message.getCompositionId(),
  85                     message.getRevisionIdComposition());
  86             if (valid) {
  87                 message.setCompositionId(null);
  88                 message.setRevisionIdComposition(null);
  89             } else {
  90                 LOGGER.debug("Composition {} missing or outdated", message.getCompositionId());
  91                 result = false;
  92             }
  93         }
  94         if (message.getCompositionTargetId() != null) {
  95             var valid = cacheProvider.isCompositionDefinitionUpdated(message.getCompositionTargetId(),
  96                     message.getRevisionIdCompositionTarget());
  97             if (valid) {
  98                 message.setCompositionTargetId(null);
  99                 message.setRevisionIdCompositionTarget(null);
 100             } else {
 101                 LOGGER.debug("Composition Target {} missing or outdated", message.getCompositionTargetId());
 102                 result = false;
 103             }
 104         }
 105         if (message.getInstanceId() != null) {
 106             var valid = cacheProvider.isInstanceUpdated(message.getInstanceId(), message.getRevisionIdInstance());
 107             if (valid) {
 108                 message.setInstanceId(null);
 109                 message.setRevisionIdInstance(null);
 110             } else {
 111                 LOGGER.debug("Instance {} missing or outdated", message.getInstanceId());
 112                 result = false;
 113             }
 114         }
 115         return result;
 116     }
 117 }