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