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

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  *  Copyright (C) 2021 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.controlloop.participant.intermediary.handler;
  22
  23 import java.io.Closeable;
  24 import java.io.IOException;
  25 import java.util.List;
  26 import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantMessageType;
  27 import org.onap.policy.clamp.controlloop.participant.intermediary.api.ParticipantIntermediaryApi;
  28 import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ControlLoopStateChangeListener;
  29 import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ControlLoopUpdateListener;
  30 import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ParticipantDeregisterAckListener;
  31 import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ParticipantHealthCheckListener;
  32 import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ParticipantRegisterAckListener;
  33 import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ParticipantUpdateListener;
  34 import org.onap.policy.clamp.controlloop.participant.intermediary.parameters.ParticipantParameters;
  35 import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager;
  36 import org.onap.policy.common.endpoints.event.comm.TopicSource;
  37 import org.onap.policy.common.endpoints.listeners.MessageTypeDispatcher;
  38 import org.onap.policy.common.utils.services.ServiceManagerContainer;
  39 import org.springframework.context.ApplicationContext;
  40 import org.springframework.context.event.ContextClosedEvent;
  41 import org.springframework.context.event.ContextRefreshedEvent;
  42 import org.springframework.context.event.EventListener;
  43 import org.springframework.stereotype.Component;
  44
  45 /**
  46  * This class activates the Participant Intermediary together with all its handlers.
  47  */
  48 @Component
  49 public class IntermediaryActivator extends ServiceManagerContainer implements Closeable {
  50
  51     private final ApplicationContext applicationContext;
  52
  53     // Topics from which the participant receives and to which the participant sends messages
  54     private List<TopicSource> topicSources;
  55
  56     ParticipantIntermediaryApi participantIntermediaryApi;
  57
  58     /**
  59      * Instantiate the activator for participant.
  60      *
  61      * @param applicationContext ApplicationContext
  62      * @param parameters the ParticipantParameters
  63      */
  64     public IntermediaryActivator(final ApplicationContext applicationContext, final ParticipantParameters parameters,
  65             ParticipantIntermediaryApi participantIntermediaryApi) {
  66         this.applicationContext = applicationContext;
  67         this.participantIntermediaryApi = participantIntermediaryApi;
  68
  69         topicSources = TopicEndpointManager.getManager()
  70                 .addTopicSources(parameters.getIntermediaryParameters().getClampControlLoopTopics().getTopicSources());
  71
  72         // @formatter:off
  73
  74         addAction("Topic endpoint management",
  75             () -> TopicEndpointManager.getManager().start(),
  76             () -> TopicEndpointManager.getManager().shutdown());
  77
  78         addAction("Topic Message Dispatcher", this::registerMsgDispatcher, this::unregisterMsgDispatcher);
  79         // @formatter:on
  80     }
  81
  82     /**
  83      * Handle ContextRefreshEvent.
  84      *
  85      * @param ctxRefreshedEvent ContextRefreshedEvent
  86      */
  87     @EventListener
  88     public void handleContextRefreshEvent(ContextRefreshedEvent ctxRefreshedEvent) {
  89         if (!isAlive()) {
  90             start();
  91             sendParticipantRegister();
  92         }
  93     }
  94
  95     /**
  96      * Handle ContextClosedEvent.
  97      *
  98      * @param ctxClosedEvent ContextClosedEvent
  99      */
 100     @EventListener
 101     public void handleContextClosedEvent(ContextClosedEvent ctxClosedEvent) {
 102         if (isAlive()) {
 103             sendParticipantDeregister();
 104             stop();
 105         }
 106     }
 107
 108     private void sendParticipantRegister() {
 109         participantIntermediaryApi.sendParticipantRegister();
 110     }
 111
 112     private void sendParticipantDeregister() {
 113         participantIntermediaryApi.sendParticipantDeregister();
 114     }
 115
 116     /**
 117      * Registers the dispatcher with the topic source(s).
 118      */
 119     private void registerMsgDispatcher() {
 120         MessageTypeDispatcher msgDispatcher = applicationContext.getBean(MessageTypeDispatcher.class);
 121
 122         msgDispatcher.register(ParticipantMessageType.PARTICIPANT_HEALTH_CHECK.name(),
 123                 applicationContext.getBean(ParticipantHealthCheckListener.class));
 124
 125         msgDispatcher.register(ParticipantMessageType.PARTICIPANT_CONTROL_LOOP_STATE_CHANGE.name(),
 126                 applicationContext.getBean(ControlLoopStateChangeListener.class));
 127
 128         msgDispatcher.register(ParticipantMessageType.CONTROL_LOOP_UPDATE.name(),
 129                 applicationContext.getBean(ControlLoopUpdateListener.class));
 130
 131         msgDispatcher.register(ParticipantMessageType.PARTICIPANT_REGISTER_ACK.name(),
 132                 applicationContext.getBean(ParticipantRegisterAckListener.class));
 133
 134         msgDispatcher.register(ParticipantMessageType.PARTICIPANT_DEREGISTER_ACK.name(),
 135                 applicationContext.getBean(ParticipantDeregisterAckListener.class));
 136
 137         msgDispatcher.register(ParticipantMessageType.PARTICIPANT_UPDATE.name(),
 138                 applicationContext.getBean(ParticipantUpdateListener.class));
 139
 140         for (final TopicSource source : topicSources) {
 141             source.register(msgDispatcher);
 142         }
 143     }
 144
 145     /**
 146      * Unregisters the dispatcher from the topic source(s).
 147      */
 148     private void unregisterMsgDispatcher() {
 149         MessageTypeDispatcher msgDispatcher = applicationContext.getBean(MessageTypeDispatcher.class);
 150
 151         for (final TopicSource source : topicSources) {
 152             source.unregister(msgDispatcher);
 153         }
 154     }
 155
 156     @Override
 157     public void close() throws IOException {
 158         super.shutdown();
 159     }
 160 }