diff options
7 files changed, 78 insertions, 25 deletions
diff --git a/main/src/main/java/org/onap/policy/pap/main/PapConstants.java b/main/src/main/java/org/onap/policy/pap/main/PapConstants.java index 7d4cb7b4..f38a82e0 100644 --- a/main/src/main/java/org/onap/policy/pap/main/PapConstants.java +++ b/main/src/main/java/org/onap/policy/pap/main/PapConstants.java @@ -25,7 +25,7 @@ import org.onap.policy.common.utils.network.NetworkUtil; /** * Names of various items contained in the Registry. */ -public class PapConstants { +public final class PapConstants { // Registry keys public static final String REG_PAP_ACTIVATOR = "object:activator/pap"; @@ -38,6 +38,7 @@ public class PapConstants { // topic names public static final String TOPIC_POLICY_PDP_PAP = "POLICY-PDP-PAP"; public static final String TOPIC_POLICY_NOTIFICATION = "POLICY-NOTIFICATION"; + public static final String TOPIC_POLICY_HEARTBEAT = "POLICY-HEARTBEAT"; // policy components names public static final String POLICY_API = "api"; diff --git a/main/src/main/java/org/onap/policy/pap/main/startstop/PapActivator.java b/main/src/main/java/org/onap/policy/pap/main/startstop/PapActivator.java index 617275b4..3b089400 100644 --- a/main/src/main/java/org/onap/policy/pap/main/startstop/PapActivator.java +++ b/main/src/main/java/org/onap/policy/pap/main/startstop/PapActivator.java @@ -22,12 +22,13 @@ 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.client.HttpClientFactoryInstance; import org.onap.policy.common.endpoints.http.server.RestServer; @@ -84,15 +85,17 @@ public class PapActivator extends ServiceManagerContainer { /** * 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. @@ -111,8 +114,10 @@ public class PapActivator extends ServiceManagerContainer { try { this.papParameterGroup = papParameterGroup; - this.msgDispatcher = new MessageTypeDispatcher(MSG_TYPE_NAMES); - this.reqIdDispatcher = new RequestIdDispatcher<>(PdpStatus.class, REQ_ID_NAMES); + 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()); @@ -149,16 +154,24 @@ public class PapActivator extends ServiceManagerContainer { () -> Registry.unregister(PapConstants.REG_PAP_DAO_FACTORY)); addAction("Pdp Heartbeat Listener", - () -> reqIdDispatcher.register(pdpHeartbeatListener), - () -> reqIdDispatcher.unregister(pdpHeartbeatListener)); + () -> heartbeatReqIdDispatcher.register(pdpHeartbeatListener), + () -> heartbeatReqIdDispatcher.unregister(pdpHeartbeatListener)); - addAction("Request ID Dispatcher", - () -> msgDispatcher.register(PdpMessageType.PDP_STATUS.name(), this.reqIdDispatcher), - () -> msgDispatcher.unregister(PdpMessageType.PDP_STATUS.name())); + addAction("Response Request ID Dispatcher", + () -> responseMsgDispatcher.register(PdpMessageType.PDP_STATUS.name(), this.responseReqIdDispatcher), + () -> responseMsgDispatcher.unregister(PdpMessageType.PDP_STATUS.name())); - addAction("Message Dispatcher", - this::registerMsgDispatcher, - this::unregisterMsgDispatcher); + addAction("Heartbeat Request ID Dispatcher", + () -> heartbeatMsgDispatcher.register(PdpMessageType.PDP_STATUS.name(), this.heartbeatReqIdDispatcher), + () -> heartbeatMsgDispatcher.unregister(PdpMessageType.PDP_STATUS.name())); + + 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, @@ -215,7 +228,7 @@ public class PapActivator extends ServiceManagerContainer { .params(pdpParams) .policyNotifier(notifier.get()) .pdpPublisher(pdpPub.get()) - .responseDispatcher(reqIdDispatcher) + .responseDispatcher(responseReqIdDispatcher) .stateChangeTimers(pdpStChgTimers.get()) .updateTimers(pdpUpdTimers.get()) .savePdpStatistics(papParameterGroup.isSavePdpStatisticsInDb()) @@ -288,21 +301,23 @@ public class PapActivator extends ServiceManagerContainer { /** * 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() { - for (final TopicSource source : TopicEndpointManager.getManager() - .getTopicSources(Arrays.asList(PapConstants.TOPIC_POLICY_PDP_PAP))) { - source.unregister(msgDispatcher); + private void unregisterMsgDispatcher(TopicListener dispatcher, String topic) { + for (final TopicSource source : TopicEndpointManager.getManager().getTopicSources(List.of(topic))) { + source.unregister(dispatcher); } } } diff --git a/main/src/test/resources/e2e/PapConfigParameters.json b/main/src/test/resources/e2e/PapConfigParameters.json index 610ded8c..b10f5801 100644 --- a/main/src/test/resources/e2e/PapConfigParameters.json +++ b/main/src/test/resources/e2e/PapConfigParameters.json @@ -32,6 +32,13 @@ "topic" : "POLICY-PDP-PAP", "servers" : [ "message-router" ], "topicCommInfrastructure" : "noop" + }, + { + "topic" : "POLICY-HEARTBEAT", + "effectiveTopic": "POLICY-PDP-PAP", + "consumerGroup": "policy-pap", + "servers" : [ "message-router" ], + "topicCommInfrastructure" : "noop" }], "topicSinks" : [{ "topic" : "POLICY-PDP-PAP", diff --git a/main/src/test/resources/parameters/PapConfigParameters.json b/main/src/test/resources/parameters/PapConfigParameters.json index 452bc9a7..09adfcec 100644 --- a/main/src/test/resources/parameters/PapConfigParameters.json +++ b/main/src/test/resources/parameters/PapConfigParameters.json @@ -33,6 +33,13 @@ "topic" : "POLICY-PDP-PAP", "servers" : [ "message-router" ], "topicCommInfrastructure" : "noop" + }, + { + "topic" : "POLICY-HEARTBEAT", + "effectiveTopic": "POLICY-PDP-PAP", + "consumerGroup": "policy-pap", + "servers" : [ "message-router" ], + "topicCommInfrastructure" : "noop" }], "topicSinks" : [{ "topic" : "POLICY-PDP-PAP", diff --git a/main/src/test/resources/parameters/PapConfigParametersStd.json b/main/src/test/resources/parameters/PapConfigParametersStd.json index 0868aa41..0f881430 100644 --- a/main/src/test/resources/parameters/PapConfigParametersStd.json +++ b/main/src/test/resources/parameters/PapConfigParametersStd.json @@ -34,6 +34,13 @@ "topic" : "POLICY-PDP-PAP", "servers" : [ "message-router" ], "topicCommInfrastructure" : "noop" + }, + { + "topic" : "POLICY-HEARTBEAT", + "effectiveTopic": "POLICY-PDP-PAP", + "consumerGroup": "policy-pap", + "servers" : [ "message-router" ], + "topicCommInfrastructure" : "noop" }], "topicSinks" : [{ "topic" : "POLICY-PDP-PAP", diff --git a/packages/policy-pap-tarball/src/main/resources/etc/defaultConfig.json b/packages/policy-pap-tarball/src/main/resources/etc/defaultConfig.json index 2f37cb66..94315bb6 100644 --- a/packages/policy-pap-tarball/src/main/resources/etc/defaultConfig.json +++ b/packages/policy-pap-tarball/src/main/resources/etc/defaultConfig.json @@ -35,6 +35,14 @@ "servers" : [ "message-router" ], "topicCommInfrastructure" : "dmaap", "fetchTimeout": 15000 + }, + { + "topic" : "POLICY-HEARTBEAT", + "effectiveTopic": "POLICY-PDP-PAP", + "consumerGroup": "policy-pap", + "servers" : [ "message-router" ], + "topicCommInfrastructure" : "dmaap", + "fetchTimeout": 15000 }], "topicSinks" : [{ "topic" : "POLICY-PDP-PAP", diff --git a/testsuites/stability/src/main/resources/papsetup/config/pap/etc/defaultConfig.json b/testsuites/stability/src/main/resources/papsetup/config/pap/etc/defaultConfig.json index 4af2f5df..b98cdef9 100644 --- a/testsuites/stability/src/main/resources/papsetup/config/pap/etc/defaultConfig.json +++ b/testsuites/stability/src/main/resources/papsetup/config/pap/etc/defaultConfig.json @@ -34,6 +34,14 @@ "servers" : [ "10.2.0.41:3904" ], "topicCommInfrastructure" : "dmaap", "fetchTimeout": 15000 + }, + { + "topic" : "PDP-HEARTBEAT", + "effectiveTopic": "POLICY-PDP-PAP", + "consumerGroup": "policy-pap", + "servers" : [ "10.2.0.41:3904" ], + "topicCommInfrastructure" : "dmaap", + "fetchTimeout": 15000 }], "topicSinks" : [{ "topic" : "POLICY-PDP-PAP", @@ -46,4 +54,4 @@ "topicCommInfrastructure" : "dmaap" }] } -}
\ No newline at end of file +} |