diff options
Diffstat (limited to 'tosca-controlloop/participant/participant-intermediary')
12 files changed, 460 insertions, 93 deletions
diff --git a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/api/ControlLoopElementListener.java b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/api/ControlLoopElementListener.java index ff24d6a10..9e5d2c663 100644 --- a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/api/ControlLoopElementListener.java +++ b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/api/ControlLoopElementListener.java @@ -24,7 +24,6 @@ import java.util.UUID; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoopElement; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoopOrderedState; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoopState; -import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoops; import org.onap.policy.models.base.PfModelException; import org.onap.policy.models.tosca.authorative.concepts.ToscaServiceTemplate; @@ -38,17 +37,26 @@ public interface ControlLoopElementListener { * @param controlLoopElementId the ID of the control loop element * @param currentState the current state of the control loop element * @param newState the state to which the control loop element is changing to + * @throws PfModelException in case of a model exception */ public void controlLoopElementStateChange(UUID controlLoopElementId, ControlLoopState currentState, - ControlLoopOrderedState newState); + ControlLoopOrderedState newState) throws PfModelException; /** * Handle an update on a control loop element. * * @param element the information on the control loop element * @param controlLoopDefinition toscaServiceTemplate - * @throws PfModelException in case of a model exception + * @throws PfModelException from Policy framework */ public void controlLoopElementUpdate(ControlLoopElement element, ToscaServiceTemplate controlLoopDefinition) throws PfModelException; + + /** + * Handle controlLoopElement statistics. + * + * @param controlLoopElementId controlLoopElement id + * @throws PfModelException in case of a model exception + */ + public void handleStatistics(UUID controlLoopElementId) throws PfModelException; } diff --git a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/api/ParticipantIntermediaryApi.java b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/api/ParticipantIntermediaryApi.java index defb1cfa1..d31ae1082 100644 --- a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/api/ParticipantIntermediaryApi.java +++ b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/api/ParticipantIntermediaryApi.java @@ -21,11 +21,13 @@ package org.onap.policy.clamp.controlloop.participant.intermediary.api; import java.util.List; +import java.util.Map; import java.util.UUID; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ClElementStatistics; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoop; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoopElement; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoopOrderedState; +import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoopState; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoops; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.Participant; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ParticipantState; @@ -98,39 +100,40 @@ public interface ParticipantIntermediaryApi { * @param version the control loop element version, null for all * @return the control loop elements */ - List<ControlLoopElement> getControlLoopElements(String name, String version); + Map<UUID, ControlLoopElement> getControlLoopElements(String name, String version); /** - * Update the state of a control loop. + * Get control loop element from the intermediary API. * - * @param definition the ID of the control loop to update the state on - * @param state the state of the control loop - * @return ControlLoop updated control loop + * @param id control loop element ID + * @return the control loop element */ - ControlLoop updateControlLoopState(ToscaConceptIdentifier definition, ControlLoopOrderedState state); + ControlLoopElement getControlLoopElement(UUID id); /** * Update the state of a control loop element. * * @param id the ID of the control loop element to update the state on - * @param state the state of the control loop element + * @param currentState the state of the control loop element + * @param newState the state of the control loop element * @return ControlLoopElement updated control loop element */ - ControlLoopElement updateControlLoopElementState(UUID id, ControlLoopOrderedState state); + ControlLoopElement updateControlLoopElementState(UUID id, ControlLoopOrderedState currentState, + ControlLoopState newState); /** * Update the control loop element statistics. * + * @param id the ID of the control loop element to update the state on * @param elementStatistics the updated statistics */ - void updateControlLoopElementStatistics(ClElementStatistics elementStatistics); + void updateControlLoopElementStatistics(UUID id, ClElementStatistics elementStatistics); /** - * Returns participantHandler, This will not be used in real world, but for junits, + * Return participantHandler, This will not be used in real world, but for junits, * if participantHandler is not returned, there is no way to test state change messages * without dmaap simulator. * - * @return ParticipantHandler returns a participantHandler */ ParticipantHandler getParticipantHandler(); } diff --git a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/api/impl/ParticipantIntermediaryApiImpl.java b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/api/impl/ParticipantIntermediaryApiImpl.java index 0702868dd..839088d72 100644 --- a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/api/impl/ParticipantIntermediaryApiImpl.java +++ b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/api/impl/ParticipantIntermediaryApiImpl.java @@ -20,13 +20,15 @@ package org.onap.policy.clamp.controlloop.participant.intermediary.api.impl; -import java.util.Collections; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import java.util.UUID; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ClElementStatistics; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoop; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoopElement; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoopOrderedState; +import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoopState; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoops; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.Participant; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ParticipantState; @@ -85,34 +87,43 @@ public class ParticipantIntermediaryApiImpl implements ParticipantIntermediaryAp } @Override - public List<ControlLoopElement> getControlLoopElements(String name, String version) { + public Map<UUID, ControlLoopElement> getControlLoopElements(String name, String version) { List<ControlLoop> controlLoops = activator.getParticipantHandler() .getControlLoopHandler().getControlLoops().getControlLoopList(); for (ControlLoop controlLoop : controlLoops) { - if (controlLoop.getDefinition().getName().equals(name)) { + if (name.equals(controlLoop.getDefinition().getName())) { return controlLoop.getElements(); } } - return Collections.emptyList(); + return new LinkedHashMap<>(); } @Override - public ControlLoop updateControlLoopState(ToscaConceptIdentifier definition, ControlLoopOrderedState state) { - return activator.getParticipantHandler().getControlLoopHandler() - .updateControlLoopState(definition, state); + public ControlLoopElement getControlLoopElement(UUID id) { + List<ControlLoop> controlLoops = activator.getParticipantHandler() + .getControlLoopHandler().getControlLoops().getControlLoopList(); + + for (ControlLoop controlLoop : controlLoops) { + ControlLoopElement clElement = controlLoop.getElements().get(id); + if (clElement != null) { + return clElement; + } + } + return null; } @Override - public ControlLoopElement updateControlLoopElementState(UUID id, ControlLoopOrderedState state) { + public ControlLoopElement updateControlLoopElementState(UUID id, ControlLoopOrderedState currentState, + ControlLoopState newState) { return activator.getParticipantHandler().getControlLoopHandler() - .updateControlLoopElementState(id, state); + .updateControlLoopElementState(id, currentState, newState); } @Override - public void updateControlLoopElementStatistics(ClElementStatistics elementStatistics) { + public void updateControlLoopElementStatistics(UUID id, ClElementStatistics elementStatistics) { activator.getParticipantHandler().getControlLoopHandler() - .updateControlLoopElementStatistics(elementStatistics); + .updateControlLoopElementStatistics(id, elementStatistics); } @Override diff --git a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ControlLoopStateChangeListener.java b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ControlLoopStateChangeListener.java new file mode 100644 index 000000000..50b8b9cdc --- /dev/null +++ b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ControlLoopStateChangeListener.java @@ -0,0 +1,68 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2021 Nordix Foundation. + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.clamp.controlloop.participant.intermediary.comm; + +import java.io.Closeable; +import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantControlLoopStateChange; +import org.onap.policy.clamp.controlloop.participant.intermediary.handler.ParticipantHandler; +import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure; +import org.onap.policy.common.endpoints.listeners.ScoListener; +import org.onap.policy.common.utils.coder.StandardCoderObject; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Listener for Participant State Change messages sent by CLAMP. + */ +public class ControlLoopStateChangeListener extends ScoListener<ParticipantControlLoopStateChange> + implements Closeable { + private static final Logger LOGGER = LoggerFactory.getLogger(ControlLoopStateChangeListener.class); + + private final ParticipantHandler participantHandler; + + /** + * Constructs the object. + * + * @param participantHandler the handler for managing the state of the participant + */ + public ControlLoopStateChangeListener(final ParticipantHandler participantHandler) { + super(ParticipantControlLoopStateChange.class); + this.participantHandler = participantHandler; + } + + @Override + public void onTopicEvent(final CommInfrastructure infra, final String topic, final StandardCoderObject sco, + final ParticipantControlLoopStateChange controlLoopStateChangeMsg) { + LOGGER.debug("Control Loop State Change received from CLAMP - {}", controlLoopStateChangeMsg); + + if (participantHandler.canHandle(controlLoopStateChangeMsg)) { + LOGGER.debug("Message for this participant"); + participantHandler.getControlLoopHandler().handleControlLoopStateChange(controlLoopStateChangeMsg); + } else { + LOGGER.debug("Message not for this participant"); + } + } + + @Override + public void close() { + // No explicit action on this class + } +} diff --git a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ControlLoopUpdateListener.java b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ControlLoopUpdateListener.java index 2ba98891f..ab2437c1c 100644 --- a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ControlLoopUpdateListener.java +++ b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ControlLoopUpdateListener.java @@ -51,7 +51,13 @@ public class ControlLoopUpdateListener extends ScoListener<ParticipantControlLoo public void onTopicEvent(final CommInfrastructure infra, final String topic, final StandardCoderObject sco, final ParticipantControlLoopUpdate participantControlLoopUpdateMsg) { LOGGER.debug("Control Loop update received from CLAMP - {}", participantControlLoopUpdateMsg); - participantHandler.getControlLoopHandler().handleControlLoopUpdate(participantControlLoopUpdateMsg); + + if (participantHandler.canHandle(participantControlLoopUpdateMsg)) { + LOGGER.debug("Message for this participant"); + participantHandler.getControlLoopHandler().handleControlLoopUpdate(participantControlLoopUpdateMsg); + } else { + LOGGER.debug("Message not for this participant"); + } } @Override diff --git a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/MessageSender.java b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/MessageSender.java index 20490f81d..3128f1eaa 100644 --- a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/MessageSender.java +++ b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/MessageSender.java @@ -21,14 +21,22 @@ package org.onap.policy.clamp.controlloop.participant.intermediary.comm; import java.io.Closeable; +import java.time.Instant; import java.util.TimerTask; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoop; +import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoopElement; +import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoops; +import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ParticipantStatistics; import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantResponseDetails; import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantResponseStatus; import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantStatus; +import org.onap.policy.clamp.controlloop.participant.intermediary.api.ControlLoopElementListener; +import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ParticipantStatusPublisher; import org.onap.policy.clamp.controlloop.participant.intermediary.handler.ParticipantHandler; +import org.onap.policy.models.base.PfModelException; import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -86,7 +94,7 @@ public class MessageSender extends TimerTask implements Closeable { } /** - * Send a response message for this participant. + * Dispatch a response message for this participant. * * @param controlLoopId the control loop to which this message is a response * @param response the details to include in the response message @@ -95,19 +103,55 @@ public class MessageSender extends TimerTask implements Closeable { ParticipantStatus status = new ParticipantStatus(); // Participant related fields + status.setParticipantType(participantHandler.getParticipantType()); status.setParticipantId(participantHandler.getParticipantId()); status.setState(participantHandler.getState()); status.setHealthStatus(participantHandler.getHealthStatus()); // Control loop related fields + ControlLoops controlLoops = participantHandler.getControlLoopHandler().getControlLoops(); status.setControlLoopId(controlLoopId); - status.setControlLoops(participantHandler.getControlLoopHandler().getControlLoops()); + status.setControlLoops(controlLoops); status.setResponse(response); + ParticipantStatistics participantStatistics = new ParticipantStatistics(); + participantStatistics.setTimeStamp(Instant.now()); + participantStatistics.setParticipantId(participantHandler.getParticipantId()); + participantStatistics.setHealthStatus(participantHandler.getHealthStatus()); + participantStatistics.setState(participantHandler.getState()); + status.setParticipantStatistics(participantStatistics); + + for (ControlLoopElementListener clElementListener : + participantHandler.getControlLoopHandler().getListeners()) { + updateClElementStatistics(controlLoops, clElementListener); + } + + status.setControlLoops(controlLoops); + publisher.send(status); } /** + * Update ControlLoopElement statistics. The control loop elements listening will be + * notified to retrieve statistics from respective controlloop elements, and controlloopelements + * data on the handler will be updated. + * + * @param controlLoops the control loops + * @param clElementListener control loop element listener + */ + public void updateClElementStatistics(ControlLoops controlLoops, ControlLoopElementListener clElementListener) { + for (ControlLoop controlLoop : controlLoops.getControlLoopList()) { + for (ControlLoopElement element : controlLoop.getElements().values()) { + try { + clElementListener.handleStatistics(element.getId()); + } catch (PfModelException e) { + LOGGER.debug("Getting statistics for Control loop element failed"); + } + } + } + } + + /** * Makes a new timer pool. * * @return a new timer pool diff --git a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantHealthCheckListener.java b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantHealthCheckListener.java new file mode 100644 index 000000000..e0e6be329 --- /dev/null +++ b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantHealthCheckListener.java @@ -0,0 +1,69 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2021 Nordix Foundation. + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.clamp.controlloop.participant.intermediary.comm; + +import java.io.Closeable; +import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantHealthCheck; +import org.onap.policy.clamp.controlloop.participant.intermediary.handler.ParticipantHandler; +import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure; +import org.onap.policy.common.endpoints.listeners.ScoListener; +import org.onap.policy.common.utils.coder.StandardCoderObject; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Listener for Participant health status messages sent by CLAMP. + */ +public class ParticipantHealthCheckListener extends ScoListener<ParticipantHealthCheck> implements Closeable { + private static final Logger LOGGER = LoggerFactory.getLogger(ParticipantHealthCheckListener.class); + + private final ParticipantHandler participantHandler; + + /** + * Constructs the object. + * + * @param participantHandler the handler for managing the state and health of the participant + */ + public ParticipantHealthCheckListener(final ParticipantHandler participantHandler) { + super(ParticipantHealthCheck.class); + this.participantHandler = participantHandler; + } + + @Override + public void onTopicEvent(final CommInfrastructure infra, final String topic, final StandardCoderObject sco, + final ParticipantHealthCheck participantHealthCheckMsg) { + LOGGER.debug("Participant Health Check message received from CLAMP - {}", participantHealthCheckMsg); + + + if (participantHandler.canHandle(participantHealthCheckMsg)) { + LOGGER.debug("Message for this participant"); + participantHandler.handleParticipantHealthCheck(participantHealthCheckMsg); + } else { + LOGGER.debug("Message not for this participant"); + } + + } + + @Override + public void close() { + // No explicit action on this class + } +} diff --git a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantStateChangeListener.java b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantStateChangeListener.java new file mode 100644 index 000000000..c1a8b5b4a --- /dev/null +++ b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantStateChangeListener.java @@ -0,0 +1,68 @@ +/*- + * ============LICENSE_START======================================================= + * Copyright (C) 2021 Nordix Foundation. + * ================================================================================ + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + * SPDX-License-Identifier: Apache-2.0 + * ============LICENSE_END========================================================= + */ + +package org.onap.policy.clamp.controlloop.participant.intermediary.comm; + +import java.io.Closeable; +import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantStateChange; +import org.onap.policy.clamp.controlloop.participant.intermediary.handler.ParticipantHandler; +import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure; +import org.onap.policy.common.endpoints.listeners.ScoListener; +import org.onap.policy.common.utils.coder.StandardCoderObject; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Listener for Participant State Change messages sent by CLAMP. + * + */ +public class ParticipantStateChangeListener extends ScoListener<ParticipantStateChange> implements Closeable { + private static final Logger LOGGER = LoggerFactory.getLogger(ParticipantStateChangeListener.class); + + private final ParticipantHandler participantHandler; + + /** + * Constructs the object. + * + * @param participantHandler the handler for managing the state of the participant + */ + public ParticipantStateChangeListener(final ParticipantHandler participantHandler) { + super(ParticipantStateChange.class); + this.participantHandler = participantHandler; + } + + @Override + public void onTopicEvent(final CommInfrastructure infra, final String topic, final StandardCoderObject sco, + final ParticipantStateChange participantStateChangeMsg) { + LOGGER.debug("Participant State Change received from CLAMP - {}", participantStateChangeMsg); + + if (participantHandler.canHandle(participantStateChangeMsg)) { + LOGGER.debug("Message for this participant"); + participantHandler.handleParticipantStateChange(participantStateChangeMsg); + } else { + LOGGER.debug("Message not for this participant"); + } + } + + @Override + public void close() { + // No explicit action on this class + } +} diff --git a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantStatusPublisher.java b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantStatusPublisher.java index e909327cd..bc53b4e9d 100644 --- a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantStatusPublisher.java +++ b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantStatusPublisher.java @@ -30,6 +30,7 @@ import org.slf4j.LoggerFactory; /** * This class is used to send Participant Status messages to clamp using TopicSinkClient. + * */ public class ParticipantStatusPublisher implements Closeable { private static final Logger LOGGER = LoggerFactory.getLogger(ParticipantStatusPublisher.class); diff --git a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/ControlLoopHandler.java b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/ControlLoopHandler.java index 6f8bfffc3..cc2a66b58 100644 --- a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/ControlLoopHandler.java +++ b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/ControlLoopHandler.java @@ -52,27 +52,29 @@ import org.slf4j.LoggerFactory; public class ControlLoopHandler implements Closeable { private static final Logger LOGGER = LoggerFactory.getLogger(ControlLoopHandler.class); + private ToscaConceptIdentifier participantType = null; private ToscaConceptIdentifier participantId = null; - private MessageSender sender = null; + private MessageSender messageSender = null; private final Map<ToscaConceptIdentifier, ControlLoop> controlLoopMap = new LinkedHashMap<>(); + private final Map<UUID, ControlLoopElement> elementsOnThisParticipant = new LinkedHashMap<>(); @Getter private List<ControlLoopElementListener> listeners = new ArrayList<>(); - public ControlLoopHandler() { - } + public ControlLoopHandler() {} /** - * Constructor, set the participant ID and sender. + * Constructor, set the participant ID and messageSender. * * @param parameters the parameters of the participant - * @param sender the sender for sending responses to messages + * @param messageSender the messageSender for sending responses to messages */ - public ControlLoopHandler(ParticipantIntermediaryParameters parameters, MessageSender sender) { + public ControlLoopHandler(ParticipantIntermediaryParameters parameters, MessageSender messageSender) { + this.participantType = parameters.getParticipantType(); this.participantId = parameters.getParticipantId(); - this.sender = sender; + this.messageSender = messageSender; } @Override @@ -83,15 +85,17 @@ public class ControlLoopHandler implements Closeable { public void registerControlLoopElementListener(ControlLoopElementListener listener) { listeners.add(listener); } - + /** * Handle a control loop element state change message. * * @param id controlloop element id - * @param state the updated state + * @param orderedState the current state + * @param newState the ordered state * @return controlLoopElement the updated controlloop element */ - public ControlLoopElement updateControlLoopElementState(UUID id, ControlLoopOrderedState state) { + public ControlLoopElement updateControlLoopElementState(UUID id, ControlLoopOrderedState orderedState, + ControlLoopState newState) { if (id == null) { return null; @@ -99,42 +103,32 @@ public class ControlLoopHandler implements Closeable { ControlLoopElement clElement = elementsOnThisParticipant.get(id); if (clElement != null) { - clElement.setOrderedState(state); - LOGGER.debug("Control loop element {} ordered state changed to {}", id, state); + clElement.setOrderedState(orderedState); + clElement.setState(newState); + LOGGER.debug("Control loop element {} state changed to {}", id, newState); ParticipantResponseDetails response = new ParticipantResponseDetails(); - sender.sendResponse(response); - return elementsOnThisParticipant.get(id); + response.setResponseStatus(ParticipantResponseStatus.SUCCESS); + response.setResponseMessage("ControlLoopElement state changed to {} " + newState); + messageSender.sendResponse(response); + return clElement; } return null; } - public void updateControlLoopElementStatistics(ClElementStatistics elementStatistics) { - // TODO Handle statistics coming from a participant implementation - } - /** - * Handle a control loop state change message. + * Handle a control loop element statistics. * - * @param definition controlloop id - * @param state the updated state - * @return controlLoop the updated controlloop + * @param id controlloop element id + * @param elementStatistics control loop element Statistics */ - public ControlLoop updateControlLoopState(ToscaConceptIdentifier definition, ControlLoopOrderedState state) { - if (definition == null) { - return null; - } - - ControlLoop controlLoop = controlLoopMap.get(definition); - if (controlLoop == null) { - LOGGER.debug("Control loop {} does not use this participant", definition.getName()); - return null; + public void updateControlLoopElementStatistics(UUID id, ClElementStatistics elementStatistics) { + ControlLoopElement clElement = elementsOnThisParticipant.get(id); + if (clElement != null) { + elementStatistics.setParticipantId(participantId); + elementStatistics.setId(id); + clElement.setClElementStatistics(elementStatistics); } - - ParticipantResponseDetails response = new ParticipantResponseDetails(); - handleState(controlLoop, response, state); - sender.sendResponse(response); - return controlLoop; } /** @@ -156,7 +150,7 @@ public class ControlLoopHandler implements Closeable { ParticipantResponseDetails response = new ParticipantResponseDetails(stateChangeMsg); handleState(controlLoop, response, stateChangeMsg.getOrderedState()); - sender.sendResponse(response); + messageSender.sendResponse(response); } /** @@ -164,19 +158,19 @@ public class ControlLoopHandler implements Closeable { * * @param controlLoop participant response * @param response participant response - * @param state controlloop ordered state + * @param orderedState controlloop ordered state */ private void handleState(final ControlLoop controlLoop, final ParticipantResponseDetails response, - ControlLoopOrderedState state) { - switch (state) { + ControlLoopOrderedState orderedState) { + switch (orderedState) { case UNINITIALISED: - handleUninitialisedState(controlLoop, response); + handleUninitialisedState(controlLoop, orderedState, response); break; case PASSIVE: - handlePassiveState(controlLoop, response); + handlePassiveState(controlLoop, orderedState, response); break; case RUNNING: - handleRunningState(controlLoop, response); + handleRunningState(controlLoop, orderedState, response); break; default: LOGGER.debug("StateChange message has no state, state is null {}", controlLoop.getDefinition()); @@ -190,7 +184,8 @@ public class ControlLoopHandler implements Closeable { * @param updateMsg the update message */ public void handleControlLoopUpdate(ParticipantControlLoopUpdate updateMsg) { - if (!updateMsg.appliesTo(participantId)) { + + if (!updateMsg.appliesTo(participantType, participantId)) { return; } @@ -205,22 +200,25 @@ public class ControlLoopHandler implements Closeable { response.setResponseMessage("Control loop " + updateMsg.getControlLoopId() + " already defined on participant " + participantId); - sender.sendResponse(response); + messageSender.sendResponse(response); return; } controlLoop = updateMsg.getControlLoop(); - controlLoop.getElements().removeIf(element -> participantId.equals(element.getParticipantId())); + controlLoop.getElements().values().removeIf(element -> !participantType.equals(element.getParticipantType())); controlLoopMap.put(updateMsg.getControlLoopId(), controlLoop); - for (ControlLoopElement element : updateMsg.getControlLoop().getElements()) { + for (ControlLoopElement element : updateMsg.getControlLoop().getElements().values()) { element.setState(element.getOrderedState().asState()); + element.setParticipantId(participantId); elementsOnThisParticipant.put(element.getId(), element); } for (ControlLoopElementListener clElementListener : listeners) { try { - clElementListener.controlLoopElementUpdate(null, updateMsg.getControlLoopDefinition()); + for (ControlLoopElement element : updateMsg.getControlLoop().getElements().values()) { + clElementListener.controlLoopElementUpdate(element, updateMsg.getControlLoopDefinition()); + } } catch (PfModelException e) { LOGGER.debug("Control loop element update failed {}", updateMsg.getControlLoopId()); } @@ -230,64 +228,88 @@ public class ControlLoopHandler implements Closeable { response.setResponseMessage( "Control loop " + updateMsg.getControlLoopId() + " defined on participant " + participantId); - sender.sendResponse(response); + messageSender.sendResponse(response); } /** * Method to handle when the new state from participant is UNINITIALISED state. * * @param controlLoop participant response + * @param orderedState orderedState * @param response participant response */ - private void handleUninitialisedState(final ControlLoop controlLoop, final ParticipantResponseDetails response) { - handleStateChange(controlLoop, ControlLoopState.UNINITIALISED, response); + private void handleUninitialisedState(final ControlLoop controlLoop, final ControlLoopOrderedState orderedState, + final ParticipantResponseDetails response) { + handleStateChange(controlLoop, orderedState, ControlLoopState.UNINITIALISED, response); controlLoopMap.remove(controlLoop.getKey().asIdentifier()); + + for (ControlLoopElementListener clElementListener : listeners) { + try { + for (ControlLoopElement element : controlLoop.getElements().values()) { + clElementListener.controlLoopElementStateChange(element.getId(), element.getState(), + orderedState); + } + } catch (PfModelException e) { + LOGGER.debug("Control loop element update failed {}", controlLoop.getDefinition()); + } + } } /** * Method to handle when the new state from participant is PASSIVE state. * * @param controlLoop participant response + * @param orderedState orderedState * @param response participant response */ - private void handlePassiveState(final ControlLoop controlLoop, final ParticipantResponseDetails response) { - handleStateChange(controlLoop, ControlLoopState.PASSIVE, response); + private void handlePassiveState(final ControlLoop controlLoop, final ControlLoopOrderedState orderedState, + final ParticipantResponseDetails response) { + handleStateChange(controlLoop, orderedState, ControlLoopState.PASSIVE, response); } /** * Method to handle when the new state from participant is RUNNING state. * * @param controlLoop participant response + * @param orderedState orderedState * @param response participant response */ - private void handleRunningState(final ControlLoop controlLoop, final ParticipantResponseDetails response) { - handleStateChange(controlLoop, ControlLoopState.RUNNING, response); + private void handleRunningState(final ControlLoop controlLoop, final ControlLoopOrderedState orderedState, + final ParticipantResponseDetails response) { + handleStateChange(controlLoop, orderedState, ControlLoopState.RUNNING, response); } - + /** * Method to update the state of control loop elements. * * @param controlLoop participant status in memory + * @param orderedState orderedState * @param state new state of the control loop elements */ - private void handleStateChange(ControlLoop controlLoop, ControlLoopState newState, - ParticipantResponseDetails response) { + private void handleStateChange(ControlLoop controlLoop, final ControlLoopOrderedState orderedState, + ControlLoopState newState, ParticipantResponseDetails response) { - if (newState.equals(controlLoop.getState())) { + if (orderedState.equals(controlLoop.getOrderedState())) { response.setResponseStatus(ParticipantResponseStatus.SUCCESS); - response.setResponseMessage("Control loop is already in state " + newState); + response.setResponseMessage("Control loop is already in state " + orderedState); return; } - if (!CollectionUtils.isEmpty(controlLoop.getElements())) { - controlLoop.getElements().forEach(element -> element.setState(newState)); + if (!CollectionUtils.isEmpty(controlLoop.getElements().values())) { + controlLoop.getElements().values().forEach(element -> { + element.setState(newState); + element.setOrderedState(orderedState); + } + ); } response.setResponseStatus(ParticipantResponseStatus.SUCCESS); - response.setResponseMessage("ControlLoop state changed from " + controlLoop.getState() + " to " + newState); - controlLoop.setState(newState); + response.setResponseMessage("ControlLoop state changed from " + controlLoop.getOrderedState() + + " to " + orderedState); + controlLoop.setOrderedState(orderedState); } + /** * Get control loops as a {@link ConrolLoops} class. * diff --git a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/IntermediaryActivator.java b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/IntermediaryActivator.java index dd0cf30a8..be2fa1a30 100644 --- a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/IntermediaryActivator.java +++ b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/IntermediaryActivator.java @@ -26,13 +26,22 @@ import javax.ws.rs.core.Response.Status; import lombok.Getter; import lombok.experimental.Delegate; import org.onap.policy.clamp.controlloop.common.exception.ControlLoopRuntimeException; +import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantControlLoopStateChange; +import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantControlLoopUpdate; +import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantHealthCheck; +import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantMessageType; +import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantStateChange; +import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ControlLoopStateChangeListener; import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ControlLoopUpdateListener; +import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ParticipantHealthCheckListener; +import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ParticipantStateChangeListener; import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ParticipantStatusPublisher; import org.onap.policy.clamp.controlloop.participant.intermediary.parameters.ParticipantIntermediaryParameters; import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager; import org.onap.policy.common.endpoints.event.comm.TopicSink; import org.onap.policy.common.endpoints.event.comm.TopicSource; import org.onap.policy.common.endpoints.listeners.MessageTypeDispatcher; +import org.onap.policy.common.endpoints.listeners.ScoListener; import org.onap.policy.common.utils.services.ServiceManagerContainer; /** @@ -80,6 +89,9 @@ public class IntermediaryActivator extends ServiceManagerContainer { // @formatter:off final AtomicReference<ParticipantStatusPublisher> statusPublisher = new AtomicReference<>(); + final AtomicReference<ParticipantStateChangeListener> participantStateChangeListener = new AtomicReference<>(); + final AtomicReference<ParticipantHealthCheckListener> participantHealthCheckListener = new AtomicReference<>(); + final AtomicReference<ControlLoopStateChangeListener> controlLoopStateChangeListener = new AtomicReference<>(); final AtomicReference<ControlLoopUpdateListener> controlLoopUpdateListener = new AtomicReference<>(); addAction("Topic endpoint management", @@ -94,6 +106,18 @@ public class IntermediaryActivator extends ServiceManagerContainer { () -> participantHandler.set(new ParticipantHandler(parameters, statusPublisher.get())), () -> participantHandler.get().close()); + addAction("Participant State Change Listener", + () -> participantStateChangeListener.set(new ParticipantStateChangeListener(participantHandler.get())), + () -> participantStateChangeListener.get().close()); + + addAction("Participant Health Check Listener", + () -> participantHealthCheckListener.set(new ParticipantHealthCheckListener(participantHandler.get())), + () -> participantHealthCheckListener.get().close()); + + addAction("Control Loop State Change Listener", + () -> controlLoopStateChangeListener.set(new ControlLoopStateChangeListener(participantHandler.get())), + () -> controlLoopStateChangeListener.get().close()); + addAction("Control Loop Update Listener", () -> controlLoopUpdateListener.set(new ControlLoopUpdateListener(participantHandler.get())), () -> controlLoopUpdateListener.get().close()); @@ -106,6 +130,18 @@ public class IntermediaryActivator extends ServiceManagerContainer { * Registers the dispatcher with the topic source(s). */ private void registerMsgDispatcher() { + msgDispatcher.register(ParticipantMessageType.PARTICIPANT_STATE_CHANGE.name(), + (ScoListener<ParticipantStateChange>) new ParticipantStateChangeListener( + participantHandler.get())); + msgDispatcher.register(ParticipantMessageType.PARTICIPANT_HEALTH_CHECK.name(), + (ScoListener<ParticipantHealthCheck>) new ParticipantHealthCheckListener( + participantHandler.get())); + msgDispatcher.register(ParticipantMessageType.PARTICIPANT_CONTROL_LOOP_STATE_CHANGE.name(), + (ScoListener<ParticipantControlLoopStateChange>) new ControlLoopStateChangeListener( + participantHandler.get())); + msgDispatcher.register(ParticipantMessageType.PARTICIPANT_CONTROL_LOOP_UPDATE.name(), + (ScoListener<ParticipantControlLoopUpdate>) new ControlLoopUpdateListener( + participantHandler.get())); for (final TopicSource source : topicSources) { source.register(msgDispatcher); } diff --git a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/ParticipantHandler.java b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/ParticipantHandler.java index 1150471ae..980ab6ec1 100644 --- a/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/ParticipantHandler.java +++ b/tosca-controlloop/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/ParticipantHandler.java @@ -27,6 +27,9 @@ import lombok.Setter; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.Participant; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ParticipantHealthStatus; import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ParticipantState; +import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ParticipantStatistics; +import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantHealthCheck; +import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantMessage; import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantResponseDetails; import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantResponseStatus; import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantStateChange; @@ -44,9 +47,11 @@ import org.slf4j.LoggerFactory; public class ParticipantHandler implements Closeable { private static final Logger LOGGER = LoggerFactory.getLogger(ParticipantHandler.class); + private final ToscaConceptIdentifier participantType; private final ToscaConceptIdentifier participantId; private final MessageSender sender; private final ControlLoopHandler controlLoopHandler; + private final ParticipantStatistics participantStatistics; @Setter private ParticipantState state = ParticipantState.UNKNOWN; @@ -61,9 +66,11 @@ public class ParticipantHandler implements Closeable { * @param publisher the publisher for sending responses to messages */ public ParticipantHandler(ParticipantIntermediaryParameters parameters, ParticipantStatusPublisher publisher) { + this.participantType = parameters.getParticipantType(); this.participantId = parameters.getParticipantId(); this.sender = new MessageSender(this, publisher, parameters.getReportingTimeInterval()); this.controlLoopHandler = new ControlLoopHandler(parameters, sender); + this.participantStatistics = new ParticipantStatistics(); } @Override @@ -79,7 +86,7 @@ public class ParticipantHandler implements Closeable { */ public void handleParticipantStateChange(final ParticipantStateChange stateChangeMsg) { - if (!stateChangeMsg.appliesTo(participantId)) { + if (!stateChangeMsg.appliesTo(participantType, participantId)) { return; } @@ -112,6 +119,20 @@ public class ParticipantHandler implements Closeable { sender.sendResponse(response); } + + /** + * Method which handles a participant health check event from clamp. + * + * @param healthCheckMsg participant health check message + */ + public void handleParticipantHealthCheck(final ParticipantHealthCheck healthCheckMsg) { + ParticipantResponseDetails response = new ParticipantResponseDetails(healthCheckMsg); + response.setResponseStatus(ParticipantResponseStatus.SUCCESS); + response.setResponseMessage(healthStatus.toString()); + + sender.sendResponse(response); + } + /** * Method to handle when the new state from participant is active. * @@ -201,4 +222,14 @@ public class ParticipantHandler implements Closeable { } return null; } + + /** + * Check if a participant message applies to this participant handler. + * + * @param partipantMsg the message to check + * @return true if it applies, false otherwise + */ + public boolean canHandle(ParticipantMessage partipantMsg) { + return partipantMsg.appliesTo(participantType, participantId); + } } |