summaryrefslogtreecommitdiffstats
path: root/runtime-controlloop/src/main/java/org/onap
diff options
context:
space:
mode:
authorFrancescoFioraEst <francesco.fiora@est.tech>2021-07-13 11:18:01 +0100
committerFrancescoFioraEst <francesco.fiora@est.tech>2021-07-15 16:49:53 +0100
commit741b6c5c5eac437ae2a54a6dd5518513cdfe9034 (patch)
tree83a827f6ec593b442c13d6950d3b9440d2f3484c /runtime-controlloop/src/main/java/org/onap
parent258fdc2ddb8b5e130ccc2b287c10c3fd782b7ee9 (diff)
Convert Control Loop components to Spring and Supervision refactor
Issue-ID: POLICY-3463 Change-Id: Icc5b2ea576dbbe4cf954b51b8074f91884e93f56 Signed-off-by: FrancescoFioraEst <francesco.fiora@est.tech>
Diffstat (limited to 'runtime-controlloop/src/main/java/org/onap')
-rw-r--r--runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/Application.java3
-rw-r--r--runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/commissioning/CommissioningProvider.java37
-rw-r--r--runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/PolicyModelConfig.java (renamed from runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/ActivatorConfig.java)29
-rw-r--r--runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/messaging/Listener.java40
-rw-r--r--runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/messaging/MessageDispatcherActivator.java (renamed from runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/main/startstop/ClRuntimeActivator.java)65
-rw-r--r--runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/messaging/Publisher.java34
-rw-r--r--runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/instantiation/ControlLoopInstantiationProvider.java32
-rw-r--r--runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/monitoring/MonitoringProvider.java33
-rw-r--r--runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/SupervisionHandler.java172
-rw-r--r--runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/SupervisionScanner.java11
-rw-r--r--runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/AbstractParticipantPublisher.java48
-rw-r--r--runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantControlLoopStateChangePublisher.java20
-rw-r--r--runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantControlLoopUpdatePublisher.java34
-rw-r--r--runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantStateChangePublisher.java13
-rw-r--r--runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantStatusListener.java16
15 files changed, 242 insertions, 345 deletions
diff --git a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/Application.java b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/Application.java
index a2b6f62d4..6b772513c 100644
--- a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/Application.java
+++ b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/Application.java
@@ -22,8 +22,11 @@ package org.onap.policy.clamp.controlloop.runtime;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.ComponentScan;
@SpringBootApplication
+@ComponentScan({"org.onap.policy.clamp.controlloop.models.controlloop.persistence.provider",
+ "org.onap.policy.clamp.controlloop.runtime"})
public class Application {
public static void main(String[] args) {
diff --git a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/commissioning/CommissioningProvider.java b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/commissioning/CommissioningProvider.java
index f291c4e89..bf6e39914 100644
--- a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/commissioning/CommissioningProvider.java
+++ b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/commissioning/CommissioningProvider.java
@@ -25,8 +25,6 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.PropertyNamingStrategies;
import com.fasterxml.jackson.module.jsonSchema.JsonSchema;
import com.fasterxml.jackson.module.jsonSchema.factories.SchemaFactoryWrapper;
-import java.io.Closeable;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -36,11 +34,8 @@ import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.onap.policy.clamp.controlloop.models.controlloop.persistence.provider.ControlLoopProvider;
import org.onap.policy.clamp.controlloop.models.messages.rest.commissioning.CommissioningResponse;
-import org.onap.policy.clamp.controlloop.runtime.main.parameters.ClRuntimeParameterGroup;
import org.onap.policy.models.base.PfModelException;
-import org.onap.policy.models.base.PfModelRuntimeException;
import org.onap.policy.models.provider.PolicyModelsProvider;
-import org.onap.policy.models.provider.PolicyModelsProviderFactory;
import org.onap.policy.models.tosca.authorative.concepts.ToscaCapabilityType;
import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
import org.onap.policy.models.tosca.authorative.concepts.ToscaDataType;
@@ -59,7 +54,7 @@ import org.springframework.stereotype.Component;
* the callers.
*/
@Component
-public class CommissioningProvider implements Closeable {
+public class CommissioningProvider {
public static final String CONTROL_LOOP_NODE_TYPE = "org.onap.policy.clamp.controlloop.ControlLoop";
private final PolicyModelsProvider modelsProvider;
@@ -70,32 +65,12 @@ public class CommissioningProvider implements Closeable {
/**
* Create a commissioning provider.
*
- * @param controlLoopParameters the parameters for access to the database
- * @throws PfModelRuntimeException on errors creating the database provider
+ * @param modelsProvider the PolicyModelsProvider
+ * @param clProvider the ControlLoopProvider
*/
- public CommissioningProvider(ClRuntimeParameterGroup controlLoopParameters) {
- try {
- modelsProvider = new PolicyModelsProviderFactory()
- .createPolicyModelsProvider(controlLoopParameters.getDatabaseProviderParameters());
- } catch (PfModelException e) {
- throw new PfModelRuntimeException(e);
- }
-
- try {
- clProvider = new ControlLoopProvider(controlLoopParameters.getDatabaseProviderParameters());
- } catch (PfModelException e) {
- throw new PfModelRuntimeException(e);
- }
- }
-
- @Override
- public void close() throws IOException {
- try {
- modelsProvider.close();
- clProvider.close();
- } catch (PfModelException e) {
- throw new IOException("error closing modelsProvider", e);
- }
+ public CommissioningProvider(PolicyModelsProvider modelsProvider, ControlLoopProvider clProvider) {
+ this.modelsProvider = modelsProvider;
+ this.clProvider = clProvider;
}
/**
diff --git a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/ActivatorConfig.java b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/PolicyModelConfig.java
index 1d6b92e77..8a151d886 100644
--- a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/ActivatorConfig.java
+++ b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/PolicyModelConfig.java
@@ -21,26 +21,25 @@
package org.onap.policy.clamp.controlloop.runtime.config;
import org.onap.policy.clamp.controlloop.runtime.main.parameters.ClRuntimeParameterGroup;
-import org.onap.policy.clamp.controlloop.runtime.main.startstop.ClRuntimeActivator;
-import org.onap.policy.clamp.controlloop.runtime.supervision.SupervisionHandler;
+import org.onap.policy.models.base.PfModelException;
+import org.onap.policy.models.provider.PolicyModelsProvider;
+import org.onap.policy.models.provider.PolicyModelsProviderFactory;
+import org.onap.policy.models.provider.PolicyModelsProviderParameters;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
-public class ActivatorConfig {
+public class PolicyModelConfig {
- /**
- * Create and start ClRuntimeActivator.
- *
- * @param clRuntimeParameterGroup the parameters for the control loop runtime service
- * @param supervisionHandler the SupervisionHandler
- * @return ClRuntimeActivator
- */
@Bean
- public ClRuntimeActivator clRuntimeActivator(ClRuntimeParameterGroup clRuntimeParameterGroup,
- SupervisionHandler supervisionHandler) {
- var clRuntimeActivator = new ClRuntimeActivator(clRuntimeParameterGroup, supervisionHandler);
- clRuntimeActivator.start();
- return clRuntimeActivator;
+ public PolicyModelsProviderParameters policyModelsProviderParameters(
+ ClRuntimeParameterGroup clRuntimeParameterGroup) {
+ return clRuntimeParameterGroup.getDatabaseProviderParameters();
+ }
+
+ @Bean
+ public PolicyModelsProvider policyModelsProvider(PolicyModelsProviderParameters policyModelsProviderParameters)
+ throws PfModelException {
+ return new PolicyModelsProviderFactory().createPolicyModelsProvider(policyModelsProviderParameters);
}
}
diff --git a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/messaging/Listener.java b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/messaging/Listener.java
new file mode 100644
index 000000000..b67ddf2a7
--- /dev/null
+++ b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/messaging/Listener.java
@@ -0,0 +1,40 @@
+/*-
+ * ============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.runtime.config.messaging;
+
+import org.onap.policy.common.endpoints.listeners.ScoListener;
+
+public interface Listener {
+
+ /**
+ * Get the type of message of interest to the listener.
+ *
+ * @return type of message of interest to the listener
+ */
+ String getType();
+
+ /**
+ * Get listener to register.
+ *
+ * @return listener to register
+ */
+ <T> ScoListener<T> getScoListener();
+}
diff --git a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/main/startstop/ClRuntimeActivator.java b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/messaging/MessageDispatcherActivator.java
index 323f76178..891dab9ae 100644
--- a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/main/startstop/ClRuntimeActivator.java
+++ b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/messaging/MessageDispatcherActivator.java
@@ -18,53 +18,47 @@
* ============LICENSE_END=========================================================
*/
-package org.onap.policy.clamp.controlloop.runtime.main.startstop;
+package org.onap.policy.clamp.controlloop.runtime.config.messaging;
import java.io.Closeable;
import java.io.IOException;
import java.util.List;
+import java.util.stream.Stream;
import javax.ws.rs.core.Response.Status;
import lombok.Getter;
import org.onap.policy.clamp.controlloop.common.exception.ControlLoopRuntimeException;
import org.onap.policy.clamp.controlloop.runtime.main.parameters.ClRuntimeParameterGroup;
-import org.onap.policy.clamp.controlloop.runtime.supervision.SupervisionHandler;
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.utils.services.ServiceManagerContainer;
+import org.springframework.context.event.ContextRefreshedEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Component;
-/**
- * This class activates the control loop runtime component as a complete service together with all its controllers,
- * listeners & handlers.
- */
-public class ClRuntimeActivator extends ServiceManagerContainer implements Closeable {
- // Name of the message type for messages on topics
- private static final String[] MSG_TYPE_NAMES = {"messageType"};
+@Component
+public class MessageDispatcherActivator extends ServiceManagerContainer implements Closeable {
- @Getter
- private final ClRuntimeParameterGroup parameterGroup;
+ private static final String[] MSG_TYPE_NAMES = {"messageType"};
// Topics from which the application receives and to which the application sends messages
private List<TopicSink> topicSinks;
private List<TopicSource> topicSources;
- /**
- * Listens for messages on the topic, decodes them into a message, and then dispatches them.
- */
+ @Getter
private final MessageTypeDispatcher msgDispatcher;
/**
- * Instantiate the activator for the control loop runtime as a complete service.
+ * Constructor.
*
* @param clRuntimeParameterGroup the parameters for the control loop runtime service
- * @param supervisionHandler SupervisionHandler
+ * @param publishers array of Publishers
+ * @param listeners array of Listeners
* @throws ControlLoopRuntimeException if the activator does not start
*/
- public ClRuntimeActivator(final ClRuntimeParameterGroup clRuntimeParameterGroup,
- SupervisionHandler supervisionHandler) {
- this.parameterGroup = clRuntimeParameterGroup;
-
+ public MessageDispatcherActivator(final ClRuntimeParameterGroup clRuntimeParameterGroup, Publisher[] publishers,
+ Listener[] listeners) {
topicSinks = TopicEndpointManager.getManager()
.addTopicSinks(clRuntimeParameterGroup.getTopicParameterGroup().getTopicSinks());
@@ -80,15 +74,18 @@ public class ClRuntimeActivator extends ServiceManagerContainer implements Close
// @formatter:off
addAction("Topic endpoint management",
- () -> TopicEndpointManager.getManager().start(),
- () -> TopicEndpointManager.getManager().shutdown());
+ () -> TopicEndpointManager.getManager().start(),
+ () -> TopicEndpointManager.getManager().shutdown());
- addAction("Supervision Providers", () -> supervisionHandler.startProviders(),
- () -> supervisionHandler.stopProviders());
- addAction("Supervision Listeners", () -> supervisionHandler.startAndRegisterListeners(msgDispatcher),
- () -> supervisionHandler.stopAndUnregisterListeners(msgDispatcher));
- addAction("Supervision Publishers", () -> supervisionHandler.startAndRegisterPublishers(topicSinks),
- () -> supervisionHandler.stopAndUnregisterPublishers());
+ Stream.of(publishers).forEach(publisher ->
+ addAction("Publisher " + publisher.getClass().getSimpleName(),
+ () -> publisher.active(topicSinks),
+ () -> publisher.stop()));
+
+ Stream.of(listeners).forEach(listener ->
+ addAction("Listener " + listener.getClass().getSimpleName(),
+ () -> msgDispatcher.register(listener.getType(), listener.getScoListener()),
+ () -> msgDispatcher.unregister(listener.getType())));
addAction("Topic Message Dispatcher", this::registerMsgDispatcher, this::unregisterMsgDispatcher);
// @formatter:on
@@ -112,6 +109,18 @@ public class ClRuntimeActivator extends ServiceManagerContainer implements Close
}
}
+ /**
+ * Start Manager after the application is Started.
+ *
+ * @param cre Refreshed Event
+ */
+ @EventListener
+ public void handleContextStart(ContextRefreshedEvent cre) {
+ if (!isAlive()) {
+ start();
+ }
+ }
+
@Override
public void close() throws IOException {
if (isAlive()) {
diff --git a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/messaging/Publisher.java b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/messaging/Publisher.java
new file mode 100644
index 000000000..3cd4dff85
--- /dev/null
+++ b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/config/messaging/Publisher.java
@@ -0,0 +1,34 @@
+/*-
+ * ============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.runtime.config.messaging;
+
+import java.util.List;
+import org.onap.policy.common.endpoints.event.comm.TopicSink;
+
+/**
+ * Publisher.
+ */
+public interface Publisher {
+
+ void active(List<TopicSink> topicSinks);
+
+ void stop();
+}
diff --git a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/instantiation/ControlLoopInstantiationProvider.java b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/instantiation/ControlLoopInstantiationProvider.java
index c01a0b989..1011f620c 100644
--- a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/instantiation/ControlLoopInstantiationProvider.java
+++ b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/instantiation/ControlLoopInstantiationProvider.java
@@ -20,8 +20,6 @@
package org.onap.policy.clamp.controlloop.runtime.instantiation;
-import java.io.Closeable;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@@ -30,6 +28,7 @@ import java.util.function.UnaryOperator;
import java.util.stream.Collectors;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
+import lombok.AllArgsConstructor;
import org.onap.policy.clamp.controlloop.common.exception.ControlLoopException;
import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoop;
import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoopElement;
@@ -39,14 +38,12 @@ import org.onap.policy.clamp.controlloop.models.controlloop.persistence.provider
import org.onap.policy.clamp.controlloop.models.messages.rest.instantiation.InstantiationCommand;
import org.onap.policy.clamp.controlloop.models.messages.rest.instantiation.InstantiationResponse;
import org.onap.policy.clamp.controlloop.runtime.commissioning.CommissioningProvider;
-import org.onap.policy.clamp.controlloop.runtime.main.parameters.ClRuntimeParameterGroup;
import org.onap.policy.clamp.controlloop.runtime.supervision.SupervisionHandler;
import org.onap.policy.common.parameters.BeanValidationResult;
import org.onap.policy.common.parameters.ObjectValidationResult;
import org.onap.policy.common.parameters.ValidationResult;
import org.onap.policy.common.parameters.ValidationStatus;
import org.onap.policy.models.base.PfModelException;
-import org.onap.policy.models.base.PfModelRuntimeException;
import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
import org.onap.policy.models.tosca.authorative.concepts.ToscaNodeTemplate;
import org.springframework.stereotype.Component;
@@ -55,7 +52,8 @@ import org.springframework.stereotype.Component;
* This class is dedicated to the Instantiation of Commissioned control loop.
*/
@Component
-public class ControlLoopInstantiationProvider implements Closeable {
+@AllArgsConstructor
+public class ControlLoopInstantiationProvider {
private final ControlLoopProvider controlLoopProvider;
private final CommissioningProvider commissioningProvider;
private final SupervisionHandler supervisionHandler;
@@ -63,30 +61,6 @@ public class ControlLoopInstantiationProvider implements Closeable {
private static final Object lockit = new Object();
/**
- * Create a instantiation provider.
- *
- * @param controlLoopParameters the parameters for access to the database
- * @param commissioningProvider CommissioningProvider
- * @param supervisionHandler SupervisionHandler
- * @throws PfModelRuntimeException on errors creating a provider
- */
- public ControlLoopInstantiationProvider(ClRuntimeParameterGroup controlLoopParameters,
- CommissioningProvider commissioningProvider, SupervisionHandler supervisionHandler) {
- this.commissioningProvider = commissioningProvider;
- this.supervisionHandler = supervisionHandler;
- try {
- controlLoopProvider = new ControlLoopProvider(controlLoopParameters.getDatabaseProviderParameters());
- } catch (PfModelException e) {
- throw new PfModelRuntimeException(e);
- }
- }
-
- @Override
- public void close() throws IOException {
- controlLoopProvider.close();
- }
-
- /**
* Create control loops.
*
* @param controlLoops the control loop
diff --git a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/monitoring/MonitoringProvider.java b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/monitoring/MonitoringProvider.java
index a68505877..1f6246bd6 100644
--- a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/monitoring/MonitoringProvider.java
+++ b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/monitoring/MonitoringProvider.java
@@ -20,13 +20,12 @@
package org.onap.policy.clamp.controlloop.runtime.monitoring;
-import java.io.Closeable;
-import java.io.IOException;
import java.time.Instant;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import lombok.AllArgsConstructor;
import lombok.NonNull;
import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ClElementStatistics;
import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ClElementStatisticsList;
@@ -36,7 +35,6 @@ import org.onap.policy.clamp.controlloop.models.controlloop.concepts.Participant
import org.onap.policy.clamp.controlloop.models.controlloop.persistence.provider.ClElementStatisticsProvider;
import org.onap.policy.clamp.controlloop.models.controlloop.persistence.provider.ControlLoopProvider;
import org.onap.policy.clamp.controlloop.models.controlloop.persistence.provider.ParticipantStatisticsProvider;
-import org.onap.policy.clamp.controlloop.runtime.main.parameters.ClRuntimeParameterGroup;
import org.onap.policy.models.base.PfModelException;
import org.onap.policy.models.base.PfModelRuntimeException;
import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
@@ -46,7 +44,8 @@ import org.springframework.stereotype.Component;
* This class provides information about statistics data of CL elements and CL Participants in database to callers.
*/
@Component
-public class MonitoringProvider implements Closeable {
+@AllArgsConstructor
+public class MonitoringProvider {
private static final String DESC_ORDER = "DESC";
private final ParticipantStatisticsProvider participantStatisticsProvider;
@@ -54,32 +53,6 @@ public class MonitoringProvider implements Closeable {
private final ControlLoopProvider controlLoopProvider;
/**
- * Create a Monitoring provider.
- *
- * @param controlLoopParameters the parameters for access to the database
- * @throws PfModelRuntimeException on errors creating the provider
- */
- public MonitoringProvider(ClRuntimeParameterGroup controlLoopParameters) {
-
- try {
- participantStatisticsProvider =
- new ParticipantStatisticsProvider(controlLoopParameters.getDatabaseProviderParameters());
- clElementStatisticsProvider =
- new ClElementStatisticsProvider(controlLoopParameters.getDatabaseProviderParameters());
- controlLoopProvider = new ControlLoopProvider(controlLoopParameters.getDatabaseProviderParameters());
- } catch (PfModelException e) {
- throw new PfModelRuntimeException(e);
- }
- }
-
- @Override
- public void close() throws IOException {
- controlLoopProvider.close();
- clElementStatisticsProvider.close();
- participantStatisticsProvider.close();
- }
-
- /**
* Create participant statistics.
*
* @param participantStatistics the participant statistics
diff --git a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/SupervisionHandler.java b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/SupervisionHandler.java
index e1b4be48b..aba545750 100644
--- a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/SupervisionHandler.java
+++ b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/SupervisionHandler.java
@@ -21,34 +21,20 @@
package org.onap.policy.clamp.controlloop.runtime.supervision;
import java.util.List;
-import java.util.UUID;
import javax.ws.rs.core.Response;
-import javax.ws.rs.core.Response.Status;
+import lombok.AllArgsConstructor;
import org.apache.commons.collections4.CollectionUtils;
import org.onap.policy.clamp.controlloop.common.exception.ControlLoopException;
-import org.onap.policy.clamp.controlloop.common.exception.ControlLoopRuntimeException;
-import org.onap.policy.clamp.controlloop.common.handler.ControlLoopHandler;
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.ControlLoopState;
import org.onap.policy.clamp.controlloop.models.controlloop.concepts.Participant;
import org.onap.policy.clamp.controlloop.models.controlloop.persistence.provider.ControlLoopProvider;
import org.onap.policy.clamp.controlloop.models.controlloop.persistence.provider.ParticipantProvider;
-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.ParticipantMessageType;
import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantStatus;
-import org.onap.policy.clamp.controlloop.runtime.commissioning.CommissioningProvider;
-import org.onap.policy.clamp.controlloop.runtime.main.parameters.ClRuntimeParameterGroup;
import org.onap.policy.clamp.controlloop.runtime.monitoring.MonitoringProvider;
import org.onap.policy.clamp.controlloop.runtime.supervision.comm.ParticipantControlLoopStateChangePublisher;
import org.onap.policy.clamp.controlloop.runtime.supervision.comm.ParticipantControlLoopUpdatePublisher;
-import org.onap.policy.clamp.controlloop.runtime.supervision.comm.ParticipantStateChangePublisher;
-import org.onap.policy.clamp.controlloop.runtime.supervision.comm.ParticipantStatusListener;
-import org.onap.policy.common.endpoints.event.comm.TopicSink;
-import org.onap.policy.common.endpoints.listeners.MessageTypeDispatcher;
-import org.onap.policy.common.utils.services.ServiceManager;
-import org.onap.policy.common.utils.services.ServiceManagerException;
import org.onap.policy.models.base.PfModelException;
import org.onap.policy.models.tosca.authorative.concepts.ToscaConceptIdentifier;
import org.slf4j.Logger;
@@ -58,10 +44,12 @@ import org.springframework.stereotype.Component;
/**
* This class handles supervision of control loop instances, so only one object of this type should be built at a time.
*
- * <p/> It is effectively a singleton that is started at system start.
+ * <p/>
+ * It is effectively a singleton that is started at system start.
*/
@Component
-public class SupervisionHandler extends ControlLoopHandler {
+@AllArgsConstructor
+public class SupervisionHandler {
private static final Logger LOGGER = LoggerFactory.getLogger(SupervisionHandler.class);
private static final String CONTROL_LOOP_CANNOT_TRANSITION_FROM_STATE = "Control loop can't transition from state ";
@@ -69,59 +57,13 @@ public class SupervisionHandler extends ControlLoopHandler {
private static final String TO_STATE = " to state ";
private static final String AND_TRANSITIONING_TO_STATE = " and transitioning to state ";
- private ControlLoopProvider controlLoopProvider;
- private ParticipantProvider participantProvider;
+ private final ControlLoopProvider controlLoopProvider;
+ private final ParticipantProvider participantProvider;
private final MonitoringProvider monitoringProvider;
- private final CommissioningProvider commissioningProvider;
// Publishers for participant communication
- private ParticipantStateChangePublisher stateChangePublisher;
- private ParticipantControlLoopUpdatePublisher controlLoopUpdatePublisher;
- private ParticipantControlLoopStateChangePublisher controlLoopStateChangePublisher;
-
- private long supervisionScannerIntervalSec;
- private long participantStateChangeIntervalSec;
- private long participantClUpdateIntervalSec;
- private long participantClStateChangeIntervalSec;
-
- // Database scanner
- private SupervisionScanner scanner;
-
- /**
- * Used to manage the services.
- */
- private ServiceManager manager;
- private ServiceManager publisherManager;
-
- /**
- * Create a handler.
- *
- * @param clRuntimeParameterGroup the parameters for the control loop runtime
- * @param monitoringProvider the MonitoringProvider
- * @param commissioningProvider the CommissioningProvider
- */
- public SupervisionHandler(ClRuntimeParameterGroup clRuntimeParameterGroup, MonitoringProvider monitoringProvider,
- CommissioningProvider commissioningProvider) {
- super(clRuntimeParameterGroup.getDatabaseProviderParameters());
- this.monitoringProvider = monitoringProvider;
- this.commissioningProvider = commissioningProvider;
-
- // @formatter:off
- this.manager = new ServiceManager()
- .addAction("ControlLoop Provider",
- () -> controlLoopProvider = new ControlLoopProvider(getDatabaseProviderParameters()),
- () -> controlLoopProvider = null)
- .addAction("Participant Provider",
- () -> participantProvider = new ParticipantProvider(getDatabaseProviderParameters()),
- () -> participantProvider = null);
- // @formatter:on
-
- supervisionScannerIntervalSec = clRuntimeParameterGroup.getSupervisionScannerIntervalSec();
- participantStateChangeIntervalSec = clRuntimeParameterGroup.getParticipantClStateChangeIntervalSec();
- participantClUpdateIntervalSec = clRuntimeParameterGroup.getParticipantClUpdateIntervalSec();
- participantClStateChangeIntervalSec = clRuntimeParameterGroup.getParticipantClStateChangeIntervalSec();
-
- }
+ private final ParticipantControlLoopUpdatePublisher controlLoopUpdatePublisher;
+ private final ParticipantControlLoopStateChangePublisher controlLoopStateChangePublisher;
/**
* Supervision trigger called when a command is issued on control loops.
@@ -155,55 +97,6 @@ public class SupervisionHandler extends ControlLoopHandler {
}
}
- @Override
- public void startAndRegisterListeners(MessageTypeDispatcher msgDispatcher) {
- msgDispatcher.register(ParticipantMessageType.PARTICIPANT_STATUS.name(), new ParticipantStatusListener(this));
- }
-
- @Override
- public void startAndRegisterPublishers(List<TopicSink> topicSinks) {
- // @formatter:off
- this.publisherManager = new ServiceManager()
- .addAction("Supervision scanner",
- () -> scanner =
- new SupervisionScanner(controlLoopProvider, supervisionScannerIntervalSec),
- () -> scanner.close())
- .addAction("ControlLoopUpdate publisher",
- () -> controlLoopUpdatePublisher =
- new ParticipantControlLoopUpdatePublisher(topicSinks, participantClUpdateIntervalSec),
- () -> controlLoopUpdatePublisher.terminate())
- .addAction("StateChange Publisher",
- () -> stateChangePublisher =
- new ParticipantStateChangePublisher(topicSinks, participantStateChangeIntervalSec),
- () -> stateChangePublisher.terminate())
- .addAction("ControlLoopStateChange Publisher",
- () -> controlLoopStateChangePublisher =
- new ParticipantControlLoopStateChangePublisher(topicSinks, participantClStateChangeIntervalSec),
- () -> controlLoopStateChangePublisher.terminate());
- // @formatter:on
- try {
- publisherManager.start();
- } catch (final ServiceManagerException exp) {
- throw new ControlLoopRuntimeException(Status.INTERNAL_SERVER_ERROR,
- "Supervision handler start of publishers or scanner failed", exp);
- }
- }
-
- @Override
- public void stopAndUnregisterPublishers() {
- try {
- publisherManager.stop();
- } catch (final ServiceManagerException exp) {
- throw new ControlLoopRuntimeException(Status.INTERNAL_SERVER_ERROR,
- "Supervision handler stop of publishers or scanner failed", exp);
- }
- }
-
- @Override
- public void stopAndUnregisterListeners(MessageTypeDispatcher msgDispatcher) {
- msgDispatcher.unregister(ParticipantMessageType.PARTICIPANT_STATUS.name());
- }
-
/**
* Handle a ParticipantStatus message from a participant.
*
@@ -270,7 +163,7 @@ public class SupervisionHandler extends ControlLoopHandler {
case UNINITIALISED2PASSIVE:
case PASSIVE:
controlLoop.setState(ControlLoopState.PASSIVE2UNINITIALISED);
- sendControlLoopStateChange(controlLoop);
+ controlLoopStateChangePublisher.send(controlLoop);
break;
case PASSIVE2UNINITIALISED:
@@ -294,7 +187,7 @@ public class SupervisionHandler extends ControlLoopHandler {
break;
case UNINITIALISED:
controlLoop.setState(ControlLoopState.UNINITIALISED2PASSIVE);
- sendControlLoopUpdate(controlLoop);
+ controlLoopUpdatePublisher.send(controlLoop);
break;
case UNINITIALISED2PASSIVE:
@@ -305,7 +198,7 @@ public class SupervisionHandler extends ControlLoopHandler {
case RUNNING:
controlLoop.setState(ControlLoopState.RUNNING2PASSIVE);
- sendControlLoopStateChange(controlLoop);
+ controlLoopStateChangePublisher.send(controlLoop);
break;
default:
@@ -329,7 +222,7 @@ public class SupervisionHandler extends ControlLoopHandler {
case PASSIVE:
controlLoop.setState(ControlLoopState.PASSIVE2RUNNING);
- sendControlLoopStateChange(controlLoop);
+ controlLoopStateChangePublisher.send(controlLoop);
break;
default:
@@ -339,25 +232,6 @@ public class SupervisionHandler extends ControlLoopHandler {
}
}
- private void sendControlLoopUpdate(ControlLoop controlLoop) throws PfModelException {
- var pclu = new ParticipantControlLoopUpdate();
- pclu.setControlLoopId(controlLoop.getKey().asIdentifier());
- pclu.setControlLoop(controlLoop);
- // TODO: We should look up the correct TOSCA node template here for the control loop
- // Tiny hack implemented to return the tosca service template entry from the database and be passed onto dmaap
- pclu.setControlLoopDefinition(commissioningProvider.getToscaServiceTemplate(null, null));
- controlLoopUpdatePublisher.send(pclu);
- }
-
- private void sendControlLoopStateChange(ControlLoop controlLoop) {
- var clsc = new ParticipantControlLoopStateChange();
- clsc.setControlLoopId(controlLoop.getKey().asIdentifier());
- clsc.setMessageId(UUID.randomUUID());
- clsc.setOrderedState(controlLoop.getOrderedState());
-
- controlLoopStateChangePublisher.send(clsc);
- }
-
private void superviseParticipant(ParticipantStatus participantStatusMessage)
throws PfModelException, ControlLoopException {
if (participantStatusMessage.getParticipantId() == null) {
@@ -427,26 +301,6 @@ public class SupervisionHandler extends ControlLoopHandler {
}
}
- @Override
- public void startProviders() {
- try {
- manager.start();
- } catch (final ServiceManagerException exp) {
- throw new ControlLoopRuntimeException(Status.INTERNAL_SERVER_ERROR,
- "Supervision handler start of providers failed", exp);
- }
- }
-
- @Override
- public void stopProviders() {
- try {
- manager.stop();
- } catch (final ServiceManagerException exp) {
- throw new ControlLoopRuntimeException(Status.INTERNAL_SERVER_ERROR,
- "Supervision handler stop of providers failed", exp);
- }
- }
-
private void exceptionOccured(Response.Status status, String reason) throws ControlLoopException {
throw new ControlLoopException(status, reason);
}
diff --git a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/SupervisionScanner.java b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/SupervisionScanner.java
index 4f3faf8af..68f5830c0 100644
--- a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/SupervisionScanner.java
+++ b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/SupervisionScanner.java
@@ -27,13 +27,16 @@ 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.persistence.provider.ControlLoopProvider;
+import org.onap.policy.clamp.controlloop.runtime.main.parameters.ClRuntimeParameterGroup;
import org.onap.policy.models.base.PfModelException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
/**
* This class is used to scan the control loops in the database and check if they are in the correct state.
*/
+@Component
public class SupervisionScanner implements Runnable, Closeable {
private static final Logger LOGGER = LoggerFactory.getLogger(SupervisionScanner.class);
@@ -43,15 +46,17 @@ public class SupervisionScanner implements Runnable, Closeable {
/**
* Constructor for instantiating SupervisionScanner.
*
+ * @param clRuntimeParameterGroup the parameters for the control loop runtime
* @param controlLoopProvider the provider to use to read control loops from the database
- * @param interval time interval to perform scans
*/
- public SupervisionScanner(final ControlLoopProvider controlLoopProvider, final long interval) {
+ public SupervisionScanner(final ControlLoopProvider controlLoopProvider,
+ ClRuntimeParameterGroup clRuntimeParameterGroup) {
this.controlLoopProvider = controlLoopProvider;
// Kick off the timer
timerPool = makeTimerPool();
- timerPool.scheduleAtFixedRate(this, 0, interval, TimeUnit.SECONDS);
+ timerPool.scheduleAtFixedRate(this, 0, clRuntimeParameterGroup.getSupervisionScannerIntervalSec(),
+ TimeUnit.SECONDS);
}
@Override
diff --git a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/AbstractParticipantPublisher.java b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/AbstractParticipantPublisher.java
index c54856101..3c87b05b4 100644
--- a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/AbstractParticipantPublisher.java
+++ b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/AbstractParticipantPublisher.java
@@ -21,35 +21,17 @@
package org.onap.policy.clamp.controlloop.runtime.supervision.comm;
import java.util.List;
-import lombok.Getter;
+import javax.ws.rs.core.Response.Status;
+import org.onap.policy.clamp.controlloop.common.exception.ControlLoopRuntimeException;
import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantMessage;
+import org.onap.policy.clamp.controlloop.runtime.config.messaging.Publisher;
import org.onap.policy.common.endpoints.event.comm.TopicSink;
import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClient;
-public abstract class AbstractParticipantPublisher<E extends ParticipantMessage> {
+public abstract class AbstractParticipantPublisher<E extends ParticipantMessage> implements Publisher {
- private final TopicSinkClient topicSinkClient;
-
- @Getter
- private final long intervalSec;
-
- /**
- * Constructor.
- *
- * @param topicSinks the topic sinks
- * @param intervalSec time interval to send ParticipantStateChange messages
- */
- protected AbstractParticipantPublisher(final List<TopicSink> topicSinks, long intervalSec) {
- this.topicSinkClient = new TopicSinkClient(topicSinks.get(0));
- this.intervalSec = intervalSec;
- }
-
- /**
- * Terminates the current timer.
- */
- public void terminate() {
- // Nothing to terminate, this publisher does not have a timer
- }
+ private TopicSinkClient topicSinkClient;
+ private boolean active = false;
/**
* Method to send Participant message to participants on demand.
@@ -57,6 +39,24 @@ public abstract class AbstractParticipantPublisher<E extends ParticipantMessage>
* @param participantMessage the Participant message
*/
public void send(final E participantMessage) {
+ if (!active) {
+ throw new ControlLoopRuntimeException(Status.NOT_ACCEPTABLE, "Not Active!");
+ }
topicSinkClient.send(participantMessage);
}
+
+
+ @Override
+ public void active(List<TopicSink> topicSinks) {
+ if (topicSinks.size() != 1) {
+ throw new IllegalArgumentException("Topic Sink must be one");
+ }
+ this.topicSinkClient = new TopicSinkClient(topicSinks.get(0));
+ active = true;
+ }
+
+ @Override
+ public void stop() {
+ active = false;
+ }
}
diff --git a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantControlLoopStateChangePublisher.java b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantControlLoopStateChangePublisher.java
index c9d0a4fe4..734ccb842 100644
--- a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantControlLoopStateChangePublisher.java
+++ b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantControlLoopStateChangePublisher.java
@@ -20,23 +20,29 @@
package org.onap.policy.clamp.controlloop.runtime.supervision.comm;
-import java.util.List;
+import java.util.UUID;
+import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoop;
import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantControlLoopStateChange;
-import org.onap.policy.common.endpoints.event.comm.TopicSink;
+import org.springframework.stereotype.Component;
/**
* This class is used to send ParticipantControlLoopStateChangePublisher messages to participants on DMaaP.
*/
+@Component
public class ParticipantControlLoopStateChangePublisher
extends AbstractParticipantPublisher<ParticipantControlLoopStateChange> {
/**
- * Constructor for instantiating ParticipantControlLoopStateChangePublisherPublisher.
+ * Send ControlLoopStateChange to Participant.
*
- * @param topicSinks the topic sinks
- * @param interval time interval to send ParticipantControlLoopStateChangePublisher messages
+ * @param controlLoop the ControlLoop
*/
- public ParticipantControlLoopStateChangePublisher(final List<TopicSink> topicSinks, final long interval) {
- super(topicSinks, interval);
+ public void send(ControlLoop controlLoop) {
+ var clsc = new ParticipantControlLoopStateChange();
+ clsc.setControlLoopId(controlLoop.getKey().asIdentifier());
+ clsc.setMessageId(UUID.randomUUID());
+ clsc.setOrderedState(controlLoop.getOrderedState());
+
+ super.send(clsc);
}
}
diff --git a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantControlLoopUpdatePublisher.java b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantControlLoopUpdatePublisher.java
index fbbd95fbc..8d40c5e69 100644
--- a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantControlLoopUpdatePublisher.java
+++ b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantControlLoopUpdatePublisher.java
@@ -20,22 +20,42 @@
package org.onap.policy.clamp.controlloop.runtime.supervision.comm;
-import java.util.List;
+import org.onap.policy.clamp.controlloop.models.controlloop.concepts.ControlLoop;
import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantControlLoopUpdate;
-import org.onap.policy.common.endpoints.event.comm.TopicSink;
+import org.onap.policy.clamp.controlloop.runtime.commissioning.CommissioningProvider;
+import org.onap.policy.models.base.PfModelException;
+import org.springframework.stereotype.Component;
/**
* This class is used to send ParticipantControlLoopUpdate messages to participants on DMaaP.
*/
+@Component
public class ParticipantControlLoopUpdatePublisher extends AbstractParticipantPublisher<ParticipantControlLoopUpdate> {
+ private final CommissioningProvider commissioningProvider;
+
+ /**
+ * Constructor.
+ *
+ * @param commissioningProvider the CommissioningProvider
+ */
+ public ParticipantControlLoopUpdatePublisher(CommissioningProvider commissioningProvider) {
+ this.commissioningProvider = commissioningProvider;
+ }
+
/**
- * Constructor for instantiating ParticipantUpdatePublisher.
+ * Send ControlLoopUpdate to Participant.
*
- * @param topicSinks the topic sinks
- * @param interval time interval to send ParticipantControlLoopUpdate messages
+ * @param controlLoop the ControlLoop
+ * @throws PfModelException on errors getting the Control Loop Definition
*/
- public ParticipantControlLoopUpdatePublisher(final List<TopicSink> topicSinks, final long interval) {
- super(topicSinks, interval);
+ public void send(ControlLoop controlLoop) throws PfModelException {
+ var pclu = new ParticipantControlLoopUpdate();
+ pclu.setControlLoopId(controlLoop.getKey().asIdentifier());
+ pclu.setControlLoop(controlLoop);
+ // TODO: We should look up the correct TOSCA node template here for the control loop
+ // Tiny hack implemented to return the tosca service template entry from the database and be passed onto dmaap
+ pclu.setControlLoopDefinition(commissioningProvider.getToscaServiceTemplate(null, null));
+ super.send(pclu);
}
}
diff --git a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantStateChangePublisher.java b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantStateChangePublisher.java
index 20cdea6f4..b63cbdf03 100644
--- a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantStateChangePublisher.java
+++ b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantStateChangePublisher.java
@@ -20,22 +20,13 @@
package org.onap.policy.clamp.controlloop.runtime.supervision.comm;
-import java.util.List;
import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantStateChange;
-import org.onap.policy.common.endpoints.event.comm.TopicSink;
+import org.springframework.stereotype.Component;
/**
* This class is used to send ParticipantStateChange messages to participants on DMaaP.
*/
+@Component
public class ParticipantStateChangePublisher extends AbstractParticipantPublisher<ParticipantStateChange> {
- /**
- * Constructor for instantiating ParticipantStateChangePublisher.
- *
- * @param topicSinks the topic sinks
- * @param interval time interval to send ParticipantStateChange messages
- */
- public ParticipantStateChangePublisher(List<TopicSink> topicSinks, long interval) {
- super(topicSinks, interval);
- }
}
diff --git a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantStatusListener.java b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantStatusListener.java
index 88b838613..8fa076240 100644
--- a/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantStatusListener.java
+++ b/runtime-controlloop/src/main/java/org/onap/policy/clamp/controlloop/runtime/supervision/comm/ParticipantStatusListener.java
@@ -20,18 +20,22 @@
package org.onap.policy.clamp.controlloop.runtime.supervision.comm;
+import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantMessageType;
import org.onap.policy.clamp.controlloop.models.messages.dmaap.participant.ParticipantStatus;
+import org.onap.policy.clamp.controlloop.runtime.config.messaging.Listener;
import org.onap.policy.clamp.controlloop.runtime.supervision.SupervisionHandler;
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;
+import org.springframework.stereotype.Component;
/**
* Listener for ParticipantStatus messages sent by participants.
*/
-public class ParticipantStatusListener extends ScoListener<ParticipantStatus> {
+@Component
+public class ParticipantStatusListener extends ScoListener<ParticipantStatus> implements Listener {
private static final Logger LOGGER = LoggerFactory.getLogger(ParticipantStatusListener.class);
private final SupervisionHandler supervisionHandler;
@@ -50,4 +54,14 @@ public class ParticipantStatusListener extends ScoListener<ParticipantStatus> {
LOGGER.debug("ParticipantStatus message received from participant - {}", participantStatusMessage);
supervisionHandler.handleParticipantStatusMessage(participantStatusMessage);
}
+
+ @Override
+ public String getType() {
+ return ParticipantMessageType.PARTICIPANT_STATUS.name();
+ }
+
+ @Override
+ public ScoListener<ParticipantStatus> getScoListener() {
+ return this;
+ }
}