/*-
* ============LICENSE_START=======================================================
* Copyright (C) 2019 Nordix Foundation.
- * Modifications Copyright (C) 2019 AT&T Intellectual Property.
+ * Modifications Copyright (C) 2019-2021 AT&T Intellectual Property.
+ * Modifications Copyright (C) 2021 Bell Canada. All rights reserved.
* ================================================================================
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
package org.onap.policy.pap.main.startstop;
-import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.onap.policy.common.endpoints.event.comm.TopicEndpointManager;
+import org.onap.policy.common.endpoints.event.comm.TopicListener;
import org.onap.policy.common.endpoints.event.comm.TopicSource;
-import org.onap.policy.common.endpoints.http.server.RestServer;
import org.onap.policy.common.endpoints.listeners.MessageTypeDispatcher;
import org.onap.policy.common.endpoints.listeners.RequestIdDispatcher;
import org.onap.policy.common.parameters.ParameterService;
import org.onap.policy.pap.main.PolicyPapRuntimeException;
import org.onap.policy.pap.main.comm.PdpHeartbeatListener;
import org.onap.policy.pap.main.comm.PdpModifyRequestMap;
-import org.onap.policy.pap.main.comm.PdpTracker;
import org.onap.policy.pap.main.comm.Publisher;
import org.onap.policy.pap.main.comm.TimerManager;
import org.onap.policy.pap.main.notification.PolicyNotifier;
import org.onap.policy.pap.main.parameters.PapParameterGroup;
import org.onap.policy.pap.main.parameters.PdpModifyRequestMapParams;
-import org.onap.policy.pap.main.parameters.PdpParameters;
-import org.onap.policy.pap.main.rest.HealthCheckRestControllerV1;
-import org.onap.policy.pap.main.rest.PapAafFilter;
import org.onap.policy.pap.main.rest.PapStatisticsManager;
-import org.onap.policy.pap.main.rest.PdpGroupHealthCheckControllerV1;
-import org.onap.policy.pap.main.rest.PdpGroupQueryControllerV1;
-import org.onap.policy.pap.main.rest.PdpGroupStateChangeControllerV1;
-import org.onap.policy.pap.main.rest.PolicyStatusControllerV1;
-import org.onap.policy.pap.main.rest.StatisticsRestControllerV1;
-import org.onap.policy.pap.main.rest.depundep.PdpGroupDeleteControllerV1;
-import org.onap.policy.pap.main.rest.depundep.PdpGroupDeployControllerV1;
-import org.onap.policy.pap.main.rest.depundep.PolicyUndeployerImpl;
+import org.onap.policy.pap.main.rest.PolicyUndeployerImpl;
+import org.springframework.context.event.ContextClosedEvent;
+import org.springframework.context.event.ContextRefreshedEvent;
+import org.springframework.context.event.EventListener;
+import org.springframework.stereotype.Component;
/**
* This class activates Policy Administration (PAP) as a complete service together with all its controllers, listeners &
*
* @author Ram Krishna Verma (ram.krishna.verma@est.tech)
*/
+@Component
public class PapActivator extends ServiceManagerContainer {
private static final String[] MSG_TYPE_NAMES = { "messageName" };
private static final String[] REQ_ID_NAMES = { "response", "responseTo" };
*/
private static final int MAX_MISSED_HEARTBEATS = 3;
- private final PapParameterGroup papParameterGroup;
+ private PapParameterGroup papParameterGroup;
/**
* Listens for messages on the topic, decodes them into a {@link PdpStatus} message, and then dispatches them to
- * {@link #reqIdDispatcher}.
+ * {@link #responseReqIdDispatcher}.
*/
- private final MessageTypeDispatcher msgDispatcher;
+ private final MessageTypeDispatcher responseMsgDispatcher;
+ private final MessageTypeDispatcher heartbeatMsgDispatcher;
/**
* Listens for {@link PdpStatus} messages and then routes them to the listener associated with the ID of the
* originating request.
*/
- private final RequestIdDispatcher<PdpStatus> reqIdDispatcher;
+ private final RequestIdDispatcher<PdpStatus> responseReqIdDispatcher;
+ private final RequestIdDispatcher<PdpStatus> heartbeatReqIdDispatcher;
/**
* Listener for anonymous {@link PdpStatus} messages either for registration or heartbeat.
*
* @param papParameterGroup the parameters for the pap service
*/
- public PapActivator(final PapParameterGroup papParameterGroup) {
+ public PapActivator(PapParameterGroup papParameterGroup) {
super("Policy PAP");
-
+ this.papParameterGroup = papParameterGroup;
TopicEndpointManager.getManager().addTopics(papParameterGroup.getTopicParameterGroup());
try {
- this.papParameterGroup = papParameterGroup;
- this.msgDispatcher = new MessageTypeDispatcher(MSG_TYPE_NAMES);
- this.reqIdDispatcher = new RequestIdDispatcher<>(PdpStatus.class, REQ_ID_NAMES);
- this.pdpHeartbeatListener = new PdpHeartbeatListener();
+ this.responseMsgDispatcher = new MessageTypeDispatcher(MSG_TYPE_NAMES);
+ this.heartbeatMsgDispatcher = new MessageTypeDispatcher(MSG_TYPE_NAMES);
+ this.responseReqIdDispatcher = new RequestIdDispatcher<>(PdpStatus.class, REQ_ID_NAMES);
+ this.heartbeatReqIdDispatcher = new RequestIdDispatcher<>(PdpStatus.class, REQ_ID_NAMES);
+ this.pdpHeartbeatListener = new PdpHeartbeatListener(papParameterGroup.getPdpParameters(),
+ papParameterGroup.isSavePdpStatisticsInDb());
} catch (final RuntimeException e) {
throw new PolicyPapRuntimeException(e);
}
- papParameterGroup.getRestServerParameters().setName(papParameterGroup.getName());
- final Object pdpUpdateLock = new Object();
- final PdpParameters pdpParams = papParameterGroup.getPdpParameters();
+ final var pdpUpdateLock = new Object();
+ final var pdpParams = papParameterGroup.getPdpParameters();
final AtomicReference<Publisher<PdpMessage>> pdpPub = new AtomicReference<>();
final AtomicReference<Publisher<PolicyNotification>> notifyPub = new AtomicReference<>();
final AtomicReference<TimerManager> pdpUpdTimers = new AtomicReference<>();
final AtomicReference<TimerManager> pdpStChgTimers = new AtomicReference<>();
- final AtomicReference<TimerManager> heartBeatTimers = new AtomicReference<>();
+ final AtomicReference<ScheduledExecutorService> pdpExpirationTimer = new AtomicReference<>();
final AtomicReference<PolicyModelsProviderFactoryWrapper> daoFactory = new AtomicReference<>();
final AtomicReference<PdpModifyRequestMap> requestMap = new AtomicReference<>();
- final AtomicReference<RestServer> restServer = new AtomicReference<>();
final AtomicReference<PolicyNotifier> notifier = new AtomicReference<>();
// @formatter:off
() -> Registry.unregister(PapConstants.REG_PAP_DAO_FACTORY));
addAction("Pdp Heartbeat Listener",
- () -> reqIdDispatcher.register(pdpHeartbeatListener),
- () -> reqIdDispatcher.unregister(pdpHeartbeatListener));
+ () -> heartbeatReqIdDispatcher.register(pdpHeartbeatListener),
+ () -> heartbeatReqIdDispatcher.unregister(pdpHeartbeatListener));
+
+ addAction("Response Request ID Dispatcher",
+ () -> responseMsgDispatcher.register(PdpMessageType.PDP_STATUS.name(), this.responseReqIdDispatcher),
+ () -> responseMsgDispatcher.unregister(PdpMessageType.PDP_STATUS.name()));
- addAction("Request ID Dispatcher",
- () -> msgDispatcher.register(PdpMessageType.PDP_STATUS.name(), this.reqIdDispatcher),
- () -> msgDispatcher.unregister(PdpMessageType.PDP_STATUS.name()));
+ addAction("Heartbeat Request ID Dispatcher",
+ () -> heartbeatMsgDispatcher.register(PdpMessageType.PDP_STATUS.name(), this.heartbeatReqIdDispatcher),
+ () -> heartbeatMsgDispatcher.unregister(PdpMessageType.PDP_STATUS.name()));
- addAction("Message Dispatcher",
- this::registerMsgDispatcher,
- this::unregisterMsgDispatcher);
+ addAction("Response Message Dispatcher",
+ () -> registerMsgDispatcher(responseMsgDispatcher, PapConstants.TOPIC_POLICY_PDP_PAP),
+ () -> unregisterMsgDispatcher(responseMsgDispatcher, PapConstants.TOPIC_POLICY_PDP_PAP));
+
+ addAction("Heartbeat Message Dispatcher",
+ () -> registerMsgDispatcher(heartbeatMsgDispatcher, PapConstants.TOPIC_POLICY_HEARTBEAT),
+ () -> unregisterMsgDispatcher(heartbeatMsgDispatcher, PapConstants.TOPIC_POLICY_HEARTBEAT));
addAction("topics",
TopicEndpointManager.getManager()::start,
() -> Registry.register(PapConstants.REG_STATISTICS_MANAGER, new PapStatisticsManager()),
() -> Registry.unregister(PapConstants.REG_STATISTICS_MANAGER));
+ addAction("PAP Activator",
+ () -> Registry.register(PapConstants.REG_PAP_ACTIVATOR, this),
+ () -> Registry.unregister(PapConstants.REG_PAP_ACTIVATOR));
+
addAction("PDP publisher",
() -> {
pdpPub.set(new Publisher<>(PapConstants.TOPIC_POLICY_PDP_PAP));
() -> Registry.register(PapConstants.REG_POLICY_NOTIFIER, notifier.get()),
() -> Registry.unregister(PapConstants.REG_POLICY_NOTIFIER));
- addAction("PDP heart beat timers",
- () -> {
- long maxWaitHeartBeatMs = MAX_MISSED_HEARTBEATS * pdpParams.getHeartBeatMs();
- heartBeatTimers.set(new TimerManager("heart beat", maxWaitHeartBeatMs));
- startThread(heartBeatTimers.get());
- },
- () -> heartBeatTimers.get().stop());
-
addAction("PDP update timers",
() -> {
pdpUpdTimers.set(new TimerManager("update", pdpParams.getUpdateParameters().getMaxWaitMs()));
addAction("PDP modification requests",
() -> {
requestMap.set(new PdpModifyRequestMap(
- new PdpModifyRequestMapParams()
- .setDaoFactory(daoFactory.get())
- .setModifyLock(pdpUpdateLock)
- .setParams(pdpParams)
- .setPolicyNotifier(notifier.get())
- .setPdpPublisher(pdpPub.get())
- .setResponseDispatcher(reqIdDispatcher)
- .setStateChangeTimers(pdpStChgTimers.get())
- .setUpdateTimers(pdpUpdTimers.get())));
+ PdpModifyRequestMapParams.builder()
+ .maxPdpAgeMs(MAX_MISSED_HEARTBEATS * pdpParams.getHeartBeatMs())
+ .daoFactory(daoFactory.get())
+ .modifyLock(pdpUpdateLock)
+ .params(pdpParams)
+ .policyNotifier(notifier.get())
+ .pdpPublisher(pdpPub.get())
+ .responseDispatcher(responseReqIdDispatcher)
+ .stateChangeTimers(pdpStChgTimers.get())
+ .updateTimers(pdpUpdTimers.get())
+ .savePdpStatistics(papParameterGroup.isSavePdpStatisticsInDb())
+ .build()));
Registry.register(PapConstants.REG_PDP_MODIFY_MAP, requestMap.get());
// now that it's registered, we can attach a "policy undeploy" provider
},
() -> Registry.unregister(PapConstants.REG_PDP_MODIFY_MAP));
- addAction("PDP heart beat tracker",
- () -> Registry.register(PapConstants.REG_PDP_TRACKER, PdpTracker.builder()
- .daoFactory(daoFactory.get())
- .timers(heartBeatTimers.get())
- .modifyLock(pdpUpdateLock)
- .requestMap(requestMap.get())
- .build()),
- () -> Registry.unregister(PapConstants.REG_PDP_TRACKER));
-
- addAction("REST server",
+ addAction("PDP expiration timer",
() -> {
- RestServer server = new RestServer(papParameterGroup.getRestServerParameters(), PapAafFilter.class,
- HealthCheckRestControllerV1.class,
- StatisticsRestControllerV1.class,
- PdpGroupDeployControllerV1.class,
- PdpGroupDeleteControllerV1.class,
- PdpGroupStateChangeControllerV1.class,
- PdpGroupQueryControllerV1.class,
- PdpGroupHealthCheckControllerV1.class,
- PolicyStatusControllerV1.class);
- restServer.set(server);
- restServer.get().start();
+ long frequencyMs = pdpParams.getHeartBeatMs();
+ pdpExpirationTimer.set(Executors.newScheduledThreadPool(1));
+ pdpExpirationTimer.get().scheduleWithFixedDelay(
+ requestMap.get()::removeExpiredPdps,
+ frequencyMs,
+ frequencyMs,
+ TimeUnit.MILLISECONDS);
},
- () -> restServer.get().stop());
+ () -> pdpExpirationTimer.get().shutdown());
+
// @formatter:on
}
* @param runner function to run in the background
*/
private void startThread(final Runnable runner) {
- final Thread thread = new Thread(runner);
+ final var thread = new Thread(runner);
thread.setDaemon(true);
thread.start();
/**
* Registers the dispatcher with the topic source(s).
+ * @param dispatcher dispatcher to register
+ * @param topic topic of interest
*/
- private void registerMsgDispatcher() {
- for (final TopicSource source : TopicEndpointManager.getManager()
- .getTopicSources(Arrays.asList(PapConstants.TOPIC_POLICY_PDP_PAP))) {
- source.register(msgDispatcher);
+ private void registerMsgDispatcher(TopicListener dispatcher, String topic) {
+ for (final TopicSource source : TopicEndpointManager.getManager().getTopicSources(List.of(topic))) {
+ source.register(dispatcher);
}
}
/**
* Unregisters the dispatcher from the topic source(s).
+ * @param dispatcher dispatcher to unregister
+ * @param topic topic of interest
+ */
+ private void unregisterMsgDispatcher(TopicListener dispatcher, String topic) {
+ for (final TopicSource source : TopicEndpointManager.getManager().getTopicSources(List.of(topic))) {
+ source.unregister(dispatcher);
+ }
+ }
+
+ /**
+ * Handle ContextRefreshEvent.
+ *
+ * @param ctxRefreshedEvent the ContextRefreshedEvent
+ */
+ @EventListener
+ public void handleContextRefreshEvent(ContextRefreshedEvent ctxRefreshedEvent) {
+ if (!isAlive()) {
+ start();
+ }
+ }
+
+ /**
+ * Handle ContextClosedEvent.
+ *
+ * @param ctxClosedEvent the ContextClosedEvent
*/
- private void unregisterMsgDispatcher() {
- for (final TopicSource source : TopicEndpointManager.getManager()
- .getTopicSources(Arrays.asList(PapConstants.TOPIC_POLICY_PDP_PAP))) {
- source.unregister(msgDispatcher);
+ @EventListener
+ public void handleContextClosedEvent(ContextClosedEvent ctxClosedEvent) {
+ if (isAlive()) {
+ stop();
}
}
}
Code Review / policy / pap.git / blobdiff