Adding PdpStatus listener & handler to policy/pap

1) Adding PdpHeartbeatListener to keep listening for PdpStatus messages
coming from Pdp's over DMaaP either for registration or just heartbeat.
2) Adding PdpStatusMessageHandler to handle new Pdp registration and
also heartbeat coming from Pdp's.
Registration Flow -
 1. Find the PdpSubgroup based on PdpType & SupportedPolicyTypes coming
    in PdpStatus message.
 2. If not found, don't register pdp and log the error message.
 3. If found, check if PdpInstance is already added.
 4. If not added, add PdpInstance to the subgroup and increment the
    currentInstanceCount.
 5. Create and send PdpUpdate & PdpStateChange message to the Pdp.
 6. Update the changes in DB.

Heatbeat Flow -
 1. Find the PdpInstance based on details in PdpStatus message.
 2. Validate the details from message to what in DB.
 3. If correct, persist the health & statistics information in DB.
 4. If not correct, send a PdpUpdate & PdpStateChange message to the Pdp.

3) Unit tests will come as seperate review.

Change-Id: If705193259999e2ab077b78961282c998b949f57
Issue-ID: POLICY-1443
Signed-off-by: ramverma <ram.krishna.verma@est.tech>

1 files changed, 11 insertions, 0 deletions

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 42634ee7..a60232fd 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
@@ -37,6 +37,7 @@ import org.onap.policy.models.pdp.enums.PdpMessageType;
 import org.onap.policy.pap.main.PapConstants;
 import org.onap.policy.pap.main.PolicyModelsProviderFactoryWrapper;
 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.Publisher;
 import org.onap.policy.pap.main.comm.TimerManager;
@@ -76,6 +77,11 @@ public class PapActivator extends ServiceManagerContainer {
     private final RequestIdDispatcher<PdpStatus> reqIdDispatcher;
 
     /**
+     * Listener for anonymous {@link PdpStatus} messages either for registration or heartbeat.
+     */
+    private final PdpHeartbeatListener pdpHeartbeatListener;
+
+    /**
      * Instantiate the activator for policy pap as a complete service.
      *
      * @param papParameterGroup the parameters for the pap service
@@ -91,6 +97,7 @@ public class PapActivator extends ServiceManagerContainer {
             this.papParameterGroup = papParameterGroup;
             this.msgDispatcher = new MessageTypeDispatcher(MSG_TYPE_NAMES);
             this.reqIdDispatcher = new RequestIdDispatcher<>(PdpStatus.class, REQ_ID_NAMES);
+            this.pdpHeartbeatListener = new PdpHeartbeatListener();
 
         } catch (final RuntimeException e) {
             throw new PolicyPapRuntimeException(e);
@@ -119,6 +126,10 @@ public class PapActivator extends ServiceManagerContainer {
             () -> Registry.register(PapConstants.REG_PAP_DAO_FACTORY, daoFactory.get()),
             () -> Registry.unregister(PapConstants.REG_PAP_DAO_FACTORY));
 
+        addAction("Pdp Heartbeat Listener",
+            () -> reqIdDispatcher.register(pdpHeartbeatListener),
+            () -> reqIdDispatcher.unregister(pdpHeartbeatListener));
+
         addAction("Request ID Dispatcher",
             () -> msgDispatcher.register(PdpMessageType.PDP_STATUS.name(), this.reqIdDispatcher),
             () -> msgDispatcher.unregister(PdpMessageType.PDP_STATUS.name()));