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

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2025 Nordix Foundation.
   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 java.util.List;
  24 import java.util.Timer;
  25 import java.util.TimerTask;
  26 import org.onap.policy.clamp.acm.participant.intermediary.parameters.ParticipantParameters;
  27 import org.onap.policy.common.message.bus.event.Topic;
  28 import org.onap.policy.common.message.bus.healthcheck.TopicHealthCheck;
  29 import org.onap.policy.common.message.bus.healthcheck.TopicHealthCheckFactory;
  30 import org.onap.policy.common.parameters.topic.TopicParameters;
  31 import org.slf4j.Logger;
  32 import org.slf4j.LoggerFactory;
  33 import org.springframework.context.event.ContextClosedEvent;
  34 import org.springframework.context.event.ContextRefreshedEvent;
  35 import org.springframework.context.event.EventListener;
  36 import org.springframework.stereotype.Component;
  37
  38 @Component
  39 public class BrokerStarter<T> {
  40     private static final Logger LOGGER = LoggerFactory.getLogger(BrokerStarter.class);
  41     private final IntermediaryActivator activator;
  42     private final ParticipantHandler participantHandler;
  43     private final TopicHealthCheck topicHealthCheck;
  44
  45     private final ParticipantParameters parameters;
  46     private final List<Publisher> publishers;
  47     private final List<Listener<T>> listeners;
  48
  49     /**
  50      * Constructor.
  51      *
  52      * @param parameters ParticipantParameters
  53      * @param activator IntermediaryActivator
  54      * @param participantHandler participantHandler
  55      */
  56     public BrokerStarter(ParticipantParameters parameters,
  57             List<Publisher> publishers, List<Listener<T>> listeners, IntermediaryActivator activator,
  58             ParticipantHandler participantHandler) {
  59         this.parameters = parameters;
  60         this.listeners = listeners;
  61         this.publishers = publishers;
  62         this.activator = activator;
  63         this.participantHandler = participantHandler;
  64         var topic = parameters.getIntermediaryParameters().getClampAdminTopics();
  65         if (topic == null) {
  66             topic = new TopicParameters();
  67             topic.setTopicCommInfrastructure(Topic.CommInfrastructure.NOOP.name());
  68         }
  69         this.topicHealthCheck = createTopicHealthCheck(topic);
  70     }
  71
  72     protected TopicHealthCheck createTopicHealthCheck(TopicParameters topic) {
  73         return new TopicHealthCheckFactory().getTopicHealthCheck(topic);
  74     }
  75
  76     /**
  77      * Handle ContextRefreshEvent.
  78      *
  79      * @param ctxRefreshedEvent ContextRefreshedEvent
  80      */
  81     @EventListener
  82     public void handleContextRefreshEvent(ContextRefreshedEvent ctxRefreshedEvent) {
  83         if (!activator.isAlive()) {
  84             runTopicHealthCheck();
  85             start();
  86         }
  87     }
  88
  89     private void runTopicHealthCheck() {
  90         var fetchTimeout = getFetchTimeout();
  91         while (!topicHealthCheck.healthCheck(getTopics())) {
  92             LOGGER.debug(" Broker not up yet!");
  93             try {
  94                 Thread.sleep(fetchTimeout);
  95             } catch (InterruptedException e) {
  96                 LOGGER.error(e.getMessage());
  97                 Thread.currentThread().interrupt();
  98             }
  99         }
 100     }
 101
 102     private List<String> getTopics() {
 103         var opTopic = parameters.getIntermediaryParameters().getTopics().getOperationTopic();
 104         var syncTopic = parameters.getIntermediaryParameters().getTopics().getSyncTopic();
 105         return Boolean.TRUE.equals(parameters.getIntermediaryParameters().getTopicValidation())
 106                 ? List.of(opTopic, syncTopic) : List.<String>of();
 107     }
 108
 109     private int getFetchTimeout() {
 110         int fetchTimeout = parameters.getIntermediaryParameters().getClampAdminTopics() == null
 111                 ? 0 : parameters.getIntermediaryParameters().getClampAdminTopics().getFetchTimeout();
 112         return Math.max(fetchTimeout, 5000);
 113     }
 114
 115     private void start() {
 116         activator.config(parameters, publishers, listeners);
 117         activator.start();
 118         var task = new TimerTask() {
 119             @Override
 120             public void run() {
 121                 new Thread(participantHandler::sendParticipantRegister).start();
 122             }
 123         };
 124         new Timer().schedule(task, 5000);
 125     }
 126
 127
 128     /**
 129      * Handle ContextClosedEvent.
 130      *
 131      * @param ctxClosedEvent ContextClosedEvent
 132      */
 133     @EventListener
 134     public void handleContextClosedEvent(ContextClosedEvent ctxClosedEvent) {
 135         if (activator.isAlive()) {
 136             participantHandler.sendParticipantDeregister();
 137             activator.stop();
 138         }
 139     }
 140 }