Changes to handle PdpUpdate and PdpStateChange messages

1) Adding listeners and handlers for PdpUpdate messages.
2) Adding listeners and handlers for PdpStateChange messages.
3) Some changes to accomodate recent changes in the models.
4) Test cases.

Change-Id: I66644d94fd2a242738d9b0f78e5d0ece76342057
Issue-ID: POLICY-1453
Signed-off-by: a.sreekumar <ajith.sreekumar@est.tech>

13 files changed, 705 insertions, 185 deletions

diff --git a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/ApexStarterActivator.java b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/ApexStarterActivator.java
index 5058e7cd4..a6bd702aa 100644
--- a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/ApexStarterActivator.java
+++ b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/ApexStarterActivator.java
@@ -26,15 +26,17 @@ import java.util.Properties;
 import lombok.Getter;
 import lombok.Setter;
 
+import org.onap.policy.apex.starter.comm.PdpStateChangeListener;
+import org.onap.policy.apex.starter.comm.PdpStatusPublisher;
+import org.onap.policy.apex.starter.comm.PdpUpdateListener;
 import org.onap.policy.apex.starter.exception.ApexStarterException;
-import org.onap.policy.apex.starter.handler.CommunicationHandler;
+import org.onap.policy.apex.starter.exception.ApexStarterRunTimeException;
 import org.onap.policy.apex.starter.handler.PdpMessageHandler;
 import org.onap.policy.apex.starter.parameters.ApexStarterParameterGroup;
-import org.onap.policy.apex.starter.parameters.PdpStatusParameters;
 import org.onap.policy.common.endpoints.event.comm.TopicEndpoint;
 import org.onap.policy.common.endpoints.event.comm.TopicSink;
 import org.onap.policy.common.endpoints.event.comm.TopicSource;
-import org.onap.policy.common.parameters.ParameterService;
+import org.onap.policy.common.endpoints.listeners.MessageTypeDispatcher;
 import org.onap.policy.common.utils.services.Registry;
 import org.onap.policy.common.utils.services.ServiceManager;
 import org.onap.policy.common.utils.services.ServiceManagerException;
