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

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 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 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         var retries = 10; // TODO - make this configurable with max number of retries or timeout
  92         while (!topicHealthCheck.healthCheck(getTopics())) {
  93             LOGGER.debug(" Broker not up yet!");
  94             try {
  95                 Thread.sleep(fetchTimeout);
  96                 retries--;
  97                 if (retries == 0) {
  98                     LOGGER.error("Broker not up after {} retries", retries);
  99                     break;
 100                 }
 101             } catch (InterruptedException e) {
 102                 LOGGER.error(e.getMessage());
 103                 Thread.currentThread().interrupt();
 104             }
 105         }
 106     }
 107
 108     private List<String> getTopics() {
 109         var opTopic = parameters.getIntermediaryParameters().getTopics().getOperationTopic();
 110         var syncTopic = parameters.getIntermediaryParameters().getTopics().getSyncTopic();
 111         return Boolean.TRUE.equals(parameters.getIntermediaryParameters().getTopicValidation())
 112                 ? List.of(opTopic, syncTopic) : List.of();
 113     }
 114
 115     private int getFetchTimeout() {
 116         int fetchTimeout = parameters.getIntermediaryParameters().getClampAdminTopics() == null
 117                 ? 0 : parameters.getIntermediaryParameters().getClampAdminTopics().getFetchTimeout();
 118         return Math.max(fetchTimeout, 5000);
 119     }
 120
 121     private void start() {
 122         activator.config(parameters, publishers, listeners);
 123         activator.start();
 124         var task = new TimerTask() {
 125             @Override
 126             public void run() {
 127                 new Thread(participantHandler::sendParticipantRegister).start();
 128             }
 129         };
 130         new Timer().schedule(task, 5000);
 131     }
 132
 133
 134     /**
 135      * Handle ContextClosedEvent.
 136      *
 137      * @param ctxClosedEvent ContextClosedEvent
 138      */
 139     @EventListener
 140     public void handleContextClosedEvent(ContextClosedEvent ctxClosedEvent) {
 141         if (activator.isAlive()) {
 142             participantHandler.sendParticipantDeregister();
 143             activator.stop();
 144         }
 145     }
 146 }