diff options
Diffstat (limited to 'participant/participant-intermediary')
10 files changed, 61 insertions, 160 deletions
diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ControlLoopStateChangeListener.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ControlLoopStateChangeListener.java index e46c6db1b..0b9110bd0 100644 --- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ControlLoopStateChangeListener.java +++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ControlLoopStateChangeListener.java @@ -20,7 +20,7 @@ package org.onap.policy.clamp.controlloop.participant.intermediary.comm; -import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantControlLoopStateChange; +import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ControlLoopStateChange; import org.onap.policy.clamp.controlloop.participant.intermediary.handler.ParticipantHandler; import org.springframework.stereotype.Component; @@ -28,7 +28,7 @@ import org.springframework.stereotype.Component; * Listener for Participant State Change messages sent by CLAMP. */ @Component -public class ControlLoopStateChangeListener extends ParticipantListener<ParticipantControlLoopStateChange> { +public class ControlLoopStateChangeListener extends ParticipantListener<ControlLoopStateChange> { /** * Constructs the object. @@ -36,7 +36,7 @@ public class ControlLoopStateChangeListener extends ParticipantListener<Particip * @param participantHandler the handler for managing the state of the participant */ public ControlLoopStateChangeListener(final ParticipantHandler participantHandler) { - super(ParticipantControlLoopStateChange.class, participantHandler, + super(ControlLoopStateChange.class, participantHandler, participantHandler::handleControlLoopStateChange); } } diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ControlLoopUpdateListener.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ControlLoopUpdateListener.java index d15643e0f..56bc1fd9a 100644 --- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ControlLoopUpdateListener.java +++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ControlLoopUpdateListener.java @@ -20,7 +20,7 @@ package org.onap.policy.clamp.controlloop.participant.intermediary.comm; -import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantControlLoopUpdate; +import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ControlLoopUpdate; import org.onap.policy.clamp.controlloop.participant.intermediary.handler.ParticipantHandler; import org.springframework.stereotype.Component; @@ -28,7 +28,7 @@ import org.springframework.stereotype.Component; * Listener for Control Loop Update messages sent by CLAMP. */ @Component -public class ControlLoopUpdateListener extends ParticipantListener<ParticipantControlLoopUpdate> { +public class ControlLoopUpdateListener extends ParticipantListener<ControlLoopUpdate> { /** * Constructs the object. @@ -36,6 +36,6 @@ public class ControlLoopUpdateListener extends ParticipantListener<ParticipantCo * @param participantHandler the handler for managing the state of the participant */ public ControlLoopUpdateListener(final ParticipantHandler participantHandler) { - super(ParticipantControlLoopUpdate.class, participantHandler, participantHandler::handleControlLoopUpdate); + super(ControlLoopUpdate.class, participantHandler, participantHandler::handleControlLoopUpdate); } } diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/MessageSender.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/MessageSender.java index 1bfce1374..1741d9514 100644 --- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/MessageSender.java +++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/MessageSender.java @@ -67,18 +67,13 @@ public class MessageSender extends TimerTask implements Closeable { // Kick off the timer timerPool = makeTimerPool(); - timerPool.scheduleAtFixedRate(this, 0, interval, TimeUnit.SECONDS); + timerPool.scheduleAtFixedRate(this, 0, interval, TimeUnit.MILLISECONDS); } @Override public void run() { LOGGER.debug("Sent heartbeat to CLAMP"); - - var response = new ParticipantResponseDetails(); - - response.setResponseTo(null); - response.setResponseStatus(ParticipantResponseStatus.PERIODIC); - response.setResponseMessage("Periodic response from participant"); + this.sendHeartbeat(); } @Override @@ -161,6 +156,13 @@ public class MessageSender extends TimerTask implements Closeable { } /** + * Dispatch a heartbeat for this participant. + */ + public void sendHeartbeat() { + publisher.sendHeartbeat(participantHandler.makeHeartbeat()); + } + + /** * 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. diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantMessagePublisher.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantMessagePublisher.java index 9e1b84620..051f00095 100644 --- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantMessagePublisher.java +++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantMessagePublisher.java @@ -90,4 +90,14 @@ public class ParticipantMessagePublisher { topicSinkClient.send(participantUpdateAck); LOGGER.debug("Sent Participant Update Ack message to CLAMP - {}", participantUpdateAck); } + + /** + * Method to send Participant heartbeat to clamp on demand. + * + * @param participantStatus the Participant Status + */ + public void sendHeartbeat(final ParticipantStatus participantStatus) { + topicSinkClient.send(participantStatus); + LOGGER.debug("Sent Participant heartbeat to CLAMP - {}", participantStatus); + } } diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantStateChangeListener.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantStateChangeListener.java deleted file mode 100644 index ec6548a7c..000000000 --- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/comm/ParticipantStateChangeListener.java +++ /dev/null @@ -1,42 +0,0 @@ -/*- - * ============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 org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantStateChange; -import org.onap.policy.clamp.controlloop.participant.intermediary.handler.ParticipantHandler; -import org.springframework.stereotype.Component; - -/** - * Listener for Participant State Change messages sent by CLAMP. - * - */ -@Component -public class ParticipantStateChangeListener extends ParticipantListener<ParticipantStateChange> { - - /** - * Constructs the object. - * - * @param participantHandler the handler for managing the state of the participant - */ - public ParticipantStateChangeListener(final ParticipantHandler participantHandler) { - super(ParticipantStateChange.class, participantHandler, participantHandler::handleParticipantStateChange); - } -} diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/ControlLoopHandler.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/ControlLoopHandler.java index 50048ffc2..c9da1279d 100644 --- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/ControlLoopHandler.java +++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/ControlLoopHandler.java @@ -31,11 +31,12 @@ import org.apache.commons.collections4.CollectionUtils; 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.ControlLoopElementDefinition; 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.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.ControlLoopStateChange; +import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ControlLoopUpdate; 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.participant.intermediary.api.ControlLoopElementListener; @@ -130,7 +131,7 @@ public class ControlLoopHandler { * * @param stateChangeMsg the state change message */ - public void handleControlLoopStateChange(ParticipantControlLoopStateChange stateChangeMsg) { + public void handleControlLoopStateChange(ControlLoopStateChange stateChangeMsg) { if (stateChangeMsg.getControlLoopId() == null) { return; } @@ -177,7 +178,8 @@ public class ControlLoopHandler { * * @param updateMsg the update message */ - public void handleControlLoopUpdate(ParticipantControlLoopUpdate updateMsg) { + public void handleControlLoopUpdate(ControlLoopUpdate updateMsg, + Map<UUID, ControlLoopElementDefinition> clElementDefinitions) { if (!updateMsg.appliesTo(participantType, participantId)) { return; @@ -211,7 +213,8 @@ public class ControlLoopHandler { for (ControlLoopElementListener clElementListener : listeners) { try { for (ControlLoopElement element : updateMsg.getControlLoop().getElements().values()) { - clElementListener.controlLoopElementUpdate(element, updateMsg.getControlLoopDefinition()); + clElementListener.controlLoopElementUpdate(element, + clElementDefinitions.get(element.getId()).getControlLoopElementToscaServiceTemplate()); } } catch (PfModelException e) { LOGGER.debug("Control loop element update failed {}", updateMsg.getControlLoopId()); diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/IntermediaryActivator.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/IntermediaryActivator.java index 0aa536746..f846b2d75 100644 --- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/IntermediaryActivator.java +++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/IntermediaryActivator.java @@ -30,7 +30,6 @@ import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ControlLo import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ParticipantDeregisterAckListener; import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ParticipantHealthCheckListener; import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ParticipantRegisterAckListener; -import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ParticipantStateChangeListener; import org.onap.policy.clamp.controlloop.participant.intermediary.comm.ParticipantUpdateListener; import org.onap.policy.clamp.controlloop.participant.intermediary.parameters.ParticipantParameters; import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager; @@ -120,16 +119,13 @@ public class IntermediaryActivator extends ServiceManagerContainer implements Cl private void registerMsgDispatcher() { MessageTypeDispatcher msgDispatcher = applicationContext.getBean(MessageTypeDispatcher.class); - msgDispatcher.register(ParticipantMessageType.PARTICIPANT_STATE_CHANGE.name(), - applicationContext.getBean(ParticipantStateChangeListener.class)); - msgDispatcher.register(ParticipantMessageType.PARTICIPANT_HEALTH_CHECK.name(), applicationContext.getBean(ParticipantHealthCheckListener.class)); - msgDispatcher.register(ParticipantMessageType.PARTICIPANT_CONTROL_LOOP_STATE_CHANGE.name(), + msgDispatcher.register(ParticipantMessageType.CONTROL_LOOP_STATE_CHANGE.name(), applicationContext.getBean(ControlLoopStateChangeListener.class)); - msgDispatcher.register(ParticipantMessageType.PARTICIPANT_CONTROL_LOOP_UPDATE.name(), + msgDispatcher.register(ParticipantMessageType.CONTROL_LOOP_UPDATE.name(), applicationContext.getBean(ControlLoopUpdateListener.class)); msgDispatcher.register(ParticipantMessageType.PARTICIPANT_REGISTER_ACK.name(), diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/ParticipantHandler.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/ParticipantHandler.java index a8913c1f0..9daff7239 100644 --- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/ParticipantHandler.java +++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/handler/ParticipantHandler.java @@ -32,8 +32,8 @@ 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.ParticipantControlLoopStateChange; -import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantControlLoopUpdate; +import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ControlLoopStateChange; +import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ControlLoopUpdate; import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantDeregister; import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantDeregisterAck; import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantHealthCheck; @@ -42,7 +42,7 @@ import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.Parti import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantRegisterAck; 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; +import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantStatus; import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantUpdate; import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantUpdateAck; import org.onap.policy.clamp.controlloop.participant.intermediary.comm.MessageSender; @@ -85,7 +85,8 @@ public class ParticipantHandler implements Closeable { this.participantType = parameters.getIntermediaryParameters().getParticipantType(); this.participantId = parameters.getIntermediaryParameters().getParticipantId(); this.sender = - new MessageSender(this, publisher, parameters.getIntermediaryParameters().getReportingTimeInterval()); + new MessageSender(this, publisher, + parameters.getIntermediaryParameters().getReportingTimeIntervalMs()); this.controlLoopHandler = new ControlLoopHandler(parameters.getIntermediaryParameters(), sender); this.participantStatistics = new ParticipantStatistics(); } @@ -96,46 +97,6 @@ public class ParticipantHandler implements Closeable { } /** - * Method which handles a participant state change event from clamp. - * - * @param stateChangeMsg participant state change message - */ - public void handleParticipantStateChange(final ParticipantStateChange stateChangeMsg) { - - if (!stateChangeMsg.appliesTo(participantType, participantId)) { - return; - } - - var response = new ParticipantResponseDetails(stateChangeMsg); - - switch (stateChangeMsg.getState()) { - case PASSIVE: - handlePassiveState(response); - break; - case ACTIVE: - handleActiveState(response); - break; - case SAFE: - handleSafeState(response); - break; - case TEST: - handleTestState(response); - break; - case TERMINATED: - handleTerminatedState(response); - break; - default: - LOGGER.debug("StateChange message has no state, state is null {}", stateChangeMsg.getParticipantId()); - response.setResponseStatus(ParticipantResponseStatus.FAIL); - response.setResponseMessage( - "StateChange message has invalid state for participantId " + stateChangeMsg.getParticipantId()); - break; - } - - sender.sendResponse(response); - } - - /** * Method which handles a participant health check event from clamp. * * @param healthCheckMsg participant health check message @@ -153,8 +114,8 @@ public class ParticipantHandler implements Closeable { * * @param updateMsg the update message */ - public void handleControlLoopUpdate(ParticipantControlLoopUpdate updateMsg) { - controlLoopHandler.handleControlLoopUpdate(updateMsg); + public void handleControlLoopUpdate(ControlLoopUpdate updateMsg) { + controlLoopHandler.handleControlLoopUpdate(updateMsg, clElementDefsOnThisParticipant); } /** @@ -162,55 +123,10 @@ public class ParticipantHandler implements Closeable { * * @param stateChangeMsg the state change message */ - public void handleControlLoopStateChange(ParticipantControlLoopStateChange stateChangeMsg) { + public void handleControlLoopStateChange(ControlLoopStateChange stateChangeMsg) { controlLoopHandler.handleControlLoopStateChange(stateChangeMsg); } - /** - * Method to handle when the new state from participant is active. - * - * @param response participant response - */ - private void handleActiveState(final ParticipantResponseDetails response) { - handleStateChange(ParticipantState.ACTIVE, response); - } - - /** - * Method to handle when the new state from participant is passive. - * - * @param response participant response - */ - private void handlePassiveState(final ParticipantResponseDetails response) { - handleStateChange(ParticipantState.PASSIVE, response); - } - - /** - * Method to handle when the new state from participant is safe. - * - * @param response participant response - */ - private void handleSafeState(final ParticipantResponseDetails response) { - handleStateChange(ParticipantState.SAFE, response); - } - - /** - * Method to handle when the new state from participant is TEST. - * - * @param response participant response - */ - private void handleTestState(final ParticipantResponseDetails response) { - handleStateChange(ParticipantState.TEST, response); - } - - /** - * Method to handle when the new state from participant is Terminated. - * - * @param response participant response - */ - private void handleTerminatedState(final ParticipantResponseDetails response) { - handleStateChange(ParticipantState.TERMINATED, response); - } - private void handleStateChange(ParticipantState newParticipantState, ParticipantResponseDetails response) { if (state.equals(newParticipantState)) { response.setResponseStatus(ParticipantResponseStatus.SUCCESS); @@ -344,4 +260,17 @@ public class ParticipantHandler implements Closeable { sender.sendParticipantUpdateAck(participantUpdateAck); } + + /** + * Method to send heartbeat to controlloop runtime. + */ + public ParticipantStatus makeHeartbeat() { + ParticipantStatus heartbeat = new ParticipantStatus(); + heartbeat.setParticipantId(participantId); + heartbeat.setParticipantStatistics(participantStatistics); + heartbeat.setParticipantType(participantType); + heartbeat.setHealthStatus(healthStatus); + heartbeat.setMessage("Participant heartbeat message sent from -> " + participantId.getName()); + return heartbeat; + } } diff --git a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/parameters/ParticipantIntermediaryParameters.java b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/parameters/ParticipantIntermediaryParameters.java index 5627abfe6..fdc451e6b 100644 --- a/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/parameters/ParticipantIntermediaryParameters.java +++ b/participant/participant-intermediary/src/main/java/org/onap/policy/clamp/controlloop/participant/intermediary/parameters/ParticipantIntermediaryParameters.java @@ -23,6 +23,7 @@ package org.onap.policy.clamp.controlloop.participant.intermediary.parameters; import javax.validation.Valid; import javax.validation.constraints.NotBlank; import javax.validation.constraints.NotNull; +import javax.validation.constraints.Positive; import lombok.Getter; import lombok.Setter; import org.onap.policy.common.endpoints.parameters.TopicParameterGroup; @@ -50,7 +51,9 @@ public class ParticipantIntermediaryParameters { private ToscaConceptIdentifier participantType; // The time interval for periodic reporting of status to the CLAMP control loop server - private long reportingTimeInterval; + @Valid + @Positive + private long reportingTimeIntervalMs; @NotNull @ParameterGroupConstraint diff --git a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/controlloop/participant/intermediary/main/parameters/CommonTestData.java b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/controlloop/participant/intermediary/main/parameters/CommonTestData.java index 93ba15846..9353cde81 100644 --- a/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/controlloop/participant/intermediary/main/parameters/CommonTestData.java +++ b/participant/participant-intermediary/src/test/java/org/onap/policy/clamp/controlloop/participant/intermediary/main/parameters/CommonTestData.java @@ -68,7 +68,7 @@ public class CommonTestData { map.put("participantId", getParticipantId()); map.put("description", DESCRIPTION); map.put("participantType", getParticipantId()); - map.put("reportingTimeInterval", TIME_INTERVAL); + map.put("reportingTimeIntervalMs", TIME_INTERVAL); map.put("clampControlLoopTopics", getTopicParametersMap(false)); return map; |