@@ -50,7 +52,13 @@ public class ApexStarterActivator {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(ApexStarterActivator.class);
     private final ApexStarterParameterGroup apexStarterParameterGroup;
-    private CommunicationHandler communicationHandler;
+    private List<TopicSink> topicSinks;// topics to which apex-pdp sends pdp status
+    private List<TopicSource> topicSources; // topics to which apex-pdp listens to for messages from pap.
+    private static final String[] MSG_TYPE_NAMES = { "messageName" };
+    /**
+     * Listens for messages on the topic, decodes them into a message, and then dispatches them.
+     */
+    private final MessageTypeDispatcher msgDispatcher;
 
     /**
      * Used to manage the services.
@@ -64,14 +72,22 @@ public class ApexStarterActivator {
     public ApexStarterActivator(final ApexStarterParameterGroup apexStarterParameterGroup,
             final Properties topicProperties) {
 
-        final List<TopicSink> topicSinks = TopicEndpoint.manager.addTopicSinks(topicProperties);
-        final List<TopicSource> topicSources = TopicEndpoint.manager.addTopicSources(topicProperties);
-        this.apexStarterParameterGroup = apexStarterParameterGroup;
+        topicSinks = TopicEndpoint.manager.addTopicSinks(topicProperties);
+        topicSources = TopicEndpoint.manager.addTopicSources(topicProperties);
 
         // TODO: instanceId currently set as a random string, could be fetched from actual deployment
         final int random = (int) (Math.random() * 100);
         final String instanceId = "apex_" + random;
 
+        try {
+            this.apexStarterParameterGroup = apexStarterParameterGroup;
+            this.msgDispatcher = new MessageTypeDispatcher(MSG_TYPE_NAMES);
+        } catch (final RuntimeException e) {
+            throw new ApexStarterRunTimeException(e);
+        }
+
+        final PdpUpdateListener pdpUpdateListener = new PdpUpdateListener();
+        final PdpStateChangeListener pdpStateChangeListener = new PdpStateChangeListener();
         // @formatter:off
         this.manager = new ServiceManager()
                 .addAction("topics",
@@ -80,22 +96,44 @@ public class ApexStarterActivator {
                 .addAction("set alive",
                         () -> setAlive(true),
                         () -> setAlive(false))
-                .addAction("register context map",
+                .addAction("register pdp status context object",
                         () -> Registry.register(ApexStarterConstants.REG_PDP_STATUS_OBJECT,
                                 new PdpMessageHandler().createPdpStatusFromParameters(instanceId,
                                         apexStarterParameterGroup.getPdpStatusParameters())),
                         () -> Registry.unregister(ApexStarterConstants.REG_PDP_STATUS_OBJECT))
-                .addAction("register parameters",
-                        () -> registerToParameterService(apexStarterParameterGroup),
-                        () -> deregisterToParameterService(apexStarterParameterGroup))
-                .addAction("Communication handler",
-                        () -> startCommunicationHandler(topicSinks, topicSources,
-                                apexStarterParameterGroup.getPdpStatusParameters()),
-                        () -> communicationHandler.stop());
+                .addAction("topic sinks",
+                        () -> Registry.register(ApexStarterConstants.REG_APEX_PDP_TOPIC_SINKS, topicSinks),
+                        () -> Registry.unregister(ApexStarterConstants.REG_APEX_PDP_TOPIC_SINKS))
+                .addAction("Pdp Status publisher",
+                        () -> Registry.register(ApexStarterConstants.REG_PDP_STATUS_PUBLISHER,
+                                new PdpStatusPublisher(topicSinks,
+                                        apexStarterParameterGroup.getPdpStatusParameters().getTimeIntervalMs())),
+                        () -> stopAndRemovePdpStatusPublisher())
+                .addAction("Register pdp update listener",
+                    () -> msgDispatcher.register("PDP_UPDATE", pdpUpdateListener),
+                    () -> msgDispatcher.unregister("PDP_UPDATE"))
+                .addAction("Register pdp state change request dispatcher",
+                        () -> msgDispatcher.register("PDP_STATE_CHANGE", pdpStateChangeListener),
+                        () -> msgDispatcher.unregister("PDP_STATE_CHANGE"))
+                .addAction("Message Dispatcher",
+                    () -> registerMsgDispatcher(),
+                    () -> unregisterMsgDispatcher());
         // @formatter:on
     }
 
     /**
+     * Method to stop and unregister the pdp status publisher.
+     */
+    private void stopAndRemovePdpStatusPublisher() {
+        final PdpStatusPublisher pdpStatusPublisher =
+                Registry.get(ApexStarterConstants.REG_PDP_STATUS_PUBLISHER, PdpStatusPublisher.class);
+        // send a final heartbeat with terminated status
+        pdpStatusPublisher.send(new PdpMessageHandler().getTerminatedPdpStatus());
+        pdpStatusPublisher.terminate();
+        Registry.unregister(ApexStarterConstants.REG_PDP_STATUS_PUBLISHER);
+    }
+
+    /**
      * Initialize ApexStarter service.
      *
      * @throws ApexStarterException on errors in initializing the service
@@ -124,9 +162,9 @@ public class ApexStarterActivator {
         if (!isAlive()) {
             throw new IllegalStateException("activator is not running");
         }
-
         try {
             manager.stop();
+            Registry.unregister(ApexStarterConstants.REG_APEX_STARTER_ACTIVATOR);
         } catch (final ServiceManagerException exp) {
             LOGGER.error("ApexStarter termination failed");
             throw new ApexStarterException(exp.getMessage(), exp);
@@ -142,39 +180,21 @@ public class ApexStarterActivator {
         return apexStarterParameterGroup;
     }
 
-
-    /**
-     * Method to register the parameters to Common Parameter Service.
-     *
-     * @param apexStarterParameterGroup the apex starter parameter group
-     */
-    public void registerToParameterService(final ApexStarterParameterGroup apexStarterParameterGroup) {
-        ParameterService.register(apexStarterParameterGroup);
-    }
-
     /**
-     * Method to deregister the parameters from Common Parameter Service.
-     *
-     * @param apexStarterParameterGroup the apex starter parameter group
+     * Registers the dispatcher with the topic source(s).
      */
-    public void deregisterToParameterService(final ApexStarterParameterGroup apexStarterParameterGroup) {
-        ParameterService.deregister(apexStarterParameterGroup.getName());
+    private void registerMsgDispatcher() {
+        for (final TopicSource source : topicSources) {
+            source.register(msgDispatcher);
+        }
     }
 
     /**
-     * Starts the communication handler which handles the communication between apex pdp and pap.
-     *
-     * @param pdpStatusParameters
-     * @param topicSources
-     * @param topicSinks
-     *
-     * @throws ApexStarterException if the handler start fails
+     * Unregisters the dispatcher from the topic source(s).
      */
-    public void startCommunicationHandler(final List<TopicSink> topicSinks, final List<TopicSource> topicSources,
-            final PdpStatusParameters pdpStatusParameters) throws ApexStarterException {
-        communicationHandler = new CommunicationHandler(topicSinks, topicSources, pdpStatusParameters);
-        if (!communicationHandler.start()) {
-            throw new ApexStarterException("Failed to start the communication handler for ApexStarter");
+    private void unregisterMsgDispatcher() {
+        for (final TopicSource source : topicSources) {
+            source.unregister(msgDispatcher);
         }
     }
 }
diff --git a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/ApexStarterConstants.java b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/ApexStarterConstants.java
index c2d3dce05..6173872e0 100644
--- a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/ApexStarterConstants.java
+++ b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/ApexStarterConstants.java
@@ -27,4 +27,8 @@ public class ApexStarterConstants {
     // Registry keys
     public static final String REG_APEX_STARTER_ACTIVATOR = "object:activator/apex_starter";
     public static final String REG_PDP_STATUS_OBJECT = "object:pdp/status";
+    public static final String REG_APEX_TOSCA_POLICY_LIST = "object:apex/tosca/policy/list";
+    public static final String REG_PDP_STATUS_PUBLISHER = "object:pdp/status/publisher";
+    public static final String REG_APEX_PDP_TOPIC_SINKS = "object:apex/pdp/topic/sinks";
+    public static final String REG_APEX_ENGINE_HANDLER = "object:engine/apex/handler";
 }
diff --git a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/ApexStarterMain.java b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/ApexStarterMain.java
index 7f11de713..e4465ed92 100644
--- a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/ApexStarterMain.java
+++ b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/ApexStarterMain.java
@@ -57,7 +57,7 @@ public class ApexStarterMain {
             // The arguments return a string if there is a message to print and we should exit
             final String argumentMessage = arguments.parse(args);
             if (argumentMessage != null) {
-                LOGGER.info(argumentMessage);
+                LOGGER.debug(argumentMessage);
                 return;
             }
             // Validate that the arguments are sane
@@ -105,7 +105,6 @@ public class ApexStarterMain {
         LOGGER.info("Started ApexStarter service");
     }
 
-
     /**
      * Get the parameters specified in JSON.
      *
@@ -154,7 +153,6 @@ public class ApexStarterMain {
         }
     }
 
-
     /**
      * The main method.
      *
diff --git a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/comm/PdpStateChangeListener.java b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/comm/PdpStateChangeListener.java
new file mode 100644
index 000000000..0bd0d935f
--- /dev/null
+++ b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/comm/PdpStateChangeListener.java
@@ -0,0 +1,52 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2019 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.apex.starter.comm;
+
+import org.onap.policy.apex.starter.handler.PdpStateChangeMessageHandler;
+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.onap.policy.models.pdp.concepts.PdpStateChange;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Listener for Pdp state change messages sent by PAP.
+ *
+ * @author Ajith Sreekumar (ajith.sreekumar@est.tech)
+ */
+public class PdpStateChangeListener extends ScoListener<PdpStateChange> {
+    private static final Logger LOGGER = LoggerFactory.getLogger(PdpStateChangeListener.class);
+
+    /**
+     * Constructs the object.
+     */
+    public PdpStateChangeListener() {
+        super(PdpStateChange.class);
+    }
+
+    @Override
+    public void onTopicEvent(final CommInfrastructure infra, final String topic, final StandardCoderObject sco,
+            final PdpStateChange pdpStateChangeMsg) {
+        LOGGER.debug("Pdp state change message received from PAP. - {}", pdpStateChangeMsg);
+        new PdpStateChangeMessageHandler().handlePdpStateChangeEvent(pdpStateChangeMsg);
+    }
+}
diff --git a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/handler/PdpStatusPublisher.java b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/comm/PdpStatusPublisher.java
index 2853ae798..69c9dd18d 100644
--- a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/handler/PdpStatusPublisher.java
+++ b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/comm/PdpStatusPublisher.java
@@ -18,17 +18,16 @@
  * ============LICENSE_END=========================================================
  */
 
-package org.onap.policy.apex.starter.handler;
+package org.onap.policy.apex.starter.comm;
 
 import java.util.List;
 import java.util.Timer;
 import java.util.TimerTask;
 
-import lombok.Getter;
-import lombok.Setter;
-
+import org.onap.policy.apex.starter.handler.PdpMessageHandler;
 import org.onap.policy.common.endpoints.event.comm.TopicSink;
 import org.onap.policy.common.endpoints.event.comm.client.TopicSinkClient;
+import org.onap.policy.models.pdp.concepts.PdpStatus;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -39,16 +38,11 @@ import org.slf4j.LoggerFactory;
  */
 public class PdpStatusPublisher extends TimerTask {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(CommunicationHandler.class);
+    private static final Logger LOGGER = LoggerFactory.getLogger(PdpStatusPublisher.class);
 
-    private List<TopicSink> topicSinks;
     private TopicSinkClient topicSinkClient;
     private Timer timer;
-    private PdpMessageHandler pdpMessageHandler;
-
-    @Getter
-    @Setter(lombok.AccessLevel.PRIVATE)
-    private volatile boolean alive = false;
+    private long interval;
 
     /**
      * Constructor for instantiating PdpStatusPublisher
@@ -57,35 +51,42 @@ public class PdpStatusPublisher extends TimerTask {
      * @param topicSinks
      * @param apexStarterParameterGroup
      */
-    public PdpStatusPublisher(final List<TopicSink> topicSinks, final int interval) {
-        this.topicSinks = topicSinks;
+    public PdpStatusPublisher(final List<TopicSink> topicSinks, final long interval) {
         this.topicSinkClient = new TopicSinkClient(topicSinks.get(0));
-        this.pdpMessageHandler = new PdpMessageHandler();
+        this.interval = interval;
         timer = new Timer(false);
-        timer.scheduleAtFixedRate(this, 0, interval * 1000L);
-        setAlive(true);
+        timer.scheduleAtFixedRate(this, 0, interval);
     }
 
     @Override
     public void run() {
-        topicSinkClient.send(pdpMessageHandler.createPdpStatusFromContext());
-        LOGGER.info("Sent heartbeat to PAP");
+        final PdpStatus pdpStatus = new PdpMessageHandler().createPdpStatusFromContext();
+        topicSinkClient.send(pdpStatus);
+        LOGGER.debug("Sent heartbeat to PAP - {}", pdpStatus);
     }
 
+    /**
+     * Terminates the current timer.
+     */
     public void terminate() {
         timer.cancel();
         timer.purge();
-        setAlive(false);
     }
 
-    public PdpStatusPublisher updateInterval(final int interval) {
-        terminate();
-        return new PdpStatusPublisher(topicSinks, interval);
+    /**
+     * Get the current time interval used by the timer task.
+     *
+     * @return interval
+     */
+    public long getInterval() {
+        return interval;
     }
 
-    public void send() {
-        topicSinkClient.send(pdpMessageHandler.createPdpStatusFromContext());
-        LOGGER.info("Sent pdp status response message to PAP");
+    /**
+     * Method to send pdp status message to pap on demand.
+     */
+    public void send(final PdpStatus pdpStatus) {
+        topicSinkClient.send(pdpStatus);
+        LOGGER.debug("Sent pdp status message to PAP - {}", pdpStatus);
     }
-
 }
diff --git a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/comm/PdpUpdateListener.java b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/comm/PdpUpdateListener.java
new file mode 100644
index 000000000..46ce30e57
--- /dev/null
+++ b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/comm/PdpUpdateListener.java
@@ -0,0 +1,54 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2019 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.apex.starter.comm;
+
+import org.onap.policy.apex.starter.handler.PdpUpdateMessageHandler;
+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.onap.policy.models.pdp.concepts.PdpUpdate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Listener for Pdp update messages sent by PAP.
+ *
+ * @author Ajith Sreekumar (ajith.sreekumar@est.tech)
+ */
+public class PdpUpdateListener extends ScoListener<PdpUpdate> {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(PdpUpdateListener.class);
+
+    /**
+     * Constructs the object.
+     */
+    public PdpUpdateListener() {
+        super(PdpUpdate.class);
+    }
+
+    @Override
+    public void onTopicEvent(final CommInfrastructure infra, final String topic, final StandardCoderObject sco,
+            final PdpUpdate pdpUpdateMsg) {
+        LOGGER.debug("Pdp update message received from PAP - {}", pdpUpdateMsg);
+        new PdpUpdateMessageHandler().handlePdpUpdateEvent(pdpUpdateMsg);
+
+    }
+}
diff --git a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/engine/ApexEngineHandler.java b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/engine/ApexEngineHandler.java
new file mode 100644
index 000000000..4682a3bc5
--- /dev/null
+++ b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/engine/ApexEngineHandler.java
@@ -0,0 +1,109 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2019 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.apex.starter.engine;
+
+import java.io.IOException;
+import java.io.StringReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Map;
+
+import org.onap.policy.apex.model.basicmodel.concepts.ApexException;
+import org.onap.policy.apex.service.engine.main.ApexMain;
+import org.onap.policy.apex.starter.exception.ApexStarterException;
+import org.onap.policy.common.utils.coder.CoderException;
+import org.onap.policy.common.utils.coder.StandardCoder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This class instantiates the Apex Engine based on instruction from PAP.
+ *
+ * @author Ajith Sreekumar (ajith.sreekumar@est.tech)
+ */
+public class ApexEngineHandler {
+    private static final Logger LOGGER = LoggerFactory.getLogger(ApexEngineHandler.class);
+
+    private ApexMain apexMain;
+
+    /**
+     * Constructs the object. Extracts the apex config and model files and instantiates the apex engine.
+     *
+     * @param properties
+     * @return ApexEngineHandler
+     * @throws ApexStarterException
+     */
+
+    @SuppressWarnings("unchecked")
+    public ApexEngineHandler(final String properties) throws ApexStarterException {
+        final StandardCoder standardCoder = new StandardCoder();
+        Map<String, Map<String, String>> body;
+        try {
+            body = standardCoder.decode(new StringReader(properties), Map.class);
+        } catch (final CoderException e) {
+            throw new ApexStarterException(e);
+        }
+        final Map<String, String> engineServiceParameters = body.get("engineServiceParameters");
+        final String policyModel = engineServiceParameters.get("policy_type_impl").toString();
+        engineServiceParameters.remove("policy_type_impl");
+        final String apexConfig = body.toString();
+
+        final String modelFilePath = createFile(policyModel, "modelFile");
+
+        final String apexConfigFilePath = createFile(apexConfig, "apexConfigFile");
+
+        final String[] apexArgs = { "-rfr", "target/classes", "-c", apexConfigFilePath, "-m", modelFilePath };
+        apexMain = new ApexMain(apexArgs);
+    }
+
+    /**
+     * Method to create the policy model file
+     *
+     * @param policyModel
+     * @param modelFilePath
+     * @throws ApexStarterException
+     */
+    private String createFile(final String fileContent, final String fileName) throws ApexStarterException {
+        try {
+            final Path path = Files.createTempFile(fileName, ".json");
+            Files.write(path, fileContent.getBytes(StandardCharsets.UTF_8));
+            return path.toAbsolutePath().toString();
+        } catch (final IOException e) {
+            final String errorMessage = "error creating  from the properties received in PdpUpdate.";
+            LOGGER.error(errorMessage, e);
+            throw new ApexStarterException(errorMessage, e);
+        }
+    }
+
+    public boolean isApexEngineRunning() {
+        return null != apexMain;
+    }
+
+    public void shutdown() throws ApexStarterException {
+        try {
+            apexMain.shutdown();
+            apexMain = null;
+        } catch (final ApexException e) {
+            throw new ApexStarterException(e);
+        }
+    }
+}
diff --git a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/handler/CommunicationHandler.java b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/handler/CommunicationHandler.java
deleted file mode 100644
index 197b34ef5..000000000
--- a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/handler/CommunicationHandler.java
+++ /dev/null
@@ -1,94 +0,0 @@
-/*-
- * ============LICENSE_START=======================================================
- *  Copyright (C) 2019 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.apex.starter.handler;
-
-import java.util.List;
-
-import org.onap.policy.apex.starter.parameters.PdpStatusParameters;
-import org.onap.policy.common.capabilities.Startable;
-import org.onap.policy.common.endpoints.event.comm.TopicSink;
-import org.onap.policy.common.endpoints.event.comm.TopicSource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * This class manages the communication between apex pdp and pap.
- *
- * @author Ajith Sreekumar (ajith.sreekumar@est.tech)
- */
-public class CommunicationHandler implements Startable {
-
-    private static final Logger LOGGER = LoggerFactory.getLogger(CommunicationHandler.class);
-
-    private List<TopicSink> topicSinks; // topics to which apex-pdp sends pdp status
-    private List<TopicSource> topicSources; // topics to which apex-pdp listens to for messages from pap.
-
-    private PdpStatusPublisher pdpStatusPublisher;
-    private PdpStatusParameters pdpStatusParameters;
-
-    /**
-     * Constructor for instantiating CommunicationHandler
-     *
-     * @param topicSinks topics to which apex-pdp sends pdp status
-     * @param topicSources topics to which apex-pdp listens to for messages from pap.
-     * @param pdpStatusParameters pdp status parameters read from the configuration file
-     */
-    public CommunicationHandler(final List<TopicSink> topicSinks, final List<TopicSource> topicSources,
-            final PdpStatusParameters pdpStatusParameters) {
-        this.topicSinks = topicSinks;
-        this.topicSources = topicSources;
-        this.pdpStatusParameters = pdpStatusParameters;
-    }
-
-    @Override
-    public boolean start() {
-        try {
-            pdpStatusPublisher = new PdpStatusPublisher(topicSinks, pdpStatusParameters.getTimeInterval());
-        } catch (final Exception e) {
-            LOGGER.error("Failed to start communication handler for apex-pdp", e);
-            return false;
-        }
-        return true;
-    }
-
-    @Override
-    public boolean stop() {
-        try {
-            pdpStatusPublisher.terminate();
-        } catch (final Exception e) {
-            LOGGER.error("Failed to stop communication handler for apex-pdp", e);
-            return false;
-        }
-        return true;
-    }
-
-    @Override
-    public void shutdown() {
-        stop();
-    }
-
-    @Override
-    public boolean isAlive() {
-        return pdpStatusPublisher.isAlive();
-    }
-
-}
diff --git a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/handler/PdpMessageHandler.java b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/handler/PdpMessageHandler.java
index a0f4f5863..a022a8a08 100644
--- a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/handler/PdpMessageHandler.java
+++ b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/handler/PdpMessageHandler.java
@@ -25,12 +25,16 @@ import java.util.List;
 
 import org.onap.policy.apex.starter.ApexStarterConstants;
 import org.onap.policy.apex.starter.parameters.PdpStatusParameters;
-import org.onap.policy.apex.starter.parameters.PolicyTypeIdentParameters;
+import org.onap.policy.apex.starter.parameters.ToscaPolicyTypeIdentifierParameters;
 import org.onap.policy.common.utils.services.Registry;
+import org.onap.policy.models.pdp.concepts.PdpResponseDetails;
 import org.onap.policy.models.pdp.concepts.PdpStatus;
-import org.onap.policy.models.pdp.concepts.PolicyTypeIdent;
 import org.onap.policy.models.pdp.enums.PdpHealthStatus;
+import org.onap.policy.models.pdp.enums.PdpResponseStatus;
 import org.onap.policy.models.pdp.enums.PdpState;
+import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicy;
+import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicyIdentifier;
+import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicyTypeIdentifier;
 
 /**
  * This class supports the handling of pdp messages.
@@ -50,25 +54,35 @@ public class PdpMessageHandler {
     public PdpStatus createPdpStatusFromParameters(final String instanceId,
             final PdpStatusParameters pdpStatusParameters) {
         final PdpStatus pdpStatus = new PdpStatus();
-        pdpStatus.setName(pdpStatusParameters.getPdpName());
-        pdpStatus.setVersion(pdpStatusParameters.getVersion());
         pdpStatus.setPdpType(pdpStatusParameters.getPdpType());
         pdpStatus.setState(PdpState.PASSIVE);
         pdpStatus.setHealthy(PdpHealthStatus.HEALTHY);
         pdpStatus.setDescription(pdpStatusParameters.getDescription());
-        pdpStatus.setInstance(instanceId);
-        final List<PolicyTypeIdent> supportedPolicyTypes = new ArrayList<PolicyTypeIdent>();
-        for (final PolicyTypeIdentParameters policyTypeIdentParameters : pdpStatusParameters
+        pdpStatus.setName(instanceId);
+        pdpStatus.setSupportedPolicyTypes(getSupportedPolicyTypesFromParameters(pdpStatusParameters));
+        return pdpStatus;
+    }
+
+    /**
+     * Method to get supported policy types from the parameters.
+     *
+     * @param pdpStatusParameters
+     * @return list of PolicyTypeIdent
+     */
+    private List<ToscaPolicyTypeIdentifier> getSupportedPolicyTypesFromParameters(
+            final PdpStatusParameters pdpStatusParameters) {
+        final List<ToscaPolicyTypeIdentifier> supportedPolicyTypes =
+                new ArrayList<>(pdpStatusParameters.getSupportedPolicyTypes().size());
+        for (final ToscaPolicyTypeIdentifierParameters policyTypeIdentParameters : pdpStatusParameters
                 .getSupportedPolicyTypes()) {
-            supportedPolicyTypes.add(
-                    new PolicyTypeIdent(policyTypeIdentParameters.getName(), policyTypeIdentParameters.getVersion()));
+            supportedPolicyTypes.add(new ToscaPolicyTypeIdentifier(policyTypeIdentParameters.getName(),
+                    policyTypeIdentParameters.getVersion()));
         }
-        pdpStatus.setSupportedPolicyTypes(supportedPolicyTypes);
-        return pdpStatus;
+        return supportedPolicyTypes;
     }
 
     /**
-     * Method to create PdpStatus message from the context, which is to be sent by apex-pdp to pap
+     * Method to create PdpStatus message from the context, which is to be sent by apex-pdp to pap.
      *
      * @return PdpStatus the pdp status message
      */
@@ -76,13 +90,67 @@ public class PdpMessageHandler {
         final PdpStatus pdpStatusContext = Registry.get(ApexStarterConstants.REG_PDP_STATUS_OBJECT, PdpStatus.class);
         final PdpStatus pdpStatus = new PdpStatus();
         pdpStatus.setName(pdpStatusContext.getName());
-        pdpStatus.setVersion(pdpStatusContext.getVersion());
         pdpStatus.setPdpType(pdpStatusContext.getPdpType());
         pdpStatus.setState(pdpStatusContext.getState());
         pdpStatus.setHealthy(pdpStatusContext.getHealthy());
         pdpStatus.setDescription(pdpStatusContext.getDescription());
-        pdpStatus.setInstance(pdpStatusContext.getInstance());
         pdpStatus.setSupportedPolicyTypes(pdpStatusContext.getSupportedPolicyTypes());
+        pdpStatus.setPolicies(pdpStatusContext.getPolicies());
+        if (null != pdpStatusContext.getPdpGroup()) {
+            pdpStatus.setPdpGroup(pdpStatusContext.getPdpGroup());
+        }
+        if (null != pdpStatusContext.getPdpSubgroup()) {
+            pdpStatus.setPdpSubgroup(pdpStatusContext.getPdpSubgroup());
+        }
         return pdpStatus;
     }
+
+    /**
+     * Method to get a final pdp status when the apex started is shutting down.
+     *
+     * @param policies list of ToscaPolicy
+     *
+     * @return PdpStatus
+     */
+    public PdpStatus getTerminatedPdpStatus() {
+        final PdpStatus pdpStatusInContext = Registry.get(ApexStarterConstants.REG_PDP_STATUS_OBJECT, PdpStatus.class);
+        pdpStatusInContext.setState(PdpState.TERMINATED);
+        pdpStatusInContext.setDescription("Apex pdp shutting down.");
+        return createPdpStatusFromContext();
+    }
+
+    /**
+     * Method create PdpResponseDetails which will be sent as part of pdp status to PAP.
+     *
+     * @param requestId request id of the PdpUpdate message from pap
+     * @param status response status to be sent back
+     * @param responseMessage response message to be sent back
+     *
+     * @return PdpResponseDetails
+     */
+    public PdpResponseDetails createPdpResonseDetails(final String requestId, final PdpResponseStatus status,
+            final String responseMessage) {
+        final PdpResponseDetails pdpResponseDetails = new PdpResponseDetails();
+        pdpResponseDetails.setResponseTo(requestId);
+        pdpResponseDetails.setResponseStatus(status);
+        pdpResponseDetails.setResponseMessage(responseMessage);
+        return pdpResponseDetails;
+    }
+
+    /**
+     * Method to retrieve list of ToscaPolicyIdentifier from the list of ToscaPolicy.
+     *
+     * @param policies list of ToscaPolicy
+     *
+     * @return policyTypeIdentifiers
+     */
+    public List<ToscaPolicyIdentifier> getToscaPolicyIdentifiers(final List<ToscaPolicy> policies) {
+        final List<ToscaPolicyIdentifier> policyIdentifiers = new ArrayList<>(policies.size());
+        for (final ToscaPolicy policy : policies) {
+            if (null != policy.getName() && null != policy.getVersion()) {
+                policyIdentifiers.add(new ToscaPolicyIdentifier(policy.getName(), policy.getVersion()));
+            }
+        }
+        return policyIdentifiers;
+    }
 }
diff --git a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/handler/PdpStateChangeMessageHandler.java b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/handler/PdpStateChangeMessageHandler.java
new file mode 100644
index 000000000..c0ec690d8
--- /dev/null
+++ b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/handler/PdpStateChangeMessageHandler.java
@@ -0,0 +1,185 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2019 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.apex.starter.handler;
+
+import java.util.List;
+
+import org.onap.policy.apex.starter.ApexStarterConstants;
+import org.onap.policy.apex.starter.comm.PdpStatusPublisher;
+import org.onap.policy.apex.starter.engine.ApexEngineHandler;
+import org.onap.policy.apex.starter.exception.ApexStarterException;
+import org.onap.policy.common.utils.services.Registry;
+import org.onap.policy.models.pdp.concepts.PdpResponseDetails;
+import org.onap.policy.models.pdp.concepts.PdpStateChange;
+import org.onap.policy.models.pdp.concepts.PdpStatus;
+import org.onap.policy.models.pdp.enums.PdpResponseStatus;
+import org.onap.policy.models.pdp.enums.PdpState;
+import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicy;
+
+/**
+ * This class supports the handling of pdp state change messages.
+ *
+ * @author Ajith Sreekumar (ajith.sreekumar@est.tech)
+ */
+public class PdpStateChangeMessageHandler {
+
+    /**
+     * Method which handles a pdp state change event from PAP.
+     *
+     * @param pdpStateChangeMsg pdp state change message
+     */
+    public void handlePdpStateChangeEvent(final PdpStateChange pdpStateChangeMsg) {
+        final PdpStatus pdpStatusContext = Registry.get(ApexStarterConstants.REG_PDP_STATUS_OBJECT, PdpStatus.class);
+        final PdpStatusPublisher pdpStatusPublisher = Registry.get(ApexStarterConstants.REG_PDP_STATUS_PUBLISHER);
+        final PdpMessageHandler pdpMessageHandler = new PdpMessageHandler();
+        PdpResponseDetails pdpResponseDetails = null;
+        if (isMessageRelevant(pdpStateChangeMsg, pdpStatusContext)) {
+            switch (pdpStateChangeMsg.getState()) {
+                case TERMINATED:
+                    pdpResponseDetails = handleTerminatedState(pdpStateChangeMsg, pdpStatusContext, pdpMessageHandler);
+                    break;
+                case PASSIVE:
+                    pdpResponseDetails = handlePassiveState(pdpStateChangeMsg, pdpStatusContext, pdpMessageHandler);
+                    break;
+                case ACTIVE:
+                    pdpResponseDetails = handleActiveState(pdpStateChangeMsg, pdpStatusContext, pdpMessageHandler);
+                    break;
+                default:
+                    break;
+            }
+            final PdpStatus pdpStatus = pdpMessageHandler.createPdpStatusFromContext();
+            pdpStatus.setResponse(pdpResponseDetails);
+            pdpStatusPublisher.send(pdpStatus);
+        }
+    }
+
+    /**
+     * Check if the pdp state change message is meant for this pdp.
+     *
+     * @param pdpStateChangeMsg
+     * @param pdpStatusContext
+     * @return boolean value if relevant or not
+     */
+    private boolean isMessageRelevant(final PdpStateChange pdpStateChangeMsg, final PdpStatus pdpStatusContext) {
+        return pdpStatusContext.getName().equals(pdpStateChangeMsg.getName()) || (null == pdpStateChangeMsg.getName()
+                && pdpStateChangeMsg.getPdpGroup().equals(pdpStatusContext.getPdpGroup())
+                && pdpStateChangeMsg.getPdpSubgroup().equals(pdpStatusContext.getPdpSubgroup()));
+    }
+
+    /**
+     * Method to handle when the new state from pap is active.
+     *
+     * @param pdpStateChangeMsg
+     * @param pdpStatusContext
+     * @param pdpMessageHandler
+     * @return pdpResponseDetails
+     */
+    private PdpResponseDetails handleActiveState(final PdpStateChange pdpStateChangeMsg,
+            final PdpStatus pdpStatusContext, final PdpMessageHandler pdpMessageHandler) {
+        PdpResponseDetails pdpResponseDetails = null;
+        if (pdpStatusContext.getState().equals(PdpState.ACTIVE)) {
+            pdpResponseDetails = pdpMessageHandler.createPdpResonseDetails(pdpStateChangeMsg.getRequestId(),
+                    PdpResponseStatus.SUCCESS, "Pdp already in active state");
+        } else {
+            final List<ToscaPolicy> policies = Registry.get(ApexStarterConstants.REG_APEX_TOSCA_POLICY_LIST);
+            if (policies.isEmpty()) {
+                pdpResponseDetails = pdpMessageHandler.createPdpResonseDetails(pdpStateChangeMsg.getRequestId(),
+                        PdpResponseStatus.SUCCESS, "No policies found. Apex engine not running.");
+            } else {
+                try {
+                    // assumed that the apex policies list contains only one entry.
+                    final ApexEngineHandler apexEngineHandler =
+                            new ApexEngineHandler((String) policies.get(0).getProperties().get("content"));
+                    Registry.register(ApexStarterConstants.REG_APEX_ENGINE_HANDLER, apexEngineHandler);
+                    pdpResponseDetails = pdpMessageHandler.createPdpResonseDetails(pdpStateChangeMsg.getRequestId(),
+                            PdpResponseStatus.SUCCESS, "Apex engine started. State changed to active.");
+                    pdpStatusContext.setState(PdpState.ACTIVE);
+                } catch (final ApexStarterException e) {
+                    pdpResponseDetails = pdpMessageHandler.createPdpResonseDetails(pdpStateChangeMsg.getRequestId(),
+                            PdpResponseStatus.FAIL, "Apex engine service running failed. " + e.getMessage());
+                }
+            }
+        }
+        return pdpResponseDetails;
+    }
+
+    /**
+     * Method to handle when the new state from pap is passive.
+     *
+     * @param pdpStateChangeMsg
+     * @param pdpStatusContext
+     * @param pdpMessageHandler
+     * @return pdpResponseDetails
+     */
+    private PdpResponseDetails handlePassiveState(final PdpStateChange pdpStateChangeMsg,
+            final PdpStatus pdpStatusContext, final PdpMessageHandler pdpMessageHandler) {
+        PdpResponseDetails pdpResponseDetails = null;
+        if (pdpStatusContext.getState().equals(PdpState.PASSIVE)) {
+            pdpResponseDetails = pdpMessageHandler.createPdpResonseDetails(pdpStateChangeMsg.getRequestId(),
+                    PdpResponseStatus.SUCCESS, "Pdp already in passive state");
+        } else {
+            final ApexEngineHandler apexEngineHandler = Registry.get(ApexStarterConstants.REG_APEX_ENGINE_HANDLER);
+            try {
+                apexEngineHandler.shutdown();
+                pdpResponseDetails = pdpMessageHandler.createPdpResonseDetails(pdpStateChangeMsg.getRequestId(),
+                        PdpResponseStatus.SUCCESS, "Apex pdp state changed from Active to Passive.");
+                pdpStatusContext.setState(PdpState.PASSIVE);
+            } catch (final Exception e) {
+                pdpResponseDetails = pdpMessageHandler.createPdpResonseDetails(pdpStateChangeMsg.getRequestId(),
+                        PdpResponseStatus.FAIL,
+                        "Stopping apex engine failed. State cannot be changed to Passive." + e.getMessage());
+            }
+        }
+        return pdpResponseDetails;
+    }
+
+    /**
+     * Method to handle when the new state from pap is terminated.
+     *
+     * @param pdpStateChangeMsg
+     * @param pdpStatusPublisher
+     * @param pdpMessageHandler
+     * @return pdpResponseDetails
+     */
+    private PdpResponseDetails handleTerminatedState(final PdpStateChange pdpStateChangeMsg,
+            final PdpStatus pdpStatusContext, final PdpMessageHandler pdpMessageHandler) {
+
+        PdpResponseDetails pdpResponseDetails;
+        if (pdpStatusContext.getState().equals(PdpState.ACTIVE)) {
+            final ApexEngineHandler apexEngineHandler = Registry.get(ApexStarterConstants.REG_APEX_ENGINE_HANDLER);
+            try {
+                apexEngineHandler.shutdown();
+                pdpStatusContext.setState(PdpState.PASSIVE);
+                pdpResponseDetails = pdpMessageHandler.createPdpResonseDetails(pdpStateChangeMsg.getRequestId(),
+                        PdpResponseStatus.SUCCESS, "Apex Engine stopped. State changed to passive.");
+            } catch (final Exception e) {
+                pdpResponseDetails = pdpMessageHandler.createPdpResonseDetails(pdpStateChangeMsg.getRequestId(),
+                        PdpResponseStatus.FAIL,
+                        "Stopping apex engine failed. State cannot be changed." + e.getMessage());
+            }
+        } else {
+            pdpResponseDetails = pdpMessageHandler.createPdpResonseDetails(pdpStateChangeMsg.getRequestId(),
+                    PdpResponseStatus.SUCCESS, "Pdp already in passive state");
+        }
+        return pdpResponseDetails;
+
+    }
+}
diff --git a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/handler/PdpUpdateMessageHandler.java b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/handler/PdpUpdateMessageHandler.java
new file mode 100644
index 000000000..835c2993a
--- /dev/null
+++ b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/handler/PdpUpdateMessageHandler.java
@@ -0,0 +1,125 @@
+/*-
+ * ============LICENSE_START=======================================================
+ *  Copyright (C) 2019 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.apex.starter.handler;
+
+import java.util.List;
+
+import org.onap.policy.apex.starter.ApexStarterConstants;
+import org.onap.policy.apex.starter.comm.PdpStatusPublisher;
+import org.onap.policy.apex.starter.engine.ApexEngineHandler;
+import org.onap.policy.apex.starter.exception.ApexStarterException;
+import org.onap.policy.common.endpoints.event.comm.TopicSink;
+import org.onap.policy.common.utils.services.Registry;
+import org.onap.policy.models.pdp.concepts.PdpResponseDetails;
+import org.onap.policy.models.pdp.concepts.PdpStatus;
+import org.onap.policy.models.pdp.concepts.PdpUpdate;
+import org.onap.policy.models.pdp.enums.PdpResponseStatus;
+
+/**
+ * This class supports the handling of pdp update messages.
+ *
+ * @author Ajith Sreekumar (ajith.sreekumar@est.tech)
+ */
+public class PdpUpdateMessageHandler {
+
+
+    /**
+     * Method which handles a pdp update event from PAP.
+     *
+     * @param pdpUpdateMsg pdp update message
+     */
+    public void handlePdpUpdateEvent(final PdpUpdate pdpUpdateMsg) {
+        final PdpMessageHandler pdpMessageHandler = new PdpMessageHandler();
+        final PdpStatus pdpStatusContext = Registry.get(ApexStarterConstants.REG_PDP_STATUS_OBJECT, PdpStatus.class);
+        PdpResponseDetails pdpResponseDetails;
+        if (pdpStatusContext.getName().equals(pdpUpdateMsg.getName())) {
+            final PdpStatusPublisher pdpStatusPublisher = Registry.get(ApexStarterConstants.REG_PDP_STATUS_PUBLISHER);
+            if (checkIfAlreadyHandled(pdpUpdateMsg, pdpStatusContext, pdpStatusPublisher.getInterval())) {
+                pdpResponseDetails = pdpMessageHandler.createPdpResonseDetails(pdpUpdateMsg.getRequestId(),
+                        PdpResponseStatus.SUCCESS, "Pdp already updated");
+            } else {
+                if (null != pdpUpdateMsg.getPdpHeartbeatIntervalMs() && pdpUpdateMsg.getPdpHeartbeatIntervalMs() > 0
+                        && pdpStatusPublisher.getInterval() != pdpUpdateMsg.getPdpHeartbeatIntervalMs()) {
+                    updateInterval(pdpUpdateMsg.getPdpHeartbeatIntervalMs());
+                }
+                pdpStatusContext.setPdpGroup(pdpUpdateMsg.getPdpGroup());
+                pdpStatusContext.setPdpSubgroup(pdpUpdateMsg.getPdpSubgroup());
+                pdpStatusContext
+                        .setPolicies(new PdpMessageHandler().getToscaPolicyIdentifiers(pdpUpdateMsg.getPolicies()));
+                if (pdpUpdateMsg.getPolicies().isEmpty()) {
+                    final ApexEngineHandler apexEngineHandler =
+                            Registry.get(ApexStarterConstants.REG_APEX_ENGINE_HANDLER);
+                    if (apexEngineHandler.isApexEngineRunning()) {
+                        try {
+                            apexEngineHandler.shutdown();
+                        } catch (final ApexStarterException e) {
+                            pdpResponseDetails = pdpMessageHandler.createPdpResonseDetails(pdpUpdateMsg.getRequestId(),
+                                    PdpResponseStatus.FAIL,
+                                    "Pdp update failed as the policies couldn't be undeployed.");
+                        }
+                    }
+                }
+                Registry.registerOrReplace(ApexStarterConstants.REG_APEX_TOSCA_POLICY_LIST, pdpUpdateMsg.getPolicies());
+                pdpResponseDetails = pdpMessageHandler.createPdpResonseDetails(pdpUpdateMsg.getRequestId(),
+                        PdpResponseStatus.SUCCESS, "Pdp update successful.");
+            }
+            final PdpStatusPublisher pdpStatusPublisherTemp =
+                    Registry.get(ApexStarterConstants.REG_PDP_STATUS_PUBLISHER);
+            final PdpStatus pdpStatus = pdpMessageHandler.createPdpStatusFromContext();
+            pdpStatus.setResponse(pdpResponseDetails);
+            pdpStatusPublisherTemp.send(pdpStatus);
+        }
+    }
+
+    /**
+     * Method checks of the Pdp update message is already handled by checking the values in the context.
+     *
+     * @param pdpUpdateMsg pdp update message received from pap
+     * @param pdpStatusContext values saved in context memory
+     * @param interval the current interval in which the pdp status publisher is sending the heartbeats
+     * @return boolean flag which tells if the information is same or not
+     */
+    private boolean checkIfAlreadyHandled(final PdpUpdate pdpUpdateMsg, final PdpStatus pdpStatusContext,
+            final long interval) {
+        return null != pdpStatusContext.getPdpGroup()
+                && pdpStatusContext.getPdpGroup().equals(pdpUpdateMsg.getPdpGroup())
+                && null != pdpStatusContext.getPdpSubgroup()
+                && pdpStatusContext.getPdpSubgroup().equals(pdpUpdateMsg.getPdpSubgroup())
+                && null != pdpStatusContext.getPolicies()
+                && pdpStatusContext.getPolicies()
+                        .containsAll(new PdpMessageHandler().getToscaPolicyIdentifiers(pdpUpdateMsg.getPolicies()))
+                && null != pdpUpdateMsg.getPdpHeartbeatIntervalMs() && pdpUpdateMsg.getPdpHeartbeatIntervalMs() > 0
+                && interval == pdpUpdateMsg.getPdpHeartbeatIntervalMs();
+    }
+
+    /**
+     * Method to update the time interval used by the timer task.
+     *
+     * @param interval time interval received in the pdp update message from pap
+     */
+    public void updateInterval(final long interval) {
+        final PdpStatusPublisher pdpStatusPublisher = Registry.get(ApexStarterConstants.REG_PDP_STATUS_PUBLISHER);
+        pdpStatusPublisher.terminate();
+        final List<TopicSink> topicSinks = Registry.get(ApexStarterConstants.REG_APEX_PDP_TOPIC_SINKS);
+        Registry.registerOrReplace(ApexStarterConstants.REG_PDP_STATUS_PUBLISHER,
+                new PdpStatusPublisher(topicSinks, interval));
+    }
+}
diff --git a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/parameters/PdpStatusParameters.java b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/parameters/PdpStatusParameters.java
index 64545492d..fbc94193d 100644
--- a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/parameters/PdpStatusParameters.java
+++ b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/parameters/PdpStatusParameters.java
@@ -40,13 +40,11 @@ import org.onap.policy.common.parameters.annotations.NotNull;
 public class PdpStatusParameters extends ParameterGroupImpl {
 
     @Min(value = 1)
-    private int timeInterval;
+    private long timeIntervalMs;
 
-    private String pdpName;
-    private String version;
     private String pdpType;
     private String description;
-    private List<PolicyTypeIdentParameters> supportedPolicyTypes;
+    private List<ToscaPolicyTypeIdentifierParameters> supportedPolicyTypes;
 
     public PdpStatusParameters() {
         super(PdpStatusParameters.class.getSimpleName());
diff --git a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/parameters/PolicyTypeIdentParameters.java b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/parameters/ToscaPolicyTypeIdentifierParameters.java
index b6760a023..37e27492e 100644
--- a/services/services-onappf/src/main/java/org/onap/policy/apex/starter/parameters/PolicyTypeIdentParameters.java
+++ b/services/services-onappf/src/main/java/org/onap/policy/apex/starter/parameters/ToscaPolicyTypeIdentifierParameters.java
@@ -37,7 +37,7 @@ import org.onap.policy.common.parameters.annotations.NotNull;
 @Getter
 @Setter
 @EqualsAndHashCode
-public class PolicyTypeIdentParameters {
+public class ToscaPolicyTypeIdentifierParameters {
     private String name;
     private String version;
 }