Merge "Refactor request map"

23 files changed, 2894 insertions, 1834 deletions

diff --git a/main/src/main/java/org/onap/policy/pap/main/comm/PdpModifyRequestMap.java b/main/src/main/java/org/onap/policy/pap/main/comm/PdpModifyRequestMap.java
index 24443cc2..6a743a31 100644
--- a/main/src/main/java/org/onap/policy/pap/main/comm/PdpModifyRequestMap.java
+++ b/main/src/main/java/org/onap/policy/pap/main/comm/PdpModifyRequestMap.java
@@ -20,26 +20,43 @@
 
 package org.onap.policy.pap.main.comm;
 
-import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
 import java.util.Map;
+import org.onap.policy.models.base.PfModelException;
+import org.onap.policy.models.pdp.concepts.Pdp;
+import org.onap.policy.models.pdp.concepts.PdpGroup;
+import org.onap.policy.models.pdp.concepts.PdpGroupFilter;
+import org.onap.policy.models.pdp.concepts.PdpMessage;
 import org.onap.policy.models.pdp.concepts.PdpStateChange;
+import org.onap.policy.models.pdp.concepts.PdpSubGroup;
 import org.onap.policy.models.pdp.concepts.PdpUpdate;
-import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicy;
-import org.onap.policy.pap.main.comm.msgdata.StateChangeData;
-import org.onap.policy.pap.main.comm.msgdata.UpdateData;
+import org.onap.policy.models.pdp.enums.PdpState;
+import org.onap.policy.models.provider.PolicyModelsProvider;
+import org.onap.policy.pap.main.PolicyModelsProviderFactoryWrapper;
+import org.onap.policy.pap.main.comm.msgdata.Request;
+import org.onap.policy.pap.main.comm.msgdata.RequestListener;
+import org.onap.policy.pap.main.comm.msgdata.StateChangeReq;
+import org.onap.policy.pap.main.comm.msgdata.UpdateReq;
 import org.onap.policy.pap.main.parameters.PdpModifyRequestMapParams;
+import org.onap.policy.pap.main.parameters.RequestParams;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Maps a PDP name to requests that modify PDPs.
  */
 public class PdpModifyRequestMap {
+    private static final Logger logger = LoggerFactory.getLogger(PdpModifyRequestMap.class);
+
+    private static final String UNEXPECTED_BROADCAST = "unexpected broadcast message: ";
 
     /**
-     * Maps a PDP name to its request data. An entry is removed once all of the requests
-     * within the data have been completed.
+     * Maps a PDP name to its outstanding requests.
      */
-    private final Map<String, ModifyReqData> name2data = new HashMap<>();
+    private final Map<String, PdpRequests> pdp2requests = new HashMap<>();
 
     /**
      * PDP modification lock.
@@ -52,7 +69,13 @@ public class PdpModifyRequestMap {
     private final PdpModifyRequestMapParams params;
 
     /**
-     * Constructs the data.
+     * Factory for PAP DAO.
+     */
+    private final PolicyModelsProviderFactoryWrapper daoFactory;
+
+
+    /**
+     * Constructs the object.
      *
      * @param params configuration parameters
      *
@@ -63,24 +86,21 @@ public class PdpModifyRequestMap {
 
         this.params = params;
         this.modifyLock = params.getModifyLock();
+        this.daoFactory = params.getDaoFactory();
     }
 
     /**
-     * Adds an UPDATE request to the map.
-     *
-     * @param update the UPDATE request or {@code null}
-     */
-    public void addRequest(PdpUpdate update) {
-        addRequest(update, null);
-    }
-
-    /**
-     * Adds STATE-CHANGE request to the map.
+     * Stops publishing requests to the given PDP.
      *
-     * @param stateChange the STATE-CHANGE request or {@code null}
+     * @param pdpName PDP name
      */
-    public void addRequest(PdpStateChange stateChange) {
-        addRequest(null, stateChange);
+    public void stopPublishing(String pdpName) {
+        synchronized (modifyLock) {
+            PdpRequests requests = pdp2requests.remove(pdpName);
+            if (requests != null) {
+                requests.stopPublishing();
+            }
+        }
     }
 
     /**
@@ -90,288 +110,255 @@ public class PdpModifyRequestMap {
      * @param stateChange the STATE-CHANGE request or {@code null}
      */
     public void addRequest(PdpUpdate update, PdpStateChange stateChange) {
-        if (update == null && stateChange == null) {
-            return;
-        }
+        if (update == null) {
+            addRequest(stateChange);
 
-        synchronized (modifyLock) {
-            String pdpName = getPdpName(update, stateChange);
-
-            ModifyReqData data = name2data.get(pdpName);
-            if (data != null) {
-                // update the existing request
-                data.add(update);
-                data.add(stateChange);
-
-            } else {
-                data = makeRequestData(update, stateChange);
-                name2data.put(pdpName, data);
-                data.startPublishing();
+        } else {
+            synchronized (modifyLock) {
+                addRequest(update);
+                addRequest(stateChange);
             }
         }
     }
 
     /**
-     * Gets the PDP name from two requests.
+     * Adds an UPDATE request to the map.
      *
-     * @param update the update request, or {@code null}
-     * @param stateChange the state-change request, or {@code null}
-     * @return the PDP name, or {@code null} if both requests are {@code null}
+     * @param update the UPDATE request or {@code null}
      */
-    private static String getPdpName(PdpUpdate update, PdpStateChange stateChange) {
-        String pdpName;
+    public void addRequest(PdpUpdate update) {
+        if (update == null) {
+            return;
+        }
 
-        if (update != null) {
-            if ((pdpName = update.getName()) == null) {
-                throw new IllegalArgumentException("missing name in " + update);
-            }
+        if (isBroadcast(update)) {
+            throw new IllegalArgumentException(UNEXPECTED_BROADCAST + update);
+        }
 
-            if (stateChange != null && !pdpName.equals(stateChange.getName())) {
-                throw new IllegalArgumentException(
-                                "name " + stateChange.getName() + " does not match " + pdpName + " " + stateChange);
-            }
+        // @formatter:off
+        RequestParams reqparams = new RequestParams()
+            .setMaxRetryCount(params.getParams().getUpdateParameters().getMaxRetryCount())
+            .setTimers(params.getUpdateTimers())
+            .setModifyLock(params.getModifyLock())
+            .setPublisher(params.getPublisher())
+            .setResponseDispatcher(params.getResponseDispatcher());
+        // @formatter:on
 
-        } else {
-            if ((pdpName = stateChange.getName()) == null) {
-                throw new IllegalArgumentException("missing name in " + stateChange);
-            }
-        }
+        String name = update.getName() + " " + PdpUpdate.class.getSimpleName();
+        UpdateReq request = new UpdateReq(reqparams, name, update);
 
-        return pdpName;
+        addSingleton(request);
     }
 
     /**
-     * Determines if two requests are the "same", which is does not necessarily mean
-     * "equals".
+     * Adds a STATE-CHANGE request to the map.
      *
-     * @param first first request to check
-     * @param second second request to check
-     * @return {@code true} if the requests are the "same", {@code false} otherwise
+     * @param stateChange the STATE-CHANGE request or {@code null}
      */
-    protected static boolean isSame(PdpUpdate first, PdpUpdate second) {
-        if (first.getPolicies().size() != second.getPolicies().size()) {
-            return false;
+    public void addRequest(PdpStateChange stateChange) {
+        if (stateChange == null) {
+            return;
         }
 
-        if (!first.getPdpGroup().equals(second.getPdpGroup())) {
-            return false;
+        if (isBroadcast(stateChange)) {
+            throw new IllegalArgumentException(UNEXPECTED_BROADCAST + stateChange);
         }
 
-        if (!first.getPdpSubgroup().equals(second.getPdpSubgroup())) {
-            return false;
-        }
+        // @formatter:off
+        RequestParams reqparams = new RequestParams()
+            .setMaxRetryCount(params.getParams().getStateChangeParameters().getMaxRetryCount())
+            .setTimers(params.getStateChangeTimers())
+            .setModifyLock(params.getModifyLock())
+            .setPublisher(params.getPublisher())
+            .setResponseDispatcher(params.getResponseDispatcher());
+        // @formatter:on
 
-        // see if the other has any policies that this does not have
-        ArrayList<ToscaPolicy> lst = new ArrayList<>(second.getPolicies());
-        lst.removeAll(first.getPolicies());
+        String name = stateChange.getName() + " " + PdpStateChange.class.getSimpleName();
+        StateChangeReq request = new StateChangeReq(reqparams, name, stateChange);
 
-        return lst.isEmpty();
+        addSingleton(request);
     }
 
     /**
-     * Determines if two requests are the "same", which is does not necessarily mean
-     * "equals".
+     * Determines if a message is a broadcast message.
      *
-     * @param first first request to check
-     * @param second second request to check
-     * @return {@code true} if this update subsumes the other, {@code false} otherwise
+     * @param message the message to examine
+     * @return {@code true} if the message is a broadcast message, {@code false} if
+     *         destined for a single PDP
      */
-    protected static boolean isSame(PdpStateChange first, PdpStateChange second) {
-        return (first.getState() == second.getState());
+    private boolean isBroadcast(PdpMessage message) {
+        return (message.getName() == null);
     }
 
     /**
-     * Request data, which contains an UPDATE or a STATE-CHANGE request, or both. The
-     * UPDATE is always published before the STATE-CHANGE. In addition, both requests may
-     * be changed at any point, possibly triggering a restart of the publishing.
+     * Configures and adds a request to the map.
+     *
+     * @param request the request to be added
      */
-    public class ModifyReqData extends RequestData {
-
-        /**
-         * The UPDATE message to be published, or {@code null}.
-         */
-        private PdpUpdate update;
-
-        /**
-         * The STATE-CHANGE message to be published, or {@code null}.
-         */
-        private PdpStateChange stateChange;
-
-
-        /**
-         * Constructs the object.
-         *
-         * @param newUpdate the UPDATE message to be sent, or {@code null}
-         * @param newState the STATE-CHANGE message to be sent, or {@code null}
-         */
-        public ModifyReqData(PdpUpdate newUpdate, PdpStateChange newState) {
-            super(params);
-
-            if (newUpdate != null) {
-                this.stateChange = newState;
-                setName(newUpdate.getName());
-                update = newUpdate;
-                configure(new ModUpdateData(newUpdate));
-
-            } else {
-                this.update = null;
-                setName(newState.getName());
-                stateChange = newState;
-                configure(new ModStateChangeData(newState));
-            }
-        }
+    private void addSingleton(Request request) {
 
-        /**
-         * Determines if this request is still in the map.
-         */
-        @Override
-        protected boolean isActive() {
-            return (name2data.get(getName()) == this);
+        synchronized (modifyLock) {
+            PdpRequests requests = pdp2requests.computeIfAbsent(request.getMessage().getName(), this::makePdpRequests);
+
+            request.setListener(new SingletonListener(requests, request));
+            requests.addSingleton(request);
         }
+    }
 
-        /**
-         * Removes this request from the map.
-         */
-        @Override
-        protected void allCompleted() {
-            name2data.remove(getName(), this);
+    /**
+     * Starts the next request associated with a PDP.
+     *
+     * @param requests current set of requests
+     * @param request the request that just completed
+     */
+    private void startNextRequest(PdpRequests requests, Request request) {
+        if (!requests.startNextRequest(request)) {
+            pdp2requests.remove(requests.getPdpName(), requests);
         }
+    }
 
-        /**
-         * Adds an UPDATE to the request data, replacing any existing UPDATE, if
-         * appropriate. If the UPDATE is replaced, then publishing is restarted.
-         *
-         * @param newRequest the new UPDATE request
-         */
-        private void add(PdpUpdate newRequest) {
-            if (newRequest == null) {
-                return;
-            }
+    /**
+     * Disables a PDP by removing it from its subgroup and then sending it a PASSIVE
+     * request.
+     *
+     * @param requests the requests associated with the PDP to be disabled
+     */
+    private void disablePdp(PdpRequests requests) {
 
-            synchronized (modifyLock) {
-                if (update != null && isSame(update, newRequest)) {
-                    // already have this update - discard it
-                    return;
-                }
+        // remove the requests from the map
+        if (!pdp2requests.remove(requests.getPdpName(), requests)) {
+            // don't have the info we need to disable it
+            logger.warn("no requests with which to disable {}", requests.getPdpName());
+            return;
+        }
 
-                // must restart from scratch
-                stopPublishing();
+        logger.warn("disabling {}", requests.getPdpName());
 
-                update = newRequest;
-                configure(new ModUpdateData(newRequest));
+        requests.stopPublishing();
 
-                startPublishing();
-            }
+        // don't do anything if we don't have a group
+        String name = requests.getLastGroupName();
+        if (name == null) {
+            logger.warn("no group with which to disable {}", requests.getPdpName());
+            return;
         }
 
-        /**
-         * Adds a STATE-CHANGE to the request data, replacing any existing UPDATE, if
-         * appropriate. If the STATE-CHANGE is replaced, and we're currently publishing
-         * the STATE-CHANGE, then publishing is restarted.
-         *
-         * @param newRequest the new STATE-CHANGE request
-         */
-        private void add(PdpStateChange newRequest) {
-            if (newRequest == null) {
+        // remove the PDP from the group
+        removeFromGroup(requests.getPdpName(), name);
+
+        // send the state change
+        PdpStateChange change = new PdpStateChange();
+        change.setName(requests.getPdpName());
+        change.setState(PdpState.PASSIVE);
+        addRequest(change);
+    }
+
+    /**
+     * Removes a PDP from its group.
+     *
+     * @param pdpName name of the PDP to be removed
+     * @param groupName name of the group from which it should be removed
+     */
+    private void removeFromGroup(String pdpName, String groupName) {
+
+        try (PolicyModelsProvider dao = daoFactory.create()) {
+
+            PdpGroupFilter filter = PdpGroupFilter.builder().name(groupName).groupState(PdpState.ACTIVE)
+                            .version(PdpGroupFilter.LATEST_VERSION).build();
+
+            List<PdpGroup> groups = dao.getFilteredPdpGroups(filter);
+            if (groups.isEmpty()) {
                 return;
             }
 
-            synchronized (modifyLock) {
-                if (stateChange != null && isSame(stateChange, newRequest)) {
-                    // already have this update - discard it
+            PdpGroup group = groups.get(0);
+
+            for (PdpSubGroup subgrp : group.getPdpSubgroups()) {
+                if (removeFromSubgroup(pdpName, group, subgrp)) {
+                    dao.updatePdpGroups(Collections.singletonList(group));
                     return;
                 }
+            }
 
-                if (getWrapper() instanceof StateChangeData) {
-                    // we were publishing STATE-CHANGE, thus must restart it
-                    stopPublishing();
+        } catch (PfModelException e) {
+            logger.info("unable to remove PDP {} from subgroup", pdpName, e);
+        }
+    }
 
-                    stateChange = newRequest;
-                    configure(new ModStateChangeData(newRequest));
+    /**
+     * Removes a PDP from a subgroup.
+     *
+     * @param pdpName name of the PDP to be removed
+     * @param group group from which to attempt to remove the PDP
+     * @param subgrp subgroup from which to attempt to remove the PDP
+     * @return {@code true} if the PDP was removed, {@code false} if the PDP was not in
+     *         the group
+     * @throws PfModelException if a DB error occurs
+     */
+    private boolean removeFromSubgroup(String pdpName, PdpGroup group, PdpSubGroup subgrp) throws PfModelException {
 
-                    startPublishing();
+        Iterator<Pdp> iter = subgrp.getPdpInstances().iterator();
 
-                } else {
-                    // haven't started publishing STATE-CHANGE yet, just replace it
-                    stateChange = newRequest;
-                }
+        while (iter.hasNext()) {
+            Pdp instance = iter.next();
+
+            if (pdpName.equals(instance.getInstanceId())) {
+                logger.info("removed {} from group={} version={} subgroup={}", pdpName, group.getName(),
+                                group.getVersion(), subgrp.getPdpType());
+                iter.remove();
+                subgrp.setCurrentInstanceCount(subgrp.getPdpInstances().size());
+                return true;
             }
         }
 
-        /**
-         * Indicates that the retry count was exhausted.
-         */
-        protected void retryCountExhausted() {
-            // remove this request data from the PDP request map
-            allCompleted();
+        return false;
+    }
 
-            // TODO what to do?
-        }
+    /**
+     * Creates a new set of requests for a PDP. May be overridden by junit tests.
+     *
+     * @param pdpName PDP name
+     * @return a new set of requests
+     */
+    protected PdpRequests makePdpRequests(String pdpName) {
+        return new PdpRequests(pdpName);
+    }
 
-        /**
-         * Indicates that a response did not match the data.
-         *
-         * @param reason the reason for the mismatch
-         */
-        protected void mismatch(String reason) {
-            // remove this request data from the PDP request map
-            allCompleted();
+    /**
+     * Listener for singleton request events.
+     */
+    private class SingletonListener implements RequestListener {
+        private final PdpRequests requests;
+        private final Request request;
 
-            // TODO what to do?
+        public SingletonListener(PdpRequests requests, Request request) {
+            this.requests = requests;
+            this.request = request;
         }
 
-        /**
-         * Wraps an UPDATE.
-         */
-        private class ModUpdateData extends UpdateData {
-
-            public ModUpdateData(PdpUpdate message) {
-                super(message, params);
-            }
-
-            @Override
-            public void mismatch(String reason) {
-                ModifyReqData.this.mismatch(reason);
+        @Override
+        public void failure(String pdpName, String reason) {
+            if (requests.getPdpName().equals(pdpName)) {
+                disablePdp(requests);
             }
+        }
 
-            @Override
-            public void completed() {
-                if (stateChange == null) {
-                    // no STATE-CHANGE request - we're done
-                    allCompleted();
+        @Override
+        public void success(String pdpName) {
+            if (requests.getPdpName().equals(pdpName)) {
+                if (pdp2requests.get(requests.getPdpName()) == requests) {
+                    startNextRequest(requests, request);
 
                 } else {
-                    // now process the STATE-CHANGE request
-                    configure(new ModStateChangeData(stateChange));
-                    startPublishing();
+                    logger.info("discard old requests for {}", pdpName);
+                    requests.stopPublishing();
                 }
             }
         }
 
-        /**
-         * Wraps a STATE-CHANGE.
-         */
-        private class ModStateChangeData extends StateChangeData {
-
-            public ModStateChangeData(PdpStateChange message) {
-                super(message, params);
-            }
-
-            @Override
-            public void mismatch(String reason) {
-                ModifyReqData.this.mismatch(reason);
-            }
-
-            @Override
-            public void completed() {
-                allCompleted();
-            }
+        @Override
+        public void retryCountExhausted() {
+            disablePdp(requests);
         }
     }
-
-    // these may be overridden by junit tests
-
-    protected ModifyReqData makeRequestData(PdpUpdate update, PdpStateChange stateChange) {
-        return new ModifyReqData(update, stateChange);
-    }
 }
diff --git a/main/src/main/java/org/onap/policy/pap/main/comm/PdpRequests.java b/main/src/main/java/org/onap/policy/pap/main/comm/PdpRequests.java
new file mode 100644
index 00000000..9fbf36d4
--- /dev/null
+++ b/main/src/main/java/org/onap/policy/pap/main/comm/PdpRequests.java
@@ -0,0 +1,271 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP PAP
+ * ================================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. 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.
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.pap.main.comm;
+
+import lombok.Getter;
+import org.onap.policy.models.pdp.concepts.PdpMessage;
+import org.onap.policy.models.pdp.concepts.PdpUpdate;
+import org.onap.policy.pap.main.comm.msgdata.Request;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tracks requests associated with a particular PDP. Requests may be broadcast requests or
+ * singleton requests (i.e., destined for a single PDP).
+ */
+public class PdpRequests {
+    private static final Logger logger = LoggerFactory.getLogger(PdpRequests.class);
+
+    /**
+     * The maximum request priority + 1.
+     */
+    private static final int MAX_PRIORITY = 2;
+
+    /**
+     * Name of the PDP with which the requests are associated.
+     */
+    @Getter
+    private final String pdpName;
+
+    /**
+     * Index of request currently being published.
+     */
+    private int curIndex = 0;
+
+    /**
+     * Singleton requests. Items may be {@code null}.
+     */
+    private Request[] singletons = new Request[MAX_PRIORITY];
+
+    /**
+     * Last group name to which the associated PDP was assigned.
+     */
+    @Getter
+    private String lastGroupName;
+
+
+    /**
+     * Constructs the object.
+     *
+     * @param pdpName name of the PDP with which the requests are associated
+     */
+    public PdpRequests(String pdpName) {
+        this.pdpName = pdpName;
+    }
+
+    /**
+     * Records the group information from the request.
+     *
+     * @param request the request from which to extract the group information
+     */
+    private void recordGroup(Request request) {
+        PdpMessage message = request.getMessage();
+        if (message instanceof PdpUpdate) {
+            lastGroupName = message.getPdpGroup();
+        }
+    }
+
+    /**
+     * Adds a singleton request.
+     *
+     * @param request the request to be added
+     */
+    public void addSingleton(Request request) {
+
+        if (request.getMessage().getName() == null) {
+            throw new IllegalArgumentException("unexpected broadcast for " + pdpName);
+        }
+
+        recordGroup(request);
+
+        if (checkExisting(request)) {
+            // have an existing request that's similar - discard this request
+            return;
+        }
+
+        // no existing request of this type
+
+        int priority = request.getPriority();
+        singletons[priority] = request;
+
+        // stop publishing anything of a lower priority
+        QueueToken<PdpMessage> token = stopPublishingLowerPriority(priority);
+
+        // start publishing if nothing of higher priority
+        if (higherPrioritySingleton(priority)) {
+            logger.info("{} not publishing due to priority higher than {}", pdpName, priority);
+            return;
+        }
+
+        curIndex = priority;
+        request.startPublishing(token);
+    }
+
+    /**
+     * Checks for an existing request.
+     *
+     * @param request the request of interest
+     * @return {@code true} if a similar request already exists, {@code false} otherwise
+     */
+    private boolean checkExisting(Request request) {
+
+        return checkExistingSingleton(request);
+    }
+
+    /**
+     * Checks for an existing singleton request.
+     *
+     * @param request the request of interest
+     * @return {@code true} if a similar singleton request already exists, {@code false}
+     *         otherwise
+     */
+    private boolean checkExistingSingleton(Request request) {
+
+        Request exsingle = singletons[request.getPriority()];
+
+        if (exsingle == null) {
+            return false;
+        }
+
+        if (exsingle.isSameContent(request)) {
+            // unchanged from existing request
+            logger.info("{} message content unchanged for {}", pdpName, exsingle.getClass().getSimpleName());
+            return true;
+        }
+
+        // reconfigure the existing request
+        PdpMessage message = request.getMessage();
+        exsingle.reconfigure(message, null);
+
+        // still have a singleton in the queue for this request
+        return true;
+    }
+
+    /**
+     * Stops all publishing and removes this PDP from any broadcast messages.
+     */
+    public void stopPublishing() {
+        // stop singletons
+        for (int x = 0; x < MAX_PRIORITY; ++x) {
+            Request single = singletons[x];
+
+            if (single != null) {
+                singletons[x] = null;
+                single.stopPublishing();
+            }
+        }
+    }
+
+    /**
+     * Stops publishing requests of a lower priority than the specified priority.
+     *
+     * @param priority priority of interest
+     * @return the token that was being used to publish a lower priority request
+     */
+    private QueueToken<PdpMessage> stopPublishingLowerPriority(int priority) {
+
+        // stop singletons
+        for (int x = 0; x < priority; ++x) {
+            Request single = singletons[x];
+
+            if (single != null) {
+                logger.info("{} stop publishing priority {}", pdpName, single.getPriority());
+
+                QueueToken<PdpMessage> token = single.stopPublishing(false);
+                if (token != null) {
+                    // found one that was publishing
+                    return token;
+                }
+            }
+        }
+
+        return null;
+    }
+
+    /**
+     * Starts publishing the next request in the queue.
+     *
+     * @param request the request that just completed
+     * @return {@code true} if there is another request in the queue, {@code false} if all
+     *         requests for this PDP have been processed
+     */
+    public boolean startNextRequest(Request request) {
+        if (!zapRequest(curIndex, request)) {
+            // not at curIndex - look for it in other indices
+            for (int x = 0; x < MAX_PRIORITY; ++x) {
+                if (zapRequest(x, request)) {
+                    break;
+                }
+            }
+        }
+
+        // find/start the highest priority request
+        for (curIndex = MAX_PRIORITY - 1; curIndex >= 0; --curIndex) {
+
+            if (singletons[curIndex] != null) {
+                logger.info("{} start publishing priority {}", pdpName, curIndex);
+
+                singletons[curIndex].startPublishing();
+                return true;
+            }
+        }
+
+        logger.info("{} has no more requests", pdpName);
+        curIndex = 0;
+
+        return false;
+    }
+
+    /**
+     * Zaps request pointers, if the request appears at the given index.
+     *
+     * @param index index to examine
+     * @param request request of interest
+     * @return {@code true} if a request pointer was zapped, {@code false} if the request
+     *         did not appear at the given index
+     */
+    private boolean zapRequest(int index, Request request) {
+        if (singletons[index] == request) {
+            singletons[index] = null;
+            return true;
+        }
+
+        return false;
+    }
+
+    /**
+     * Determines if any singleton request, with a higher priority, is associated with the
+     * PDP.
+     *
+     * @param priority priority of interest
+     *
+     * @return {@code true} if the PDP has a singleton, {@code false} otherwise
+     */
+    private boolean higherPrioritySingleton(int priority) {
+        for (int x = priority + 1; x < MAX_PRIORITY; ++x) {
+            if (singletons[x] != null) {
+                return true;
+            }
+        }
+
+        return false;
+    }
+}
diff --git a/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/MessageData.java b/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/MessageData.java
deleted file mode 100644
index aa288f7d..00000000
--- a/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/MessageData.java
+++ /dev/null
@@ -1,104 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * ONAP PAP
- * ================================================================================
- * Copyright (C) 2019 AT&T Intellectual Property. 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.
- * 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.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.policy.pap.main.comm.msgdata;
-
-import org.onap.policy.models.pdp.concepts.PdpMessage;
-import org.onap.policy.models.pdp.concepts.PdpStatus;
-import org.onap.policy.pap.main.comm.TimerManager;
-
-
-/**
- * Wraps a message, providing methods appropriate to the message type.
- */
-public abstract class MessageData {
-    private final PdpMessage message;
-    private final int maxRetries;
-    private final TimerManager timers;
-
-    /**
-     * Constructs the object.
-     *
-     * @param message message to be wrapped by this
-     * @param maxRetries max number of retries
-     * @param timers the timer manager for messages of this type
-     */
-    public MessageData(PdpMessage message, int maxRetries, TimerManager timers) {
-        this.message = message;
-        this.maxRetries = maxRetries;
-        this.timers = timers;
-    }
-
-    /**
-     * Gets the wrapped message.
-     *
-     * @return the wrapped message
-     */
-    public PdpMessage getMessage() {
-        return message;
-    }
-
-    /**
-     * Gets a string, suitable for logging, identifying the message type.
-     *
-     * @return the message type
-     */
-    public String getType() {
-        return message.getClass().getSimpleName();
-    }
-
-    /**
-     * Gets the maximum retry count for the particular message type.
-     *
-     * @return the maximum retry count
-     */
-    public int getMaxRetryCount() {
-        return maxRetries;
-    }
-
-    /**
-     * Gets the timer manager for the particular message type.
-     *
-     * @return the timer manager
-     */
-    public TimerManager getTimers() {
-        return timers;
-    }
-
-    /**
-     * Indicates that the response did not match what was expected.
-     *
-     * @param reason the reason for the mismatch
-     */
-    public abstract void mismatch(String reason);
-
-    /**
-     * Indicates that processing of this particular message has completed successfully.
-     */
-    public abstract void completed();
-
-    /**
-     * Checks the response to ensure it is as expected.
-     *
-     * @param response the response to check
-     * @return an error message, if a fatal error has occurred, {@code null} otherwise
-     */
-    public abstract String checkResponse(PdpStatus response);
-}
diff --git a/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/Request.java b/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/Request.java
new file mode 100644
index 00000000..1f69dcb5
--- /dev/null
+++ b/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/Request.java
@@ -0,0 +1,124 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP PAP
+ * ================================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. 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.
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.pap.main.comm.msgdata;
+
+import org.onap.policy.models.pdp.concepts.PdpMessage;
+import org.onap.policy.models.pdp.concepts.PdpStatus;
+import org.onap.policy.pap.main.comm.QueueToken;
+
+/**
+ * Request data, whose message may be changed at any point, possibly triggering a restart
+ * of the publishing.
+ */
+public interface Request {
+
+    /**
+     * Gets the request priority. Higher priority requests are published before lower
+     * priority requests.
+     *
+     * @return the request priority
+     */
+    public int getPriority();
+
+    /**
+     * Gets the name with which this data is associated, used for logging purposes. This
+     * may be changed when this is reconfigured.
+     *
+     * @return the name with which this data is associated
+     */
+    public String getName();
+
+    /**
+     * Gets the current message.
+     *
+     * @return the current message
+     */
+    public PdpMessage getMessage();
+
+    /**
+     * Sets the listener that will receive request events.
+     *
+     * @param listener the request listener
+     */
+    public void setListener(RequestListener listener);
+
+    /**
+     * Determines if this request is currently being published.
+     *
+     * @return {@code true} if this request is being published, {@code false} otherwise
+     */
+    public boolean isPublishing();
+
+    /**
+     * Starts the publishing process, registering any listeners or timeout handlers, and
+     * adding the request to the publisher queue.
+     */
+    public void startPublishing();
+
+    /**
+     * Starts the publishing process.
+     *
+     * @param token2 token that can be used when publishing, or {@code null} to allocate a
+     *        new token
+     */
+    public void startPublishing(QueueToken<PdpMessage> token2);
+
+    /**
+     * Unregisters the listener, cancels the timer, and removes the message from the
+     * queue.
+     */
+    public void stopPublishing();
+
+    /**
+     * Unregisters the listener and cancels the timer.
+     *
+     * @param removeFromQueue {@code true} if the message should be removed from the
+     *        queue, {@code false} otherwise
+     * @return the token that was being used to publish the message, or {@code null} if
+     *         the request was not being published
+     */
+    public QueueToken<PdpMessage> stopPublishing(boolean removeFromQueue);
+
+    /**
+     * Reconfigures the fields based on the {@link #message} type. Suspends publishing,
+     * updates the configuration, and then resumes publishing.
+     *
+     * @param newMessage the new message
+     * @param token2 token to use when publishing, or {@code null} to allocate a new token
+     */
+    public void reconfigure(PdpMessage newMessage, QueueToken<PdpMessage> token2);
+
+    /**
+     * Checks the response to ensure it is as expected.
+     *
+     * @param response the response to check
+     * @return an error message, if a fatal error has occurred, {@code null} otherwise
+     */
+    public String checkResponse(PdpStatus response);
+
+    /**
+     * Determines if this request has the same content as another request.
+     *
+     * @param other request against which to compare
+     * @return {@code true} if the requests have the same content, {@code false} otherwise
+     */
+    public boolean isSameContent(Request other);
+}
diff --git a/main/src/main/java/org/onap/policy/pap/main/comm/RequestData.java b/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/RequestImpl.java
index 29ad85bc..45ca2db4 100644
--- a/main/src/main/java/org/onap/policy/pap/main/comm/RequestData.java
+++ b/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/RequestImpl.java
@@ -18,7 +18,7 @@
  * ============LICENSE_END=========================================================
  */
 
-package org.onap.policy.pap.main.comm;
+package org.onap.policy.pap.main.comm.msgdata;
 
 import lombok.AccessLevel;
 import lombok.Getter;
@@ -28,54 +28,60 @@ import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
 import org.onap.policy.common.utils.services.ServiceManager;
 import org.onap.policy.models.pdp.concepts.PdpMessage;
 import org.onap.policy.models.pdp.concepts.PdpStatus;
-import org.onap.policy.pap.main.comm.msgdata.MessageData;
-import org.onap.policy.pap.main.parameters.RequestDataParams;
+import org.onap.policy.pap.main.comm.QueueToken;
+import org.onap.policy.pap.main.comm.TimerManager;
+import org.onap.policy.pap.main.parameters.RequestParams;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
- * Request data, the content of which may be changed at any point, possibly triggering a
- * restart of the publishing.
+ * Request data implementation.
  */
-public abstract class RequestData {
-    private static final Logger logger = LoggerFactory.getLogger(RequestData.class);
+public abstract class RequestImpl implements Request {
+    private static final Logger logger = LoggerFactory.getLogger(RequestImpl.class);
 
     /**
      * Name with which this data is associated, used for logging purposes.
      */
     @Getter
-    @Setter(AccessLevel.PROTECTED)
-    private String name;
+    private final String name;
 
     /**
      * The configuration parameters.
      */
-    private final RequestDataParams params;
+    @Getter(AccessLevel.PROTECTED)
+    private final RequestParams params;
 
     /**
-     * Current retry count.
+     * Used to register/unregister the listener and the timer.
      */
-    private int retryCount = 0;
+    private final ServiceManager svcmgr;
 
     /**
-     * Used to register/unregister the listener and the timer.
+     * Handles events associated with the request.
+     */
+    @Setter
+    private RequestListener listener;
+
+    /**
+     * Current retry count.
      */
-    private ServiceManager svcmgr;
+    @Getter
+    private int retryCount = 0;
 
     /**
-     * Wrapper for the message that is currently being published (i.e., {@link #update} or
-     * {@link #stateChange}.
+     * The current message.
      */
-    @Getter(AccessLevel.PROTECTED)
-    private MessageData wrapper;
+    @Getter
+    private PdpMessage message;
 
     /**
-     * Used to cancel a timer.
+     * The currently running timer.
      */
     private TimerManager.Timer timer;
 
     /**
-     * Token that is placed on the queue.
+     * Token that has been placed on the queue.
      */
     private QueueToken<PdpMessage> token = null;
 
@@ -84,72 +90,119 @@ public abstract class RequestData {
      * Constructs the object, and validates the parameters.
      *
      * @param params configuration parameters
+     * @param name the request name, used for logging purposes
+     * @param message the initial message
      *
      * @throws IllegalArgumentException if a required parameter is not set
      */
-    public RequestData(@NonNull RequestDataParams params) {
+    public RequestImpl(@NonNull RequestParams params, @NonNull String name, @NonNull PdpMessage message) {
         params.validate();
 
+        this.name = name;
         this.params = params;
+        this.message = message;
+
+        // @formatter:off
+        this.svcmgr = new ServiceManager(name)
+                        .addAction("listener",
+                            () -> params.getResponseDispatcher()
+                                            .register(this.message.getRequestId(), this::processResponse),
+                            () -> params.getResponseDispatcher().unregister(this.message.getRequestId()))
+                        .addAction("timer",
+                            () -> timer = params.getTimers().register(this.message.getRequestId(), this::handleTimeout),
+                            () -> timer.cancel())
+                        .addAction("enqueue",
+                            () -> enqueue(),
+                            () -> {
+                                // do not remove from the queue - token may be re-used
+                            });
+        // @formatter:on
     }
 
-    /**
-     * Starts the publishing process, registering any listeners or timeout handlers, and
-     * adding the request to the publisher queue. This should not be invoked until after
-     * {@link #configure(MessageData)} is invoked.
-     */
+    @Override
+    public void reconfigure(PdpMessage newMessage, QueueToken<PdpMessage> token2) {
+        if (newMessage.getClass() != message.getClass()) {
+            throw new IllegalArgumentException("expecting " + message.getClass().getSimpleName() + " instead of "
+                            + newMessage.getClass().getSimpleName());
+        }
+
+        logger.info("reconfiguring {} with new message", getName());
+
+        if (svcmgr.isAlive()) {
+            token = stopPublishing(false);
+            message = newMessage;
+            startPublishing(token2);
+
+        } else {
+            message = newMessage;
+        }
+    }
+
+    @Override
+    public boolean isPublishing() {
+        return svcmgr.isAlive();
+    }
+
+    @Override
     public void startPublishing() {
+        startPublishing(null);
+    }
+
+    @Override
+    public void startPublishing(QueueToken<PdpMessage> token2) {
+        if (listener == null) {
+            throw new IllegalStateException("listener has not been set");
+        }
 
         synchronized (params.getModifyLock()) {
-            if (!svcmgr.isAlive()) {
+            replaceToken(token2);
+
+            if (svcmgr.isAlive()) {
+                logger.info("{} is already publishing", getName());
+
+            } else {
+                resetRetryCount();
                 svcmgr.start();
             }
         }
     }
 
     /**
-     * Unregisters the listener and cancels the timer.
+     * Replaces the current token with a new token.
+     * @param newToken the new token
      */
-    protected void stopPublishing() {
-        if (svcmgr.isAlive()) {
-            svcmgr.stop();
+    private void replaceToken(QueueToken<PdpMessage> newToken) {
+        if (newToken != null) {
+            if (token == null) {
+                token = newToken;
+
+            } else if (token != newToken) {
+                // already have a token - discard the new token
+                newToken.replaceItem(null);
+            }
         }
     }
 
-    /**
-     * Configures the fields based on the {@link #message} type.
-     *
-     * @param newWrapper the new message wrapper
-     */
-    protected void configure(MessageData newWrapper) {
-
-        wrapper = newWrapper;
+    @Override
+    public void stopPublishing() {
+        stopPublishing(true);
+    }
 
-        resetRetryCount();
+    @Override
+    public QueueToken<PdpMessage> stopPublishing(boolean removeFromQueue) {
+        if (svcmgr.isAlive()) {
+            svcmgr.stop();
 
-        TimerManager timerManager = wrapper.getTimers();
-        String msgType = wrapper.getType();
-        String reqid = wrapper.getMessage().getRequestId();
+            if (removeFromQueue) {
+                token.replaceItem(null);
+                token = null;
+            }
+        }
 
-        /*
-         * We have to configure the service manager HERE, because it's name changes if the
-         * message class changes.
-         */
+        QueueToken<PdpMessage> tok = token;
+        token = null;
 
-        // @formatter:off
-        this.svcmgr = new ServiceManager(name + " " + msgType)
-                        .addAction("listener",
-                            () -> params.getResponseDispatcher().register(reqid, this::processResponse),
-                            () -> params.getResponseDispatcher().unregister(reqid))
-                        .addAction("timer",
-                            () -> timer = timerManager.register(name, this::handleTimeout),
-                            () -> timer.cancel())
-                        .addAction("enqueue",
-                            () -> enqueue(),
-                            () -> {
-                                // nothing to "stop"
-                            });
-        // @formatter:on
+        return tok;
     }
 
     /**
@@ -157,7 +210,6 @@ public abstract class RequestData {
      * if possible. Otherwise, it adds a new token to the queue.
      */
     private void enqueue() {
-        PdpMessage message = wrapper.getMessage();
         if (token != null && token.replaceItem(message) != null) {
             // took the other's place in the queue - continue using the token
             return;
@@ -171,7 +223,7 @@ public abstract class RequestData {
     /**
      * Resets the retry count.
      */
-    protected void resetRetryCount() {
+    public void resetRetryCount() {
         retryCount = 0;
     }
 
@@ -180,8 +232,8 @@ public abstract class RequestData {
      *
      * @return {@code true} if successful, {@code false} if the limit has been reached
      */
-    protected boolean bumpRetryCount() {
-        if (retryCount >= wrapper.getMaxRetryCount()) {
+    public boolean bumpRetryCount() {
+        if (retryCount >= params.getMaxRetryCount()) {
             return false;
         }
 
@@ -190,15 +242,6 @@ public abstract class RequestData {
     }
 
     /**
-     * Indicates that the retry count was exhausted. The default method simply invokes
-     * {@link #allCompleted()}.
-     */
-    protected void retryCountExhausted() {
-        // remove this request data from the PDP request map
-        allCompleted();
-    }
-
-    /**
      * Processes a response received from the PDP.
      *
      * @param infra infrastructure on which the response was received
@@ -208,26 +251,22 @@ public abstract class RequestData {
     private void processResponse(CommInfrastructure infra, String topic, PdpStatus response) {
 
         synchronized (params.getModifyLock()) {
+            String pdpName = response.getName();
+
             if (!svcmgr.isAlive()) {
                 // this particular request must have been discarded
                 return;
             }
 
-            stopPublishing();
-
-            if (!isActive()) {
-                return;
-            }
-
-            String reason = wrapper.checkResponse(response);
+            String reason = checkResponse(response);
             if (reason != null) {
                 logger.info("{} PDP data mismatch: {}", getName(), reason);
-                wrapper.mismatch(reason);
-
-            } else {
-                logger.info("{} {} successful", getName(), wrapper.getType());
-                wrapper.completed();
+                listener.failure(pdpName, reason);
+                return;
             }
+
+            logger.info("{} successful", getName());
+            listener.success(pdpName);
         }
     }
 
@@ -246,51 +285,36 @@ public abstract class RequestData {
 
             stopPublishing();
 
-            if (!isActive()) {
-                return;
-            }
-
-            if (isInQueue()) {
-                // haven't published yet - just leave it in the queue and reset counts
-                logger.info("{} timeout - request still in the queue", getName());
-                resetRetryCount();
-                startPublishing();
-                return;
-            }
-
             if (!bumpRetryCount()) {
-                logger.info("{} timeout - retry count exhausted", getName());
-                retryCountExhausted();
+                logger.info("{} timeout - retry count {} exhausted", getName(), retryCount);
+                listener.retryCountExhausted();
                 return;
             }
 
             // re-publish
-            logger.info("{} timeout - re-publish", getName());
+            logger.info("{} timeout - re-publish count {}", getName(), retryCount);
+
+            // startPublishing() resets the count, so save & restore it here
+            int count = retryCount;
             startPublishing();
+            retryCount = count;
         }
     }
 
     /**
-     * Determines if the current message is still in the queue. Assumes that
-     * {@link #startPublishing()} has been invoked and thus {@link #token} has been
-     * initialized.
-     *
-     * @return {@code true} if the current message is in the queue, {@code false}
-     *         otherwise
+     * Verifies that the name is not null. Also verifies that it matches the name in the
+     * message, if the message has a name.
      */
-    private boolean isInQueue() {
-        return (token.get() == wrapper.getMessage());
-    }
+    @Override
+    public String checkResponse(PdpStatus response) {
+        if (response.getName() == null) {
+            return "null PDP name";
+        }
 
-    /**
-     * Determines if this request data is still active.
-     *
-     * @return {@code true} if this request is active, {@code false} otherwise
-     */
-    protected abstract boolean isActive();
+        if (message.getName() != null && !message.getName().equals(response.getName())) {
+            return "PDP name does not match";
+        }
 
-    /**
-     * Indicates that this entire request has completed.
-     */
-    protected abstract void allCompleted();
+        return null;
+    }
 }
diff --git a/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/RequestListener.java b/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/RequestListener.java
new file mode 100644
index 00000000..4c53bd64
--- /dev/null
+++ b/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/RequestListener.java
@@ -0,0 +1,47 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP PAP
+ * ================================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. 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.
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.pap.main.comm.msgdata;
+
+/**
+ * Listener for request events.
+ */
+public interface RequestListener {
+
+    /**
+     * Indicates that an invalid response was received from a PDP.
+     *
+     * @param pdpName name of the PDP from which the response was received
+     * @param reason the reason for the mismatch
+     */
+    public void failure(String pdpName, String reason);
+
+    /**
+     * Indicates that a successful response was received from a PDP.
+     *
+     * @param pdpName name of the PDP from which the response was received
+     */
+    public void success(String pdpName);
+
+    /**
+     * Indicates that the retry count was exhausted.
+     */
+    public void retryCountExhausted();
+}
diff --git a/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/StateChangeData.java b/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/StateChangeReq.java
index ecbf5dfa..e38cb0fd 100644
--- a/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/StateChangeData.java
+++ b/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/StateChangeReq.java
@@ -20,38 +20,61 @@
 
 package org.onap.policy.pap.main.comm.msgdata;
 
+import org.onap.policy.models.pdp.concepts.PdpMessage;
 import org.onap.policy.models.pdp.concepts.PdpStateChange;
 import org.onap.policy.models.pdp.concepts.PdpStatus;
-import org.onap.policy.pap.main.parameters.PdpModifyRequestMapParams;
+import org.onap.policy.pap.main.parameters.RequestParams;
 
 /**
  * Wraps a STATE-CHANGE.
  */
-public abstract class StateChangeData extends MessageData {
-    private PdpStateChange stateChange;
+public class StateChangeReq extends RequestImpl {
 
     /**
-     * Constructs the object.
+     * Constructs the object, and validates the parameters.
      *
-     * @param message message to be wrapped by this
-     * @param params the parameters
+     * @param params configuration parameters
+     * @param name the request name, used for logging purposes
+     * @param message the initial message
+     *
+     * @throws IllegalArgumentException if a required parameter is not set
      */
-    public StateChangeData(PdpStateChange message, PdpModifyRequestMapParams params) {
-        super(message, params.getParams().getStateChangeParameters().getMaxRetryCount(), params.getStateChangeTimers());
+    public StateChangeReq(RequestParams params, String name, PdpMessage message) {
+        super(params, name, message);
+    }
 
-        stateChange = message;
+    @Override
+    public PdpStateChange getMessage() {
+        return (PdpStateChange) super.getMessage();
     }
 
     @Override
     public String checkResponse(PdpStatus response) {
-        if (!stateChange.getName().equals(response.getName())) {
-            return "name does not match";
+        String reason = super.checkResponse(response);
+        if (reason != null) {
+            return reason;
         }
 
-        if (response.getState() != stateChange.getState()) {
-            return "state is " + response.getState() + ", but expected " + stateChange.getState();
+        PdpStateChange message = getMessage();
+        if (response.getState() != message.getState()) {
+            return "state is " + response.getState() + ", but expected " + message.getState();
         }
 
         return null;
     }
+
+    @Override
+    public boolean isSameContent(Request other) {
+        if (!(other instanceof StateChangeReq)) {
+            return false;
+        }
+
+        PdpStateChange message = (PdpStateChange) other.getMessage();
+        return (getMessage().getState() == message.getState());
+    }
+
+    @Override
+    public int getPriority() {
+        return 0;
+    }
 }
diff --git a/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/UpdateData.java b/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/UpdateData.java
deleted file mode 100644
index eca0b384..00000000
--- a/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/UpdateData.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * ONAP PAP
- * ================================================================================
- * Copyright (C) 2019 AT&T Intellectual Property. 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.
- * 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.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.policy.pap.main.comm.msgdata;
-
-import java.util.HashSet;
-import java.util.Set;
-import org.onap.policy.models.pdp.concepts.PdpStatus;
-import org.onap.policy.models.pdp.concepts.PdpUpdate;
-import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicy;
-import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicyIdentifier;
-import org.onap.policy.pap.main.parameters.PdpModifyRequestMapParams;
-
-
-/**
- * Wraps an UPDATE.
- */
-public abstract class UpdateData extends MessageData {
-    private PdpUpdate update;
-
-    /**
-     * Constructs the object.
-     *
-     * @param message message to be wrapped by this
-     * @param params the parameters
-     */
-    public UpdateData(PdpUpdate message, PdpModifyRequestMapParams params) {
-        super(message, params.getParams().getUpdateParameters().getMaxRetryCount(), params.getUpdateTimers());
-
-        update = message;
-    }
-
-    @Override
-    public String checkResponse(PdpStatus response) {
-        if (!update.getName().equals(response.getName())) {
-            return "name does not match";
-        }
-
-        if (!update.getPdpGroup().equals(response.getPdpGroup())) {
-            return "group does not match";
-        }
-
-        if (!update.getPdpSubgroup().equals(response.getPdpSubgroup())) {
-            return "subgroup does not match";
-        }
-
-        // see if the other has any policies that this does not have
-        Set<ToscaPolicyIdentifier> set = new HashSet<>(response.getPolicies());
-
-        for (ToscaPolicy policy : update.getPolicies()) {
-            set.remove(policy.getIdentifier());
-        }
-
-        if (!set.isEmpty()) {
-            return "policies do not match";
-        }
-
-        return null;
-    }
-}
diff --git a/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/UpdateReq.java b/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/UpdateReq.java
new file mode 100644
index 00000000..0f6d73fc
--- /dev/null
+++ b/main/src/main/java/org/onap/policy/pap/main/comm/msgdata/UpdateReq.java
@@ -0,0 +1,114 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP PAP
+ * ================================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. 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.
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.pap.main.comm.msgdata;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
+import org.onap.policy.models.pdp.concepts.PdpMessage;
+import org.onap.policy.models.pdp.concepts.PdpStatus;
+import org.onap.policy.models.pdp.concepts.PdpUpdate;
+import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicy;
+import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicyIdentifier;
+import org.onap.policy.pap.main.parameters.RequestParams;
+
+
+/**
+ * Wraps an UPDATE.
+ */
+public class UpdateReq extends RequestImpl {
+
+    /**
+     * Constructs the object, and validates the parameters.
+     *
+     * @param params configuration parameters
+     * @param name the request name, used for logging purposes
+     * @param message the initial message
+     *
+     * @throws IllegalArgumentException if a required parameter is not set
+     */
+    public UpdateReq(RequestParams params, String name, PdpMessage message) {
+        super(params, name, message);
+    }
+
+    @Override
+    public PdpUpdate getMessage() {
+        return (PdpUpdate) super.getMessage();
+    }
+
+    @Override
+    public String checkResponse(PdpStatus response) {
+        String reason = super.checkResponse(response);
+        if (reason != null) {
+            return reason;
+        }
+
+        PdpUpdate message = (PdpUpdate) getMessage();
+        if (!StringUtils.equals(message.getPdpGroup(), response.getPdpGroup())) {
+            return "group does not match";
+        }
+
+        if (!StringUtils.equals(message.getPdpSubgroup(), response.getPdpSubgroup())) {
+            return "subgroup does not match";
+        }
+
+        // see if the policies match
+        Set<ToscaPolicyIdentifier> set1 = new HashSet<>(response.getPolicies());
+        Set<ToscaPolicyIdentifier> set2 = new HashSet<>(
+                        message.getPolicies().stream().map(ToscaPolicy::getIdentifier).collect(Collectors.toSet()));
+
+        if (!set1.equals(set2)) {
+            return "policies do not match";
+        }
+
+        return null;
+    }
+
+    @Override
+    public boolean isSameContent(Request other) {
+        if (!(other instanceof UpdateReq)) {
+            return false;
+        }
+
+        PdpUpdate first = getMessage();
+        PdpUpdate second = (PdpUpdate) other.getMessage();
+
+        if (!StringUtils.equals(first.getPdpGroup(), second.getPdpGroup())) {
+            return false;
+        }
+
+        if (!StringUtils.equals(first.getPdpSubgroup(), second.getPdpSubgroup())) {
+            return false;
+        }
+
+        // see if the policies are the same
+        Set<ToscaPolicy> set1 = new HashSet<>(first.getPolicies());
+        Set<ToscaPolicy> set2 = new HashSet<>(second.getPolicies());
+
+        return set1.equals(set2);
+    }
+
+    @Override
+    public int getPriority() {
+        return 1;
+    }
+}
diff --git a/main/src/main/java/org/onap/policy/pap/main/parameters/PdpModifyRequestMapParams.java b/main/src/main/java/org/onap/policy/pap/main/parameters/PdpModifyRequestMapParams.java
index 2c17a0b2..2f74bf3d 100644
--- a/main/src/main/java/org/onap/policy/pap/main/parameters/PdpModifyRequestMapParams.java
+++ b/main/src/main/java/org/onap/policy/pap/main/parameters/PdpModifyRequestMapParams.java
@@ -23,6 +23,7 @@ package org.onap.policy.pap.main.parameters;
 import lombok.Getter;
 import org.onap.policy.common.endpoints.listeners.RequestIdDispatcher;
 import org.onap.policy.models.pdp.concepts.PdpStatus;
+import org.onap.policy.pap.main.PolicyModelsProviderFactoryWrapper;
 import org.onap.policy.pap.main.comm.Publisher;
 import org.onap.policy.pap.main.comm.TimerManager;
 
@@ -31,10 +32,14 @@ import org.onap.policy.pap.main.comm.TimerManager;
  * Parameters needed to create a {@link PdpModifyRequestMapParams}.
  */
 @Getter
-public class PdpModifyRequestMapParams extends RequestDataParams {
+public class PdpModifyRequestMapParams {
+    private Publisher publisher;
+    private RequestIdDispatcher<PdpStatus> responseDispatcher;
+    private Object modifyLock;
     private PdpParameters params;
     private TimerManager updateTimers;
     private TimerManager stateChangeTimers;
+    private PolicyModelsProviderFactoryWrapper daoFactory;
 
     public PdpModifyRequestMapParams setParams(PdpParameters params) {
         this.params = params;
@@ -51,27 +56,41 @@ public class PdpModifyRequestMapParams extends RequestDataParams {
         return this;
     }
 
-    @Override
+    public PdpModifyRequestMapParams setDaoFactory(PolicyModelsProviderFactoryWrapper daoFactory) {
+        this.daoFactory = daoFactory;
+        return this;
+    }
+
     public PdpModifyRequestMapParams setPublisher(Publisher publisher) {
-        super.setPublisher(publisher);
+        this.publisher = publisher;
         return this;
     }
 
-    @Override
     public PdpModifyRequestMapParams setResponseDispatcher(RequestIdDispatcher<PdpStatus> responseDispatcher) {
-        super.setResponseDispatcher(responseDispatcher);
+        this.responseDispatcher = responseDispatcher;
         return this;
     }
 
-    @Override
     public PdpModifyRequestMapParams setModifyLock(Object modifyLock) {
-        super.setModifyLock(modifyLock);
+        this.modifyLock = modifyLock;
         return this;
     }
 
-    @Override
+    /**
+     * Validates the parameters.
+     */
     public void validate() {
-        super.validate();
+        if (publisher == null) {
+            throw new IllegalArgumentException("missing publisher");
+        }
+
+        if (responseDispatcher == null) {
+            throw new IllegalArgumentException("missing responseDispatcher");
+        }
+
+        if (modifyLock == null) {
+            throw new IllegalArgumentException("missing modifyLock");
+        }
 
         if (params == null) {
             throw new IllegalArgumentException("missing PDP parameters");
diff --git a/main/src/main/java/org/onap/policy/pap/main/parameters/RequestDataParams.java b/main/src/main/java/org/onap/policy/pap/main/parameters/RequestParams.java
index ea4b02ca..b9083864 100644
--- a/main/src/main/java/org/onap/policy/pap/main/parameters/RequestDataParams.java
+++ b/main/src/main/java/org/onap/policy/pap/main/parameters/RequestParams.java
@@ -24,33 +24,46 @@ import lombok.Getter;
 import org.onap.policy.common.endpoints.listeners.RequestIdDispatcher;
 import org.onap.policy.models.pdp.concepts.PdpStatus;
 import org.onap.policy.pap.main.comm.Publisher;
-import org.onap.policy.pap.main.comm.RequestData;
+import org.onap.policy.pap.main.comm.TimerManager;
 
 
 /**
- * Parameters needed to create a {@link RequestData}.
+ * Parameters needed to create a Request.
  */
 @Getter
-public class RequestDataParams {
+public class RequestParams {
     private Publisher publisher;
     private RequestIdDispatcher<PdpStatus> responseDispatcher;
     private Object modifyLock;
+    private TimerManager timers;
+    private int maxRetryCount;
 
-    public RequestDataParams setPublisher(Publisher publisher) {
+
+    public RequestParams setPublisher(Publisher publisher) {
         this.publisher = publisher;
         return this;
     }
 
-    public RequestDataParams setResponseDispatcher(RequestIdDispatcher<PdpStatus> responseDispatcher) {
+    public RequestParams setResponseDispatcher(RequestIdDispatcher<PdpStatus> responseDispatcher) {
         this.responseDispatcher = responseDispatcher;
         return this;
     }
 
-    public RequestDataParams setModifyLock(Object modifyLock) {
+    public RequestParams setModifyLock(Object modifyLock) {
         this.modifyLock = modifyLock;
         return this;
     }
 
+    public RequestParams setTimers(TimerManager timers) {
+        this.timers = timers;
+        return this;
+    }
+
+    public RequestParams setMaxRetryCount(int maxRetryCount) {
+        this.maxRetryCount = maxRetryCount;
+        return this;
+    }
+
     /**
      * Validates the parameters.
      */
@@ -66,5 +79,9 @@ public class RequestDataParams {
         if (modifyLock == null) {
             throw new IllegalArgumentException("missing modifyLock");
         }
+
+        if (timers == null) {
+            throw new IllegalArgumentException("missing timers");
+        }
     }
 }
diff --git a/main/src/test/java/org/onap/policy/pap/main/comm/CommonRequestBase.java b/main/src/test/java/org/onap/policy/pap/main/comm/CommonRequestBase.java
new file mode 100644
index 00000000..ceda1bab
--- /dev/null
+++ b/main/src/test/java/org/onap/policy/pap/main/comm/CommonRequestBase.java
@@ -0,0 +1,245 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP PAP
+ * ================================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. 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.
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.pap.main.comm;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.Queue;
+import java.util.function.Consumer;
+import org.junit.Before;
+import org.mockito.ArgumentCaptor;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
+import org.onap.policy.common.endpoints.listeners.RequestIdDispatcher;
+import org.onap.policy.common.endpoints.listeners.TypedMessageListener;
+import org.onap.policy.models.pdp.concepts.PdpMessage;
+import org.onap.policy.models.pdp.concepts.PdpStateChange;
+import org.onap.policy.models.pdp.concepts.PdpStatus;
+import org.onap.policy.models.pdp.concepts.PdpUpdate;
+import org.onap.policy.models.pdp.enums.PdpState;
+import org.onap.policy.models.provider.PolicyModelsProvider;
+import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicy;
+import org.onap.policy.pap.main.PapConstants;
+import org.onap.policy.pap.main.PolicyModelsProviderFactoryWrapper;
+import org.onap.policy.pap.main.comm.msgdata.RequestListener;
+import org.onap.policy.pap.main.comm.msgdata.StateChangeReq;
+import org.onap.policy.pap.main.comm.msgdata.UpdateReq;
+import org.onap.policy.pap.main.parameters.PdpModifyRequestMapParams;
+import org.onap.policy.pap.main.parameters.PdpParameters;
+import org.onap.policy.pap.main.parameters.PdpStateChangeParameters;
+import org.onap.policy.pap.main.parameters.PdpUpdateParameters;
+import org.onap.policy.pap.main.parameters.RequestParams;
+
+/**
+ * Common base class for request tests.
+ */
+public class CommonRequestBase {
+    protected static final String PDP1 = "pdp_1";
+    protected static final String MY_REQ_NAME = "my-request";
+    protected static final String DIFFERENT = "different-value";
+    protected static final String MY_GROUP = "my-group";
+    protected static final String MY_SUBGROUP = "my-subgroup";
+    protected static final String MY_NAME = "my-name";
+    protected static final PdpState MY_STATE = PdpState.SAFE;
+    protected static final PdpState DIFF_STATE = PdpState.TERMINATED;
+    protected static final int RETRIES = 1;
+
+    protected Publisher publisher;
+    protected RequestIdDispatcher<PdpStatus> dispatcher;
+    protected Object lock;
+    protected TimerManager timers;
+    protected TimerManager.Timer timer;
+    protected Queue<QueueToken<PdpMessage>> queue;
+    protected RequestListener listener;
+    protected PolicyModelsProviderFactoryWrapper daoFactory;
+    protected PolicyModelsProvider dao;
+    protected RequestParams reqParams;
+    protected PdpModifyRequestMapParams mapParams;
+
+    /**
+     * Sets up.
+     *
+     * @throws Exception if an error occurs
+     */
+    @Before
+    @SuppressWarnings("unchecked")
+    public void setUp() throws Exception {
+        publisher = mock(Publisher.class);
+        dispatcher = mock(RequestIdDispatcher.class);
+        lock = new Object();
+        timers = mock(TimerManager.class);
+        timer = mock(TimerManager.Timer.class);
+        queue = new LinkedList<>();
+        listener = mock(RequestListener.class);
+        daoFactory = mock(PolicyModelsProviderFactoryWrapper.class);
+        dao = mock(PolicyModelsProvider.class);
+
+        PdpParameters pdpParams = mock(PdpParameters.class);
+
+        doAnswer(new Answer<Object>() {
+            @Override
+            public Object answer(InvocationOnMock invocation) throws Throwable {
+                queue.add(invocation.getArgumentAt(0, QueueToken.class));
+                return null;
+            }
+        }).when(publisher).enqueue(any());
+
+        when(timers.register(any(), any())).thenReturn(timer);
+
+        when(daoFactory.create()).thenReturn(dao);
+
+        PdpStateChangeParameters stateParams = mock(PdpStateChangeParameters.class);
+        when(stateParams.getMaxRetryCount()).thenReturn(RETRIES);
+        when(pdpParams.getStateChangeParameters()).thenReturn(stateParams);
+
+        PdpUpdateParameters updateParams = mock(PdpUpdateParameters.class);
+        when(updateParams.getMaxRetryCount()).thenReturn(RETRIES);
+        when(pdpParams.getUpdateParameters()).thenReturn(updateParams);
+
+        reqParams = new RequestParams().setMaxRetryCount(RETRIES).setModifyLock(lock).setPublisher(publisher)
+                        .setResponseDispatcher(dispatcher).setTimers(timers);
+
+        mapParams = new PdpModifyRequestMapParams().setModifyLock(lock).setPublisher(publisher)
+                        .setResponseDispatcher(dispatcher).setDaoFactory(daoFactory).setUpdateTimers(timers)
+                        .setStateChangeTimers(timers).setParams(pdpParams);
+    }
+
+    /**
+     * Gets the listener that was registered with the dispatcher and invokes it.
+     *
+     * @param response the response to pass to the listener
+     */
+    @SuppressWarnings("unchecked")
+    protected void invokeProcessResponse(PdpStatus response) {
+        @SuppressWarnings("rawtypes")
+        ArgumentCaptor<TypedMessageListener> processResp = ArgumentCaptor.forClass(TypedMessageListener.class);
+
+        verify(dispatcher).register(any(), processResp.capture());
+
+        processResp.getValue().onTopicEvent(CommInfrastructure.NOOP, PapConstants.TOPIC_POLICY_PDP_PAP, response);
+    }
+
+    /**
+     * Gets the timeout handler that was registered with the timer manager and invokes it.
+     */
+    @SuppressWarnings("unchecked")
+    protected void invokeTimeoutHandler() {
+        @SuppressWarnings("rawtypes")
+        ArgumentCaptor<Consumer> timeoutHdlr = ArgumentCaptor.forClass(Consumer.class);
+
+        verify(timers).register(any(), timeoutHdlr.capture());
+
+        timeoutHdlr.getValue().accept(PDP1);
+    }
+
+    /**
+     * Creates a policy with the given name and version.
+     *
+     * @param name policy name
+     * @param version policy version
+     * @return a new policy
+     */
+    protected ToscaPolicy makePolicy(String name, String version) {
+        ToscaPolicy policy = new ToscaPolicy();
+
+        policy.setName(name);
+        policy.setVersion(version);
+
+        return policy;
+    }
+
+    /**
+     * Makes an update request with a new message.
+     *
+     * @param pdpName PDP name
+     * @param group group name
+     * @param subgroup subgroup name
+     * @return a new update request
+     */
+    protected UpdateReq makeUpdateReq(String pdpName, String group, String subgroup) {
+        UpdateReq req = mock(UpdateReq.class);
+
+        when(req.getName()).thenReturn(MY_REQ_NAME);
+        when(req.getPriority()).thenReturn(1);
+        when(req.getMessage()).thenReturn(makeUpdate(pdpName, group, subgroup));
+
+        return req;
+    }
+
+    /**
+     * Makes an update message.
+     *
+     * @param pdpName PDP name
+     * @param group group name
+     * @param subgroup subgroup name
+     * @return a new update message
+     */
+    protected PdpUpdate makeUpdate(String pdpName, String group, String subgroup) {
+        PdpUpdate message = new PdpUpdate();
+
+        message.setName(pdpName);
+        message.setPolicies(Collections.emptyList());
+        message.setPdpGroup(group);
+        message.setPdpSubgroup(subgroup);
+
+        return message;
+    }
+
+    /**
+     * Makes a state-change request with a new message.
+     *
+     * @param pdpName PDP name
+     * @param state desired PDP state
+     * @return a new state-change request
+     */
+    protected StateChangeReq makeStateChangeReq(String pdpName, PdpState state) {
+        StateChangeReq req = mock(StateChangeReq.class);
+
+        when(req.getName()).thenReturn(MY_REQ_NAME);
+        when(req.getPriority()).thenReturn(0);
+        when(req.getMessage()).thenReturn(makeStateChange(pdpName, state));
+
+        return req;
+    }
+
+    /**
+     * Makes a state-change message.
+     *
+     * @param pdpName PDP name
+     * @param state desired PDP state
+     * @return a new state-change message
+     */
+    protected PdpStateChange makeStateChange(String pdpName, PdpState state) {
+        PdpStateChange message = new PdpStateChange();
+
+        message.setName(pdpName);
+        message.setState(state);
+
+        return message;
+    }
+}
diff --git a/main/src/test/java/org/onap/policy/pap/main/comm/PdpModifyRequestMapTest.java b/main/src/test/java/org/onap/policy/pap/main/comm/PdpModifyRequestMapTest.java
index c36a7d4b..199ebcf1 100644
--- a/main/src/test/java/org/onap/policy/pap/main/comm/PdpModifyRequestMapTest.java
+++ b/main/src/test/java/org/onap/policy/pap/main/comm/PdpModifyRequestMapTest.java
@@ -22,570 +22,617 @@ package org.onap.policy.pap.main.comm;
 
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.LinkedList;
+import java.util.Collections;
 import java.util.List;
-import java.util.Map;
-import java.util.Queue;
-import java.util.function.Consumer;
 import java.util.stream.Collectors;
+import javax.ws.rs.core.Response.Status;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
-import org.onap.policy.common.endpoints.listeners.RequestIdDispatcher;
-import org.onap.policy.common.endpoints.listeners.TypedMessageListener;
+import org.mockito.Captor;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+import org.onap.policy.models.base.PfModelException;
+import org.onap.policy.models.pdp.concepts.Pdp;
+import org.onap.policy.models.pdp.concepts.PdpGroup;
 import org.onap.policy.models.pdp.concepts.PdpMessage;
 import org.onap.policy.models.pdp.concepts.PdpStateChange;
 import org.onap.policy.models.pdp.concepts.PdpStatus;
+import org.onap.policy.models.pdp.concepts.PdpSubGroup;
 import org.onap.policy.models.pdp.concepts.PdpUpdate;
 import org.onap.policy.models.pdp.enums.PdpState;
-import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicy;
-import org.onap.policy.pap.main.PapConstants;
-import org.onap.policy.pap.main.comm.PdpModifyRequestMap.ModifyReqData;
+import org.onap.policy.pap.main.comm.msgdata.Request;
+import org.onap.policy.pap.main.comm.msgdata.RequestListener;
 import org.onap.policy.pap.main.parameters.PdpModifyRequestMapParams;
-import org.onap.policy.pap.main.parameters.PdpParameters;
-import org.onap.policy.pap.main.parameters.PdpStateChangeParameters;
-import org.onap.policy.pap.main.parameters.PdpUpdateParameters;
 import org.powermock.reflect.Whitebox;
 
-public class PdpModifyRequestMapTest {
-    private static final String DIFFERENT = "-diff";
-    private static final String PDP1 = "pdp_1";
-
-    private static final int UPDATE_RETRIES = 2;
-    private static final int STATE_RETRIES = 1;
-
-    private PdpModifyRequestMap map;
-    private Publisher pub;
-    private RequestIdDispatcher<PdpStatus> disp;
-    private Object lock;
-    private TimerManager updTimers;
-    private TimerManager stateTimers;
-    private TimerManager.Timer timer;
-    private Queue<QueueToken<PdpMessage>> queue;
-    private PdpStatus response;
-    private PdpParameters pdpParams;
-    private PdpUpdateParameters updParams;
-    private PdpStateChangeParameters stateParams;
+public class PdpModifyRequestMapTest extends CommonRequestBase {
+    private static final String MY_REASON = "my reason";
+    private static final String MY_VERSION = "1.2.3";
+
+    /**
+     * Used to capture input to dao.createPdpGroups().
+     */
+    @Captor
+    private ArgumentCaptor<List<PdpGroup>> createCaptor;
+
+
+    /**
+     * Used to capture input to dao.updatePdpGroups().
+     */
+    @Captor
+    private ArgumentCaptor<List<PdpGroup>> updateCaptor;
+
+    @Mock
+    private PdpRequests requests;
+
+    private MyMap map;
     private PdpUpdate update;
-    private PdpStateChange state;
-    private String mismatchReason;
+    private PdpStateChange change;
+    private PdpStatus response;
 
     /**
      * Sets up.
+     *
+     * @throws Exception if an error occurs
      */
     @Before
-    @SuppressWarnings("unchecked")
-    public void setUp() {
-        pub = mock(Publisher.class);
-        disp = mock(RequestIdDispatcher.class);
-        lock = new Object();
-        updTimers = mock(TimerManager.class);
-        stateTimers = mock(TimerManager.class);
-        timer = mock(TimerManager.Timer.class);
-        queue = new LinkedList<>();
+    public void setUp() throws Exception {
+        super.setUp();
+
+        MockitoAnnotations.initMocks(this);
+
         response = new PdpStatus();
-        pdpParams = mock(PdpParameters.class);
-        updParams = mock(PdpUpdateParameters.class);
-        stateParams = mock(PdpStateChangeParameters.class);
-        update = makeUpdate();
-        state = makeStateChange();
-        mismatchReason = null;
-
-        doAnswer(new Answer<Object>() {
-            @Override
-            public Object answer(InvocationOnMock invocation) throws Throwable {
-                queue.add(invocation.getArgumentAt(0, QueueToken.class));
-                return null;
-            }
-        }).when(pub).enqueue(any());
-
-        when(updTimers.register(any(), any())).thenReturn(timer);
-        when(stateTimers.register(any(), any())).thenReturn(timer);
-
-        when(pdpParams.getUpdateParameters()).thenReturn(updParams);
-        when(pdpParams.getStateChangeParameters()).thenReturn(stateParams);
-
-        when(updParams.getMaxRetryCount()).thenReturn(UPDATE_RETRIES);
-        when(updParams.getMaxWaitMs()).thenReturn(1000L);
-
-        when(stateParams.getMaxRetryCount()).thenReturn(STATE_RETRIES);
-        when(stateParams.getMaxWaitMs()).thenReturn(1000L);
-
-        response.setName(PDP1);
-        response.setState(PdpState.SAFE);
-        response.setPdpGroup(update.getPdpGroup());
-        response.setPdpSubgroup(update.getPdpSubgroup());
-        response.setPolicies(update.getPolicies().stream().map(ToscaPolicy::getIdentifier)
-                        .collect(Collectors.toList()));
 
-        map = new PdpModifyRequestMap(makeParameters()) {
+        update = makeUpdate(PDP1, MY_GROUP, MY_SUBGROUP);
+        change = makeStateChange(PDP1, MY_STATE);
 
-            @Override
-            protected ModifyReqData makeRequestData(PdpUpdate update, PdpStateChange stateChange) {
-                return new ModifyReqData(update, stateChange) {
-                    @Override
-                    protected void mismatch(String reason) {
-                        mismatchReason = reason;
-                        super.mismatch(reason);
-                    }
-                };
-            }
-        };
+        when(requests.getPdpName()).thenReturn(PDP1);
 
-        map = spy(map);
+        response.setName(MY_NAME);
+        response.setState(MY_STATE);
+        response.setPdpGroup(update.getPdpGroup());
+        response.setPdpSubgroup(update.getPdpSubgroup());
+        response.setPolicies(Collections.emptyList());
+
+        map = new MyMap(mapParams);
     }
 
     @Test
-    public void testAdd_DifferentPdps() {
-        map.addRequest(update);
+    public void testPdpModifyRequestMap() {
+        assertSame(mapParams, Whitebox.getInternalState(map, "params"));
+        assertSame(lock, Whitebox.getInternalState(map, "modifyLock"));
+        assertSame(daoFactory, Whitebox.getInternalState(map, "daoFactory"));
+    }
 
-        state.setName(DIFFERENT);
-        map.addRequest(state);
+    @Test
+    public void testStopPublishing() {
+        // try with non-existent PDP
+        map.stopPublishing(PDP1);
 
-        assertNotNull(getReqData(PDP1));
-        assertNotNull(getReqData(DIFFERENT));
+        // now start a PDP and try it
+        map.addRequest(change);
+        map.stopPublishing(PDP1);
+        verify(requests).stopPublishing();
 
-        assertQueueContains("testAdd_DifferentPdps", update, state);
+        // try again - it shouldn't stop publishing again
+        map.stopPublishing(PDP1);
+        verify(requests, times(1)).stopPublishing();
     }
 
     @Test
-    public void testAddRequestPdpUpdate() {
-        map.addRequest(update);
-
-        assertQueueContains("testAddRequestPdpUpdate", update);
+    public void testAddRequestPdpUpdatePdpStateChange_BothNull() {
+        // nulls should be ok
+        map.addRequest(null, null);
     }
 
     @Test
-    public void testAddRequestPdpStateChange() {
-        map.addRequest(state);
+    public void testAddRequestPdpUpdatePdpStateChange_NullUpdate() {
+        map.addRequest(null, change);
 
-        assertQueueContains("testAddRequestPdpStateChange", state);
+        Request req = getSingletons(1).get(0);
+        assertSame(change, req.getMessage());
+        assertEquals("pdp_1 PdpStateChange", req.getName());
     }
 
     @Test
-    public void testAddRequestPdpUpdatePdpStateChange_Both() {
-        map.addRequest(update, state);
+    public void testAddRequestPdpUpdatePdpStateChange_NullStateChange() {
+        map.addRequest(update, null);
 
-        assertQueueContains("testAddRequestPdpUpdatePdpStateChange_Both", update);
+        Request req = getSingletons(1).get(0);
+        assertSame(update, req.getMessage());
+        assertEquals("pdp_1 PdpUpdate", req.getName());
     }
 
     @Test
-    public void testAddRequestPdpUpdatePdpStateChange_BothNull() {
-        map.addRequest(null, null);
+    public void testAddRequestPdpUpdatePdpStateChange_BothProvided() {
+        map.addRequest(update, change);
 
-        // nothing should have been added to the queue
-        assertTrue(queue.isEmpty());
-    }
+        // should have only allocated one request structure
+        assertEquals(1, map.nalloc);
 
-    @Test
-    public void testGetPdpName_SameNames() {
-        // should be no exception
-        map.addRequest(update, state);
-    }
+        // both requests should have been added
+        List<Request> values = getSingletons(2);
 
-    @Test
-    public void testGetPdpName_DifferentNames() {
-        // should be no exception
-        state.setName(update.getName() + "X");
-        assertThatIllegalArgumentException().isThrownBy(() -> map.addRequest(update, state))
-                        .withMessageContaining("does not match");
+        Request req = values.remove(0);
+        assertSame(update, req.getMessage());
+        assertEquals("pdp_1 PdpUpdate", req.getName());
+
+        req = values.remove(0);
+        assertSame(change, req.getMessage());
+        assertEquals("pdp_1 PdpStateChange", req.getName());
     }
 
     @Test
-    public void testGetPdpName_NullUpdateName() {
-        update.setName(null);
-        assertThatIllegalArgumentException().isThrownBy(() -> map.addRequest(update)).withMessageContaining("update");
+    public void testAddRequestPdpUpdatePdpStateChange() {
+        // null should be ok
+        map.addRequest(null, null);
 
-        assertThatIllegalArgumentException().isThrownBy(() -> map.addRequest(update, state))
-                        .withMessageContaining("update");
+        map.addRequest(change);
 
-        // both names are null
-        state.setName(null);
-        assertThatIllegalArgumentException().isThrownBy(() -> map.addRequest(update, state));
+        Request req = getSingletons(1).get(0);
+        assertSame(change, req.getMessage());
+        assertEquals("pdp_1 PdpStateChange", req.getName());
+
+        // broadcast should throw an exception
+        change.setName(null);
+        assertThatIllegalArgumentException().isThrownBy(() -> map.addRequest(change))
+                        .withMessageStartingWith("unexpected broadcast message: PdpStateChange");
     }
 
     @Test
-    public void testGetPdpName_NullStateName() {
-        state.setName(null);
-        assertThatIllegalArgumentException().isThrownBy(() -> map.addRequest(state)).withMessageContaining("state");
+    public void testAddRequestPdpUpdate() {
+        // null should be ok
+        map.addRequest((PdpUpdate) null);
 
-        assertThatIllegalArgumentException().isThrownBy(() -> map.addRequest(update, state))
-                        .withMessageContaining("state");
+        map.addRequest(update);
+
+        Request req = getSingletons(1).get(0);
+        assertSame(update, req.getMessage());
+        assertEquals("pdp_1 PdpUpdate", req.getName());
 
-        // both names are null
+        // broadcast should throw an exception
         update.setName(null);
-        assertThatIllegalArgumentException().isThrownBy(() -> map.addRequest(update, state));
+        assertThatIllegalArgumentException().isThrownBy(() -> map.addRequest(update))
+                        .withMessageStartingWith("unexpected broadcast message: PdpUpdate");
     }
 
     @Test
-    public void testIsSamePdpUpdatePdpUpdate() {
-        map.addRequest(update);
+    public void testAddRequestPdpStateChange() {
+        // null should be ok
+        map.addRequest((PdpStateChange) null);
+
+        map.addRequest(change);
 
-        // queue a similar request
-        PdpUpdate update2 = makeUpdate();
-        map.addRequest(update2);
+        Request req = getSingletons(1).get(0);
+        assertSame(change, req.getMessage());
+        assertEquals("pdp_1 PdpStateChange", req.getName());
 
-        // token should still have original message
-        assertQueueContains("testIsSamePdpUpdatePdpUpdate", update);
+        // broadcast should throw an exception
+        change.setName(null);
+        assertThatIllegalArgumentException().isThrownBy(() -> map.addRequest(change))
+                        .withMessageStartingWith("unexpected broadcast message: PdpStateChange");
     }
 
     @Test
-    public void testIsSamePdpUpdatePdpUpdate_DifferentPolicyCount() {
-        map.addRequest(update);
+    public void testAddSingleton() {
+        map.addRequest(change);
+        assertEquals(1, map.nalloc);
+
+        // should have one singleton
+        getSingletons(1);
+
+        // add another request with the same PDP
+        map.addRequest(makeStateChange(PDP1, MY_STATE));
+        assertEquals(1, map.nalloc);
+
+        // should now have another singleton
+        getSingletons(2);
+
 
-        PdpUpdate update2 = makeUpdate();
-        update2.setPolicies(Arrays.asList(update.getPolicies().get(0)));
-        map.addRequest(update2);
+        // add another request with a different PDP
+        map.addRequest(makeStateChange(DIFFERENT, MY_STATE));
 
-        // should have replaced the message in the token
-        assertQueueContains("testIsSamePdpUpdatePdpUpdate_DifferentPolicyCount", update2);
+        // should now have another allocation
+        assertEquals(2, map.nalloc);
+
+        // should now have another singleton
+        getSingletons(3);
     }
 
     @Test
-    public void testIsSamePdpUpdatePdpUpdate_DifferentGroup() {
-        map.addRequest(update);
+    public void testStartNextRequest_NoMore() {
+        map.addRequest(change);
 
-        // queue a similar request
-        PdpUpdate update2 = makeUpdate();
-        update2.setPdpGroup(update.getPdpGroup() + DIFFERENT);
-        map.addRequest(update2);
+        // indicate success
+        getListener(getSingletons(1).get(0)).success(PDP1);
 
-        // should have replaced the message in the token
-        assertQueueContains("testIsSamePdpUpdatePdpUpdate_DifferentGroup", update2);
+        /*
+         * the above should have removed the requests so next time should allocate a new
+         * one
+         */
+        map.addRequest(change);
+        assertEquals(2, map.nalloc);
     }
 
     @Test
-    public void testIsSamePdpUpdatePdpUpdate_DifferentSubGroup() {
+    public void testStartNextRequest_HaveMore() {
         map.addRequest(update);
+        map.addRequest(change);
+
+        Request updateReq = getSingletons(2).get(0);
 
-        PdpUpdate update2 = makeUpdate();
-        update2.setPdpSubgroup(update.getPdpSubgroup() + DIFFERENT);
-        map.addRequest(update2);
+        // indicate success with the update
+        when(requests.startNextRequest(updateReq)).thenReturn(true);
+        getListener(updateReq).success(PDP1);
 
-        // should have replaced the message in the token
-        assertQueueContains("testIsSamePdpUpdatePdpUpdate_DifferentSubGroup", update2);
+        // should have started the next request
+        verify(requests).startNextRequest(updateReq);
+
+        /*
+         * requests should still be there, so adding another request should not allocate a
+         * new one
+         */
+        map.addRequest(update);
+        assertEquals(1, map.nalloc);
     }
 
     @Test
-    public void testIsSamePdpUpdatePdpUpdate_DifferentPolicies() {
+    public void testDisablePdp() {
         map.addRequest(update);
 
-        ArrayList<ToscaPolicy> policies = new ArrayList<>(update.getPolicies());
-        policies.set(0, makePolicy("policy-3-x", "2.0.0"));
+        when(requests.getLastGroupName()).thenReturn(MY_GROUP);
+
+        // indicate failure
+        invokeFailureHandler(1);
 
-        PdpUpdate update2 = makeUpdate();
-        update2.setPolicies(policies);
-        map.addRequest(update2);
+        // should have stopped publishing
+        verify(requests).stopPublishing();
 
-        // should have replaced the message in the token
-        assertQueueContains("testIsSamePdpUpdatePdpUpdate_DifferentPolicies", update2);
+        // should have published a new state-change
+        PdpMessage msg2 = getSingletons(2).get(1).getMessage();
+        assertNotNull(msg2);
+        assertTrue(msg2 instanceof PdpStateChange);
+
+        change = (PdpStateChange) msg2;
+        assertEquals(PDP1, change.getName());
+        assertEquals(PdpState.PASSIVE, change.getState());
     }
 
     @Test
-    public void testIsSamePdpStateChangePdpStateChange() {
-        map.addRequest(state);
+    public void testDisablePdp_AlreadyRemoved() {
+        map.addRequest(change);
+        map.stopPublishing(PDP1);
+
+        when(requests.getLastGroupName()).thenReturn(MY_GROUP);
 
-        // queue a similar request
-        PdpStateChange state2 = makeStateChange();
-        map.addRequest(state2);
+        invokeFailureHandler(1);
 
-        // token should still have original message
-        assertQueueContains("testIsSamePdpStateChangePdpStateChange", state);
+        // should not have stopped publishing a second time
+        verify(requests, times(1)).stopPublishing();
     }
 
     @Test
-    public void testIsSamePdpStateChangePdpStateChange_DifferentState() {
-        map.addRequest(state);
+    public void testDisablePdp_NoGroup() {
+        map.addRequest(change);
 
-        // queue a similar request
-        PdpStateChange state2 = makeStateChange();
-        state2.setState(PdpState.TERMINATED);
-        map.addRequest(state2);
+        invokeFailureHandler(1);
 
-        // should have replaced the message in the token
-        assertQueueContains("testIsSamePdpStateChangePdpStateChange_DifferentState", state2);
+        // should not have stopped publishing
+        verify(requests).stopPublishing();
     }
 
     @Test
-    public void testModifyReqDataIsActive() {
-        map.addRequest(update);
+    public void testRemoveFromGroup() throws Exception {
+        map.addRequest(change);
+
+        when(requests.getLastGroupName()).thenReturn(MY_GROUP);
 
-        ModifyReqData reqdata = getReqData(PDP1);
-        assertNotNull(reqdata);
+        PdpGroup group = makeGroup(MY_GROUP, MY_VERSION);
+        group.setPdpSubgroups(Arrays.asList(makeSubGroup(MY_SUBGROUP + "a", PDP1 + "a"),
+                        makeSubGroup(MY_SUBGROUP, PDP1), makeSubGroup(MY_SUBGROUP + "c", PDP1 + "c")));
 
-        // this should remove it from the map
-        invokeProcessResponse();
+        when(dao.getFilteredPdpGroups(any())).thenReturn(Arrays.asList(group));
 
-        assertFalse(reqdata.isActive());
+        invokeFailureHandler(1);
+
+        // verify that the PDP was removed from the subgroup
+        List<PdpGroup> groups = getGroupUpdates();
+        assertEquals(1, groups.size());
+        assertSame(group, groups.get(0));
+
+        List<PdpSubGroup> subgroups = group.getPdpSubgroups();
+        assertEquals(3, subgroups.size());
+        assertEquals("[pdp_1a]", getPdpNames(subgroups.get(0)));
+        assertEquals("[]", getPdpNames(subgroups.get(1)));
+        assertEquals("[pdp_1c]", getPdpNames(subgroups.get(2)));
     }
 
     @Test
-    public void testModifyReqDataAddPdpUpdate() {
-        map.addRequest(state);
+    public void testRemoveFromGroup_DaoEx() throws Exception {
+        map.addRequest(change);
 
-        map.addRequest(update);
+        when(requests.getLastGroupName()).thenReturn(MY_GROUP);
 
-        // update should have replaced the state-change in the queue
-        assertQueueContains("testModifyReqDataAddPdpUpdate", update);
+        when(dao.getFilteredPdpGroups(any())).thenThrow(new PfModelException(Status.BAD_REQUEST, "expected exception"));
+
+        invokeFailureHandler(1);
+
+        // should still stop publishing
+        verify(requests).stopPublishing();
+
+        // requests should have been removed from the map so this should allocate another
+        map.addRequest(update);
+        assertEquals(2, map.nalloc);
     }
 
     @Test
-    public void testModifyReqDataAddPdpStateChange() {
-        map.addRequest(update);
+    public void testRemoveFromGroup_NoGroups() throws Exception {
+        map.addRequest(change);
+
+        when(requests.getLastGroupName()).thenReturn(MY_GROUP);
 
-        map.addRequest(state);
+        invokeFailureHandler(1);
 
-        // update should still be in the queue
-        assertQueueContains("testModifyReqDataAddPdpStateChange", update);
+        verify(dao, never()).updatePdpGroups(any());
     }
 
     @Test
-    public void testModifyReqDataRetryCountExhausted() {
-        map.addRequest(state);
+    public void testRemoveFromGroup_NoMatchingSubgroup() throws Exception {
+        map.addRequest(change);
+
+        when(requests.getLastGroupName()).thenReturn(MY_GROUP);
 
-        // timeout twice so that retry count is exhausted
-        invokeTimeoutHandler(stateTimers, STATE_RETRIES + 1);
+        PdpGroup group = makeGroup(MY_GROUP, MY_VERSION);
+        group.setPdpSubgroups(Arrays.asList(makeSubGroup(MY_SUBGROUP, DIFFERENT)));
 
-        // name should have been removed
-        assertNull(getReqData(PDP1));
+        when(dao.getFilteredPdpGroups(any())).thenReturn(Arrays.asList(group));
+
+        invokeFailureHandler(1);
+
+        verify(dao, never()).updatePdpGroups(any());
     }
 
     @Test
-    public void testModifyReqDataMismatch() {
-        map.addRequest(state);
+    public void testRemoveFromSubgroup() throws Exception {
+        map.addRequest(change);
+
+        when(requests.getLastGroupName()).thenReturn(MY_GROUP);
+
+        PdpGroup group = makeGroup(MY_GROUP, MY_VERSION);
+        group.setPdpSubgroups(Arrays.asList(makeSubGroup(MY_SUBGROUP, PDP1, PDP1 + "x", PDP1 + "y")));
 
-        // set up a response with incorrect info
-        response.setName(state.getName() + DIFFERENT);
+        when(dao.getFilteredPdpGroups(any())).thenReturn(Arrays.asList(group));
 
-        invokeProcessResponse();
+        invokeFailureHandler(1);
 
-        assertNotNull(mismatchReason);
+        // verify that the PDP was removed from the subgroup
+        List<PdpGroup> groups = getGroupUpdates();
+        assertEquals(1, groups.size());
+        assertSame(group, groups.get(0));
 
-        // name should have been removed
-        assertNull(getReqData(PDP1));
+        PdpSubGroup subgroup = group.getPdpSubgroups().get(0);
+        assertEquals(2, subgroup.getCurrentInstanceCount());
+        assertEquals("[pdp_1x, pdp_1y]", getPdpNames(subgroup));
     }
 
     @Test
-    public void testUpdateDataGetMaxRetryCount() {
-        map.addRequest(update);
-        ModifyReqData reqdata = getReqData(PDP1);
+    public void testMakePdpRequests() {
+        // this should invoke the real method without throwing an exception
+        new PdpModifyRequestMap(mapParams).addRequest(change);
 
-        for (int count = 0; count < UPDATE_RETRIES; ++count) {
-            assertTrue("update bump " + count, reqdata.bumpRetryCount());
-        }
+        QueueToken<PdpMessage> token = queue.poll();
+        assertNotNull(token);
+        assertSame(change, token.get());
 
-        assertFalse("update bump final", reqdata.bumpRetryCount());
+        verify(dispatcher).register(eq(change.getRequestId()), any());
+        verify(timers).register(eq(change.getRequestId()), any());
     }
 
     @Test
-    public void testUpdateDataMismatch() {
-        map.addRequest(update);
+    public void testSingletonListenerFailure() throws Exception {
+        map.addRequest(change);
 
-        response.setName(DIFFERENT);
-        invokeProcessResponse();
+        // invoke the method
+        invokeFailureHandler(1);
 
-        assertNull(getReqData(PDP1));
+        verify(requests).stopPublishing();
     }
 
     @Test
-    public void testUpdateDataComplete() {
-        map.addRequest(update);
+    public void testSingletonListenerFailure_WrongPdpName() throws Exception {
+        map.addRequest(change);
 
-        invokeProcessResponse();
+        // invoke the method - has wrong PDP name
+        when(requests.getPdpName()).thenReturn(DIFFERENT);
+        invokeFailureHandler(1);
 
-        assertNull(getReqData(PDP1));
+        verify(requests, never()).stopPublishing();
     }
 
     @Test
-    public void testUpdateDataComplete_MoreToGo() {
-        map.addRequest(update, state);
+    public void testSingletonListenerSuccess_LastRequest() throws Exception {
+        map.addRequest(change);
 
-        invokeProcessResponse();
+        // invoke the method
+        invokeSuccessHandler(1);
 
-        assertNotNull(getReqData(PDP1));
+        verify(requests, never()).stopPublishing();
 
-        assertSame(state, queue.poll().get());
+        // requests should have been removed from the map so this should allocate another
+        map.addRequest(update);
+        assertEquals(2, map.nalloc);
     }
 
     @Test
-    public void testStateChangeDataMismatch() {
-        map.addRequest(state);
+    public void testSingletonListenerSuccess_NameMismatch() throws Exception {
+        map.addRequest(change);
 
-        response.setName(DIFFERENT);
-        invokeProcessResponse();
+        // invoke the method - with a different name
+        when(requests.getPdpName()).thenReturn(DIFFERENT);
+        invokeSuccessHandler(1);
 
-        assertNull(getReqData(PDP1));
+        verify(requests, never()).stopPublishing();
+
+        // no effect on the map
+        map.addRequest(update);
+        assertEquals(1, map.nalloc);
     }
 
     @Test
-    public void testStateChangeDataCompleted() {
-        map.addRequest(state);
+    public void testSingletonListenerSuccess_AlreadyStopped() throws Exception {
+        map.addRequest(change);
 
-        invokeProcessResponse();
+        map.stopPublishing(PDP1);
 
-        assertNull(getReqData(PDP1));
+        // invoke the method
+        invokeSuccessHandler(1);
+
+        // should have called this a second time
+        verify(requests, times(2)).stopPublishing();
+
+        // requests should have been removed from the map so this should allocate another
+        map.addRequest(update);
+        assertEquals(2, map.nalloc);
     }
 
     @Test
-    public void testMakeRequestData() {
-        // need a map that doesn't override the method
-        map = new PdpModifyRequestMap(makeParameters());
+    public void testSingletonListenerRetryCountExhausted() throws Exception {
+        map.addRequest(change);
 
-        // this will invoke makeRequestData() - should not throw an exception
-        map.addRequest(update);
+        // invoke the method
+        getListener(getSingletons(1).get(0)).retryCountExhausted();
 
-        assertNotNull(getReqData(PDP1));
+        verify(requests).stopPublishing();
     }
 
+
     /**
-     * Asserts that the queue contains the specified messages.
+     * Invokes the first request's listener.success() method.
      *
-     * @param testName the test name
-     * @param messages messages that are expected in the queue
+     * @param count expected number of requests
      */
-    private void assertQueueContains(String testName, PdpMessage... messages) {
-        assertEquals(testName, messages.length, queue.size());
-
-        int count = 0;
-        for (PdpMessage msg : messages) {
-            ++count;
-
-            QueueToken<PdpMessage> token = queue.remove();
-            assertSame(testName + "-" + count, msg, token.get());
-        }
+    private void invokeSuccessHandler(int count) {
+        getListener(getSingletons(count).get(0)).success(PDP1);
     }
 
     /**
-     * Makes parameters to configure a map.
+     * Invokes the first request's listener.failure() method.
      *
-     * @return new parameters
+     * @param count expected number of requests
      */
-    private PdpModifyRequestMapParams makeParameters() {
-        return new PdpModifyRequestMapParams().setModifyLock(lock).setParams(pdpParams).setPublisher(pub)
-                        .setResponseDispatcher(disp).setStateChangeTimers(stateTimers).setUpdateTimers(updTimers);
+    private void invokeFailureHandler(int count) {
+        getListener(getSingletons(count).get(0)).failure(PDP1, MY_REASON);
     }
 
     /**
-     * Gets the listener that was registered with the dispatcher and invokes it.
+     * Gets the name of the PDPs contained within a subgroup.
      *
-     * @return the response processor
+     * @param subgroup subgroup of interest
+     * @return the name of the PDPs contained within the subgroup
      */
-    @SuppressWarnings("unchecked")
-    private TypedMessageListener<PdpStatus> invokeProcessResponse() {
-        @SuppressWarnings("rawtypes")
-        ArgumentCaptor<TypedMessageListener> processResp = ArgumentCaptor.forClass(TypedMessageListener.class);
-
-        // indicate that is has been published
-        queue.remove().replaceItem(null);
-
-        verify(disp).register(any(), processResp.capture());
-
-        TypedMessageListener<PdpStatus> func = processResp.getValue();
-        func.onTopicEvent(CommInfrastructure.NOOP, PapConstants.TOPIC_POLICY_PDP_PAP, response);
-
-        return func;
+    private String getPdpNames(PdpSubGroup subgroup) {
+        return subgroup.getPdpInstances().stream().map(Pdp::getInstanceId).collect(Collectors.toList()).toString();
     }
 
     /**
-     * Gets the timeout handler that was registered with the timer manager and invokes it.
+     * Gets the singleton requests added to {@link #requests}.
      *
-     * @param timers the timer manager whose handler is to be invoked
-     * @param ntimes number of times to invoke the timeout handler
-     * @return the timeout handler
+     * @param count number of singletons expected
+     * @return the singleton requests
      */
-    @SuppressWarnings("unchecked")
-    private void invokeTimeoutHandler(TimerManager timers, int ntimes) {
-        @SuppressWarnings("rawtypes")
-        ArgumentCaptor<Consumer> timeoutHdlr = ArgumentCaptor.forClass(Consumer.class);
-
-        for (int count = 1; count <= ntimes; ++count) {
-            // indicate that is has been published
-            queue.remove().replaceItem(null);
-
-            verify(timers, times(count)).register(any(), timeoutHdlr.capture());
+    private List<Request> getSingletons(int count) {
+        ArgumentCaptor<Request> captor = ArgumentCaptor.forClass(Request.class);
 
-            @SuppressWarnings("rawtypes")
-            List<Consumer> lst = timeoutHdlr.getAllValues();
-
-            Consumer<String> hdlr = lst.get(lst.size() - 1);
-            hdlr.accept(PDP1);
-        }
+        verify(requests, times(count)).addSingleton(captor.capture());
+        return captor.getAllValues();
     }
 
     /**
-     * Gets the request data from the map.
+     * Gets the listener from a request.
      *
-     * @param pdpName name of the PDP whose data is desired
-     * @return the request data, or {@code null} if the PDP is not in the map
+     * @param request request of interest
+     * @return the request's listener
      */
-    private ModifyReqData getReqData(String pdpName) {
-        Map<String, ModifyReqData> name2data = Whitebox.getInternalState(map, "name2data");
-        return name2data.get(pdpName);
+    private RequestListener getListener(Request request) {
+        return Whitebox.getInternalState(request, "listener");
     }
 
-    /**
-     * Makes an update message.
-     *
-     * @return a new update message
-     */
-    private PdpUpdate makeUpdate() {
-        PdpUpdate upd = new PdpUpdate();
+    private PdpGroup makeGroup(String name, String version) {
+        PdpGroup group = new PdpGroup();
 
-        upd.setDescription("update-description");
-        upd.setName(PDP1);
-        upd.setPdpGroup("group1-a");
-        upd.setPdpSubgroup("sub1-a");
+        group.setName(name);
+        group.setVersion(version);
+
+        return group;
+    }
 
-        upd.setPolicies(Arrays.asList(makePolicy("policy-1-a", "1.0.0"), makePolicy("policy-2-a", "1.1.0")));
+    private PdpSubGroup makeSubGroup(String pdpType, String... pdpNames) {
+        PdpSubGroup subgroup = new PdpSubGroup();
 
-        return upd;
+        subgroup.setPdpType(pdpType);
+        subgroup.setPdpInstances(Arrays.asList(pdpNames).stream().map(this::makePdp).collect(Collectors.toList()));
+
+        return subgroup;
+    }
+
+    private Pdp makePdp(String pdpName) {
+        Pdp pdp = new Pdp();
+        pdp.setInstanceId(pdpName);
+
+        return pdp;
     }
 
     /**
-     * Creates a new policy.
+     * Gets the input to the method.
      *
-     * @param name policy name
-     * @param version policy version
-     * @return a new policy
+     * @return the input that was passed to the dao.updatePdpGroups() method
+     * @throws Exception if an error occurred
      */
-    private ToscaPolicy makePolicy(String name, String version) {
-        ToscaPolicy policy = new ToscaPolicy();
-
-        policy.setName(name);
-        policy.setVersion(version);
+    private List<PdpGroup> getGroupUpdates() throws Exception {
+        verify(dao).updatePdpGroups(updateCaptor.capture());
 
-        return policy;
+        return copyList(updateCaptor.getValue());
     }
 
     /**
-     * Makes a state-change message.
+     * Copies a list and sorts it by group name.
      *
-     * @return a new state-change message
+     * @param source source list to copy
+     * @return a copy of the source list
      */
-    private PdpStateChange makeStateChange() {
-        PdpStateChange cng = new PdpStateChange();
+    private List<PdpGroup> copyList(List<PdpGroup> source) {
+        List<PdpGroup> newlst = new ArrayList<>(source);
+        Collections.sort(newlst, (left, right) -> left.getName().compareTo(right.getName()));
+        return newlst;
+    }
 
-        cng.setName(PDP1);
-        cng.setState(PdpState.SAFE);
+    private class MyMap extends PdpModifyRequestMap {
+        /**
+         * Number of times requests were allocated.
+         */
+        private int nalloc = 0;
 
-        return cng;
+        public MyMap(PdpModifyRequestMapParams params) {
+            super(params);
+        }
+
+        @Override
+        protected PdpRequests makePdpRequests(String pdpName) {
+            ++nalloc;
+            return requests;
+        }
     }
 }
diff --git a/main/src/test/java/org/onap/policy/pap/main/comm/PdpRequestsTest.java b/main/src/test/java/org/onap/policy/pap/main/comm/PdpRequestsTest.java
new file mode 100644
index 00000000..e219c1d5
--- /dev/null
+++ b/main/src/test/java/org/onap/policy/pap/main/comm/PdpRequestsTest.java
@@ -0,0 +1,312 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP PAP
+ * ================================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. 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.
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.pap.main.comm;
+
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.models.pdp.concepts.PdpMessage;
+import org.onap.policy.pap.main.comm.msgdata.StateChangeReq;
+import org.onap.policy.pap.main.comm.msgdata.UpdateReq;
+
+public class PdpRequestsTest extends CommonRequestBase {
+
+    private PdpRequests data;
+    private UpdateReq update;
+    private StateChangeReq change;
+
+    /**
+     * Sets up.
+     */
+    @Before
+    public void setUp() {
+        update = makeUpdateReq(PDP1, MY_GROUP, MY_SUBGROUP);
+        change = makeStateChangeReq(PDP1, MY_STATE);
+
+        data = new PdpRequests(PDP1);
+    }
+
+    @Test
+    public void testPdpRequests_testGetLastGroupName() {
+        assertEquals(PDP1, data.getPdpName());
+    }
+
+    @Test
+    public void testRecordGroup_testGetLatestGroupXxx() {
+        assertNull(data.getLastGroupName());
+
+        data.addSingleton(update);
+        assertEquals(MY_GROUP, data.getLastGroupName());
+
+        UpdateReq req = makeUpdateReq(PDP1, MY_GROUP, MY_SUBGROUP);
+        req.getMessage().setPdpGroup(DIFFERENT);
+        data.addSingleton(req);
+        assertEquals(DIFFERENT, data.getLastGroupName());
+
+        // doesn't record info from other message types
+        StateChangeReq req2 = change;
+        req2.getMessage().setPdpGroup(MY_GROUP);
+        data.addSingleton(req2);
+
+        // should be unchanged
+        assertEquals(DIFFERENT, data.getLastGroupName());
+    }
+
+    @Test
+    public void testAddSingleton() {
+        data.addSingleton(update);
+
+        verify(update).startPublishing(any());
+    }
+
+    @Test
+    public void testAddSingleton_SameAsExisting() {
+        data.addSingleton(update);
+
+        // add duplicate update
+        UpdateReq req2 = makeUpdateReq(PDP1, MY_GROUP, MY_SUBGROUP);
+        data.addSingleton(req2);
+
+        // should not publish duplicate
+        verify(req2, never()).startPublishing(any());
+    }
+
+    @Test
+    public void testAddSingleton_LowerPriority() {
+        data.addSingleton(update);
+
+        // add lower priority request
+        data.addSingleton(change);
+
+        // should not publish lower priority request
+        verify(change, never()).startPublishing(any());
+    }
+
+    @Test
+    public void testAddSingleton_HigherPriority() {
+        data.addSingleton(change);
+
+        QueueToken<PdpMessage> token = new QueueToken<>(change.getMessage());
+        when(change.stopPublishing(false)).thenReturn(token);
+
+        // add higher priority request
+        data.addSingleton(update);
+
+        // should stop publishing lower priority request
+        verify(change).stopPublishing(false);
+
+        // should start publishing higher priority request
+        verify(update).startPublishing(token);
+    }
+
+    @Test
+    public void testAddSingleton_Broadcast() {
+        UpdateReq req = makeUpdateReq(null, MY_GROUP, MY_SUBGROUP);
+        assertThatIllegalArgumentException().isThrownBy(() -> data.addSingleton(req))
+                        .withMessage("unexpected broadcast for pdp_1");
+    }
+
+    @Test
+    public void testCheckExistingSingleton_DoesNotExist() {
+        data.addSingleton(update);
+        verify(update).startPublishing(any());
+    }
+
+    @Test
+    public void testCheckExistingSingleton_SameContent() {
+        data.addSingleton(update);
+
+        // add duplicate update
+        UpdateReq req2 = makeUpdateReq(PDP1, MY_GROUP, MY_SUBGROUP);
+        when(update.isSameContent(req2)).thenReturn(true);
+        data.addSingleton(req2);
+
+        // should not publish duplicate
+        verify(req2, never()).startPublishing(any());
+    }
+
+    @Test
+    public void testCheckExistingSingleton_DifferentContent() {
+        data.addSingleton(update);
+
+        // add different update
+        UpdateReq req2 = makeUpdateReq(PDP1, MY_GROUP, MY_SUBGROUP);
+        when(req2.isSameContent(update)).thenReturn(false);
+        data.addSingleton(req2);
+
+        // should not publish duplicate
+        verify(req2, never()).startPublishing(any());
+
+        // should have re-configured the original
+        verify(update).reconfigure(req2.getMessage(), null);
+
+        // should not have started publishing again
+        verify(update).startPublishing(any());
+    }
+
+    @Test
+    public void testStopPublishing() {
+        data.addSingleton(update);
+        data.addSingleton(change);
+
+        data.stopPublishing();
+
+        verify(update).stopPublishing();
+        verify(change).stopPublishing();
+
+        // repeat, but with only one request in the queue
+        data.addSingleton(update);
+        data.stopPublishing();
+        verify(update, times(2)).stopPublishing();
+
+        // should not have been invoked again
+        verify(change).stopPublishing();
+    }
+
+    @Test
+    public void testStopPublishingLowerPriority() {
+        data.addSingleton(change);
+
+        QueueToken<PdpMessage> token = new QueueToken<>(change.getMessage());
+        when(change.stopPublishing(false)).thenReturn(token);
+
+        // add higher priority request
+        data.addSingleton(update);
+
+        // should stop publishing lower priority request
+        verify(change).stopPublishing(false);
+
+        // should start publishing higher priority request, with the old token
+        verify(update).startPublishing(token);
+    }
+
+    @Test
+    public void testStopPublishingLowerPriority_NothingPublishing() {
+        data.addSingleton(change);
+
+        // change will return a null token when stopPublishing(false) is called
+
+        data.addSingleton(update);
+
+        // should stop publishing lower priority request
+        verify(change).stopPublishing(false);
+
+        // should start publishing higher priority request
+        verify(update).startPublishing(null);
+    }
+
+    @Test
+    public void testStartNextRequest_NothingToStart() {
+        assertFalse(data.startNextRequest(update));
+    }
+
+    @Test
+    public void testStartNextRequest_ZapCurrent() {
+        data.addSingleton(update);
+        assertFalse(data.startNextRequest(update));
+
+        // invoke again
+        assertFalse(data.startNextRequest(update));
+    }
+
+    @Test
+    public void testStartNextRequest_ZapOther() {
+        data.addSingleton(update);
+        data.addSingleton(change);
+
+        // update is still active - should return true
+        assertTrue(data.startNextRequest(change));
+
+        // invoke again
+        assertTrue(data.startNextRequest(change));
+
+        // nothing more once update completes
+        assertFalse(data.startNextRequest(update));
+
+        assertFalse(data.startNextRequest(change));
+    }
+
+    @Test
+    public void testStartNextRequest_StartOther() {
+        data.addSingleton(update);
+        data.addSingleton(change);
+
+        assertTrue(data.startNextRequest(change));
+
+        // should have published update twice, with and without a token
+        verify(update).startPublishing(any());
+        verify(update).startPublishing();
+    }
+
+    @Test
+    public void testStartNextRequest_NoOther() {
+        data.addSingleton(update);
+
+        // nothing else to start
+        assertFalse(data.startNextRequest(update));
+
+        verify(update).startPublishing(any());
+        verify(update, never()).startPublishing();
+    }
+
+    @Test
+    public void testHigherPrioritySingleton_True() {
+        data.addSingleton(update);
+        data.addSingleton(change);
+
+        verify(update).startPublishing(any());
+
+        verify(update, never()).startPublishing();
+        verify(change, never()).startPublishing();
+        verify(change, never()).startPublishing(any());
+    }
+
+    @Test
+    public void testHigherPrioritySingleton_FalseWithUpdate() {
+        data.addSingleton(update);
+
+        verify(update).startPublishing(any());
+        verify(update, never()).startPublishing();
+    }
+
+    @Test
+    public void testHigherPrioritySingleton_FalseWithStateChange() {
+        data.addSingleton(change);
+
+        verify(change).startPublishing(any());
+        verify(change, never()).startPublishing();
+    }
+
+    @Test
+    public void testGetPdpName() {
+        assertEquals(PDP1, data.getPdpName());
+    }
+}
diff --git a/main/src/test/java/org/onap/policy/pap/main/comm/RequestDataTest.java b/main/src/test/java/org/onap/policy/pap/main/comm/RequestDataTest.java
deleted file mode 100644
index 28e5cf96..00000000
--- a/main/src/test/java/org/onap/policy/pap/main/comm/RequestDataTest.java
+++ /dev/null
@@ -1,476 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * ONAP PAP
- * ================================================================================
- * Copyright (C) 2019 AT&T Intellectual Property. 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.
- * 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.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.policy.pap.main.comm;
-
-import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.eq;
-import static org.mockito.Mockito.doAnswer;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-import java.util.LinkedList;
-import java.util.Queue;
-import java.util.function.Consumer;
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.ArgumentCaptor;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
-import org.onap.policy.common.endpoints.listeners.RequestIdDispatcher;
-import org.onap.policy.common.endpoints.listeners.TypedMessageListener;
-import org.onap.policy.common.utils.services.ServiceManager;
-import org.onap.policy.models.pdp.concepts.PdpMessage;
-import org.onap.policy.models.pdp.concepts.PdpStateChange;
-import org.onap.policy.models.pdp.concepts.PdpStatus;
-import org.onap.policy.models.pdp.enums.PdpState;
-import org.onap.policy.pap.main.PapConstants;
-import org.onap.policy.pap.main.comm.msgdata.MessageData;
-import org.onap.policy.pap.main.parameters.RequestDataParams;
-import org.powermock.reflect.Whitebox;
-
-public class RequestDataTest {
-    private static final String PDP1 = "pdp_1";
-    private static final String MY_MSG_TYPE = "my-type";
-
-    private MyRequestData reqdata;
-    private Publisher pub;
-    private RequestIdDispatcher<PdpStatus> disp;
-    private Object lock;
-    private TimerManager timers;
-    private TimerManager.Timer timer;
-    private MyMessageData msgdata;
-    private Queue<QueueToken<PdpMessage>> queue;
-    private PdpStatus response;
-
-    /**
-     * Sets up.
-     */
-    @Before
-    @SuppressWarnings("unchecked")
-    public void setUp() {
-        pub = mock(Publisher.class);
-        disp = mock(RequestIdDispatcher.class);
-        lock = new Object();
-        timers = mock(TimerManager.class);
-        timer = mock(TimerManager.Timer.class);
-        msgdata = new MyMessageData(PDP1);
-        queue = new LinkedList<>();
-        response = new PdpStatus();
-
-        doAnswer(new Answer<Object>() {
-            @Override
-            public Object answer(InvocationOnMock invocation) throws Throwable {
-                queue.add(invocation.getArgumentAt(0, QueueToken.class));
-                return null;
-            }
-        }).when(pub).enqueue(any());
-
-        when(timers.register(any(), any())).thenReturn(timer);
-
-        reqdata = new MyRequestData(
-                        new RequestDataParams().setModifyLock(lock).setPublisher(pub).setResponseDispatcher(disp));
-
-        reqdata.setName(PDP1);
-
-        msgdata = spy(msgdata);
-        reqdata = spy(reqdata);
-    }
-
-    @Test
-    public void testRequestData_Invalid() {
-        // null params
-        assertThatThrownBy(() -> new MyRequestData(null)).isInstanceOf(NullPointerException.class);
-
-        // invalid params
-        assertThatIllegalArgumentException().isThrownBy(() -> new MyRequestData(new RequestDataParams()));
-    }
-
-    @Test
-    public void testStartPublishing() {
-        reqdata.configure(msgdata);
-        reqdata.startPublishing();
-
-        verify(disp).register(eq(msgdata.getMessage().getRequestId()), any());
-        verify(timers).register(eq(PDP1), any());
-        verify(pub).enqueue(any());
-
-        QueueToken<PdpMessage> token = queue.poll();
-        assertNotNull(token);
-        assertSame(msgdata.getMessage(), token.get());
-
-
-        // invoking start() again has no effect - invocation counts remain the same
-        reqdata.startPublishing();
-        verify(disp, times(1)).register(eq(msgdata.getMessage().getRequestId()), any());
-        verify(timers, times(1)).register(eq(PDP1), any());
-        verify(pub, times(1)).enqueue(any());
-    }
-
-    @Test
-    public void testStopPublishing() {
-        reqdata.configure(msgdata);
-        reqdata.startPublishing();
-        reqdata.stopPublishing();
-
-        verify(disp).unregister(msgdata.getMessage().getRequestId());
-        verify(timer).cancel();
-
-
-        // invoking stop() again has no effect - invocation counts remain the same
-        reqdata.stopPublishing();
-
-        verify(disp, times(1)).unregister(msgdata.getMessage().getRequestId());
-        verify(timer, times(1)).cancel();
-    }
-
-    @Test
-    public void testConfigure() {
-        reqdata.configure(msgdata);
-        reqdata.startPublishing();
-
-        verify(disp).register(eq(msgdata.getMessage().getRequestId()), any());
-        verify(timers).register(eq(PDP1), any());
-        verify(pub).enqueue(any());
-
-        ServiceManager svcmgr = Whitebox.getInternalState(reqdata, "svcmgr");
-        assertEquals(PDP1 + " " + MY_MSG_TYPE, svcmgr.getName());
-
-
-        // bump this so we can verify that it is reset by configure()
-        reqdata.bumpRetryCount();
-
-        reqdata.configure(msgdata);
-        assertEquals(0, getRetryCount());
-    }
-
-    @Test
-    public void testEnqueue() {
-        reqdata.configure(msgdata);
-        reqdata.startPublishing();
-
-        // replace the message with a new message
-        reqdata.stopPublishing();
-        MyMessageData msgdata2 = new MyMessageData(PDP1);
-        reqdata.configure(msgdata2);
-        reqdata.startPublishing();
-
-        // should still only be one token in the queue
-        QueueToken<PdpMessage> token = queue.poll();
-        assertNull(queue.poll());
-        assertNotNull(token);
-        assertSame(msgdata2.getMessage(), token.get());
-
-        // null out the token
-        token.replaceItem(null);
-
-        // enqueue a new message
-        reqdata.stopPublishing();
-        MyMessageData msgdata3 = new MyMessageData(PDP1);
-        reqdata.configure(msgdata3);
-        reqdata.startPublishing();
-
-        // a new token should have been placed in the queue
-        QueueToken<PdpMessage> token2 = queue.poll();
-        assertTrue(token != token2);
-        assertNull(queue.poll());
-        assertNotNull(token2);
-        assertSame(msgdata3.getMessage(), token2.get());
-    }
-
-    @Test
-    public void testResetRetryCount_testBumpRetryCount() {
-        when(msgdata.getMaxRetryCount()).thenReturn(2);
-
-        reqdata.configure(msgdata);
-
-        assertEquals(0, getRetryCount());
-        assertTrue(reqdata.bumpRetryCount());
-        assertTrue(reqdata.bumpRetryCount());
-
-        // limit should now be reached and it should go no further
-        assertFalse(reqdata.bumpRetryCount());
-        assertFalse(reqdata.bumpRetryCount());
-
-        assertEquals(2, getRetryCount());
-
-        reqdata.resetRetryCount();
-        assertEquals(0, getRetryCount());
-    }
-
-    @Test
-    public void testRetryCountExhausted() {
-        reqdata.configure(msgdata);
-
-        reqdata.retryCountExhausted();
-
-        verify(reqdata).allCompleted();
-    }
-
-    @Test
-    public void testProcessResponse() {
-        reqdata.configure(msgdata);
-        reqdata.startPublishing();
-
-        invokeProcessResponse();
-
-        verify(reqdata).stopPublishing();
-        verify(msgdata).checkResponse(response);
-        verify(msgdata).completed();
-    }
-
-    @Test
-    public void testProcessResponse_NotPublishing() {
-        reqdata.configure(msgdata);
-        reqdata.startPublishing();
-
-        reqdata.stopPublishing();
-
-        invokeProcessResponse();
-
-        // only invocation should have been the one before calling invokeProcessResponse()
-        verify(reqdata, times(1)).stopPublishing();
-
-        verify(msgdata, never()).checkResponse(response);
-        verify(msgdata, never()).completed();
-    }
-
-    @Test
-    public void testProcessResponse_NotActive() {
-        reqdata.configure(msgdata);
-        reqdata.startPublishing();
-
-        when(reqdata.isActive()).thenReturn(false);
-
-        invokeProcessResponse();
-
-        // it should still stop publishing
-        verify(reqdata).stopPublishing();
-
-        verify(msgdata, never()).checkResponse(response);
-        verify(msgdata, never()).completed();
-    }
-
-    @Test
-    public void testProcessResponse_ResponseFailed() {
-        reqdata.configure(msgdata);
-        reqdata.startPublishing();
-
-        when(msgdata.checkResponse(response)).thenReturn("failed");
-
-        invokeProcessResponse();
-
-        verify(reqdata).stopPublishing();
-        verify(msgdata).checkResponse(response);
-
-        verify(msgdata, never()).completed();
-        verify(msgdata).mismatch("failed");
-    }
-
-    @Test
-    public void testHandleTimeout() {
-        reqdata.configure(msgdata);
-        reqdata.startPublishing();
-
-        // remove it from the queue
-        queue.poll().replaceItem(null);
-
-        invokeTimeoutHandler();
-
-        // count should have been bumped
-        assertEquals(1, getRetryCount());
-
-        // should have invoked startPublishing() a second time
-        verify(reqdata, times(2)).startPublishing();
-    }
-
-    @Test
-    public void testHandleTimeout_NotPublishing() {
-        reqdata.configure(msgdata);
-        reqdata.startPublishing();
-
-        reqdata.stopPublishing();
-
-        invokeTimeoutHandler();
-
-        // should NOT have invoked startPublishing() a second time
-        verify(reqdata, times(1)).startPublishing();
-        verify(reqdata, never()).retryCountExhausted();
-    }
-
-    @Test
-    public void testHandleTimeout_NotActive() {
-        reqdata.configure(msgdata);
-        reqdata.startPublishing();
-
-        when(reqdata.isActive()).thenReturn(false);
-
-        invokeTimeoutHandler();
-
-        // should NOT have invoked startPublishing() a second time
-        verify(reqdata, times(1)).startPublishing();
-        verify(reqdata, never()).retryCountExhausted();
-    }
-
-    @Test
-    public void testHandleTimeout_StillInQueue() {
-        reqdata.configure(msgdata);
-        reqdata.startPublishing();
-
-        reqdata.bumpRetryCount();
-
-        invokeTimeoutHandler();
-
-        // count should reset the count
-        assertEquals(0, getRetryCount());
-
-        // should have invoked startPublishing() a second time
-        verify(reqdata, times(2)).startPublishing();
-    }
-
-    @Test
-    public void testHandleTimeout_RetryExhausted() {
-        reqdata.configure(msgdata);
-        reqdata.startPublishing();
-
-        // exhaust the count
-        reqdata.bumpRetryCount();
-        reqdata.bumpRetryCount();
-        reqdata.bumpRetryCount();
-
-        // remove it from the queue
-        queue.poll().replaceItem(null);
-
-        invokeTimeoutHandler();
-
-        // should NOT have invoked startPublishing() a second time
-        verify(reqdata, times(1)).startPublishing();
-
-        verify(reqdata).retryCountExhausted();
-    }
-
-    @Test
-    public void testGetName_testSetName() {
-        reqdata.setName("abc");
-        assertEquals("abc", reqdata.getName());
-    }
-
-    @Test
-    public void testGetWrapper() {
-        reqdata.configure(msgdata);
-        assertSame(msgdata, reqdata.getWrapper());
-    }
-
-    /**
-     * Gets the retry count from the data.
-     * @return the current retry count
-     */
-    private int getRetryCount() {
-        return Whitebox.getInternalState(reqdata, "retryCount");
-    }
-
-    /**
-     * Gets the listener that was registered with the dispatcher and invokes it.
-     */
-    @SuppressWarnings("unchecked")
-    private void invokeProcessResponse() {
-        @SuppressWarnings("rawtypes")
-        ArgumentCaptor<TypedMessageListener> processResp = ArgumentCaptor.forClass(TypedMessageListener.class);
-
-        verify(disp).register(any(), processResp.capture());
-
-        processResp.getValue().onTopicEvent(CommInfrastructure.NOOP, PapConstants.TOPIC_POLICY_PDP_PAP, response);
-    }
-
-    /**
-     * Gets the timeout handler that was registered with the timer manager and invokes it.
-     */
-    @SuppressWarnings("unchecked")
-    private void invokeTimeoutHandler() {
-        @SuppressWarnings("rawtypes")
-        ArgumentCaptor<Consumer> timeoutHdlr = ArgumentCaptor.forClass(Consumer.class);
-
-        verify(timers).register(any(), timeoutHdlr.capture());
-
-        timeoutHdlr.getValue().accept(PDP1);
-    }
-
-    private class MyRequestData extends RequestData {
-
-        public MyRequestData(RequestDataParams params) {
-            super(params);
-        }
-
-        @Override
-        protected boolean isActive() {
-            return true;
-        }
-
-        @Override
-        protected void allCompleted() {
-            // do nothing
-        }
-    }
-
-    private class MyMessageData extends MessageData {
-
-        public MyMessageData(String pdpName) {
-            super(new PdpStateChange(), 1, timers);
-
-            PdpStateChange msg = (PdpStateChange) getMessage();
-            msg.setName(pdpName);
-            msg.setState(PdpState.ACTIVE);
-        }
-
-        @Override
-        public String getType() {
-            return MY_MSG_TYPE;
-        }
-
-        @Override
-        public void mismatch(String reason) {
-            // do nothing
-        }
-
-        @Override
-        public void completed() {
-            // do nothing
-        }
-
-        @Override
-        public String checkResponse(PdpStatus response) {
-            // always valid - return null
-            return null;
-        }
-    }
-}
diff --git a/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/MessageDataTest.java b/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/MessageDataTest.java
deleted file mode 100644
index 68b02635..00000000
--- a/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/MessageDataTest.java
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * ONAP PAP
- * ================================================================================
- * Copyright (C) 2019 AT&T Intellectual Property. 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.
- * 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.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.policy.pap.main.comm.msgdata;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertSame;
-import static org.mockito.Mockito.mock;
-
-import org.junit.Before;
-import org.junit.Test;
-import org.onap.policy.models.pdp.concepts.PdpStateChange;
-import org.onap.policy.models.pdp.concepts.PdpStatus;
-import org.onap.policy.pap.main.comm.TimerManager;
-
-public class MessageDataTest {
-    private static final int RETRIES = 1;
-
-    private MyData data;
-    private TimerManager timers;
-
-    /**
-     * Sets up.
-     */
-    @Before
-    public void setUp() {
-        timers = mock(TimerManager.class);
-
-        data = new MyData();
-    }
-
-    @Test
-    public void testGetMessage() {
-        assertNotNull(data.getMessage());
-    }
-
-    @Test
-    public void testGetType() {
-        assertEquals(PdpStateChange.class.getSimpleName(), data.getType());
-    }
-
-    @Test
-    public void testGetMaxRetryCount() {
-        assertEquals(RETRIES, data.getMaxRetryCount());
-    }
-
-    @Test
-    public void testGetTimers() {
-        assertSame(timers, data.getTimers());
-    }
-
-    private class MyData extends MessageData {
-
-        public MyData() {
-            super(new PdpStateChange(), RETRIES, timers);
-        }
-
-        @Override
-        public void mismatch(String reason) {
-            // do nothing
-        }
-
-        @Override
-        public void completed() {
-            // do nothing
-        }
-
-        @Override
-        public String checkResponse(PdpStatus response) {
-            // always succeed
-            return null;
-        }
-    }
-}
diff --git a/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/RequestImplTest.java b/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/RequestImplTest.java
new file mode 100644
index 00000000..2446533e
--- /dev/null
+++ b/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/RequestImplTest.java
@@ -0,0 +1,560 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP PAP
+ * ================================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. 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.
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.pap.main.comm.msgdata;
+
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.models.pdp.concepts.PdpMessage;
+import org.onap.policy.models.pdp.concepts.PdpStateChange;
+import org.onap.policy.models.pdp.concepts.PdpStatus;
+import org.onap.policy.models.pdp.concepts.PdpUpdate;
+import org.onap.policy.pap.main.comm.CommonRequestBase;
+import org.onap.policy.pap.main.comm.QueueToken;
+import org.onap.policy.pap.main.parameters.RequestParams;
+
+public class RequestImplTest extends CommonRequestBase {
+    private static final int MY_PRIORITY = 10;
+
+    private MyRequest req;
+    private PdpStatus response;
+    private PdpStateChange msg;
+
+    /**
+     * Sets up.
+     * @throws Exception if an error occurs
+     */
+    @Before
+    public void setUp() throws Exception {
+        super.setUp();
+
+        response = new PdpStatus();
+        msg = new PdpStateChange();
+
+        response.setName(PDP1);
+        msg.setName(PDP1);
+
+        req = new MyRequest(reqParams, MY_REQ_NAME, msg);
+        req.setListener(listener);
+    }
+
+    @Test
+    public void testRequest_InvalidArgs() {
+        // null params
+        assertThatThrownBy(() -> new MyRequest(null, MY_REQ_NAME, msg)).isInstanceOf(NullPointerException.class);
+
+        // null name
+        assertThatThrownBy(() -> new MyRequest(reqParams, null, msg)).isInstanceOf(NullPointerException.class);
+
+        // null message
+        assertThatThrownBy(() -> new MyRequest(reqParams, MY_REQ_NAME, null)).isInstanceOf(NullPointerException.class);
+
+        // invalid params
+        assertThatIllegalArgumentException().isThrownBy(() -> new MyRequest(new RequestParams(), MY_REQ_NAME, msg));
+    }
+
+    @Test
+    public void testReconfigure_WrongMsgClass() {
+        assertThatIllegalArgumentException().isThrownBy(() -> req.reconfigure(new PdpUpdate(), null))
+                        .withMessage("expecting PdpStateChange instead of PdpUpdate");
+    }
+
+    @Test
+    public void testReconfigure_NotPublishing() {
+
+        // replace the message with a new message
+        req.reconfigure(new PdpStateChange(), null);
+
+        // nothing should have been placed in the queue
+        assertNull(queue.poll());
+    }
+
+    @Test
+    public void testRequestImpl_testReconfigure_Publishing() {
+        req.startPublishing();
+
+        // replace the message with a new message
+        PdpStateChange msg2 = new PdpStateChange();
+        req.reconfigure(msg2, null);
+
+        // should only be one token in the queue
+        QueueToken<PdpMessage> token = queue.poll();
+        assertNotNull(token);
+        assertSame(msg2, token.get());
+
+        verify(dispatcher).register(eq(msg.getRequestId()), any());
+        verify(timers).register(eq(msg.getRequestId()), any());
+        verify(publisher).enqueue(token);
+
+        verify(dispatcher).unregister(eq(msg.getRequestId()));
+
+        verify(dispatcher).register(eq(msg2.getRequestId()), any());
+        verify(timers).register(eq(msg2.getRequestId()), any());
+        verify(publisher).enqueue(any());
+    }
+
+    @Test
+    public void testReconfigure_PublishingNullToken() {
+        req.startPublishing();
+
+        // replace the message with a new message
+        PdpStateChange msg2 = new PdpStateChange();
+        req.reconfigure(msg2, null);
+
+        // should only be one token in the queue
+        QueueToken<PdpMessage> token = queue.poll();
+        assertNotNull(token);
+        assertSame(msg2, token.get());
+    }
+
+    @Test
+    public void testReconfigure_PublishingNewToken() {
+        req.startPublishing();
+
+        // null out the original token so it isn't reused
+        QueueToken<PdpMessage> token = queue.poll();
+        assertNotNull(token);
+        token.replaceItem(null);
+
+        QueueToken<PdpMessage> token2 = new QueueToken<>(new PdpStateChange());
+
+        // replace the message with a new message
+        PdpStateChange msg2 = new PdpStateChange();
+        req.reconfigure(msg2, token2);
+
+        // new token should have the new message
+        token = queue.poll();
+        assertSame(msg2, token.get());
+
+        assertNull(queue.poll());
+    }
+
+    @Test
+    public void testIsPublishing() {
+        assertFalse(req.isPublishing());
+
+        req.startPublishing();
+        assertTrue(req.isPublishing());
+
+        req.stopPublishing();
+        assertFalse(req.isPublishing());
+    }
+
+    @Test
+    public void testStartPublishingQueueToken() {
+        req.startPublishing(null);
+
+        assertTrue(req.isPublishing());
+
+        verify(dispatcher).register(eq(msg.getRequestId()), any());
+        verify(timers).register(eq(msg.getRequestId()), any());
+        verify(publisher).enqueue(any());
+
+        QueueToken<PdpMessage> token = queue.poll();
+        assertNotNull(token);
+        assertSame(msg, token.get());
+
+
+        // invoking start() again has no effect - invocation counts remain the same
+        req.startPublishing(null);
+        verify(dispatcher, times(1)).register(any(), any());
+        verify(timers, times(1)).register(any(), any());
+        verify(publisher, times(1)).enqueue(any());
+        assertNull(queue.poll());
+    }
+
+    @Test
+    public void testStartPublishingQueueToken_NoListener() {
+        req.setListener(null);
+        assertThatIllegalStateException().isThrownBy(() -> req.startPublishing())
+                        .withMessage("listener has not been set");
+    }
+
+    @Test
+    public void testStartPublishing() {
+        req.startPublishing();
+
+        assertTrue(req.isPublishing());
+
+        verify(dispatcher).register(eq(msg.getRequestId()), any());
+        verify(timers).register(eq(msg.getRequestId()), any());
+        verify(publisher).enqueue(any());
+
+        QueueToken<PdpMessage> token = queue.poll();
+        assertNotNull(token);
+        assertSame(msg, token.get());
+
+
+        // invoking start() again has no effect - invocation counts remain the same
+        req.startPublishing();
+        verify(dispatcher, times(1)).register(any(), any());
+        verify(timers, times(1)).register(any(), any());
+        verify(publisher, times(1)).enqueue(any());
+        assertNull(queue.poll());
+    }
+
+    @Test
+    public void testReplaceToken_NullNewToken() {
+        req.startPublishing(null);
+        assertSame(msg, queue.poll().get());
+    }
+
+    @Test
+    public void testReplaceToken_NullOldToken() {
+        QueueToken<PdpMessage> token = new QueueToken<>(new PdpStateChange());
+
+        req.startPublishing(token);
+        assertNull(queue.poll());
+        assertSame(msg, token.get());
+    }
+
+    @Test
+    public void testReplaceToken_SameToken() {
+        req.startPublishing();
+
+        QueueToken<PdpMessage> token = queue.poll();
+        req.startPublishing(token);
+
+        // nothing else should have been enqueued
+        assertNull(queue.poll());
+
+        assertSame(msg, token.get());
+    }
+
+    @Test
+    public void testReplaceToken_DifferentToken() {
+        req.startPublishing();
+
+        QueueToken<PdpMessage> token2 = new QueueToken<>(new PdpStateChange());
+        req.startPublishing(token2);
+
+        QueueToken<PdpMessage> token = queue.poll();
+
+        // old token should still have the message
+        assertSame(msg, token.get());
+
+        // should not have added new token to the queue
+        assertNull(queue.poll());
+
+        // new token should have been nulled out
+        assertNull(token2.get());
+    }
+
+    @Test
+    public void testStopPublishing() {
+        // not publishing yet
+        req.stopPublishing();
+        assertFalse(req.isPublishing());
+
+        // now we'll publish
+        req.startPublishing();
+
+        req.stopPublishing();
+        assertFalse(req.isPublishing());
+
+        // should only be one token in the queue - should be nulled out
+        QueueToken<PdpMessage> token = queue.poll();
+        assertNotNull(token);
+        assertNull(token.get());
+
+        verify(dispatcher).unregister(eq(msg.getRequestId()));
+        verify(timer).cancel();
+    }
+
+    @Test
+    public void testStopPublishingBoolean_NotPublishing() {
+        assertNull(req.stopPublishing(false));
+    }
+
+    @Test
+    public void testStopPublishingBoolean_TruePublishing() {
+        req.startPublishing();
+
+        assertNull(req.stopPublishing(true));
+
+        // should be nulled out
+        QueueToken<PdpMessage> token = queue.poll();
+        assertNotNull(token);
+        assertNull(token.get());
+
+        verify(dispatcher).unregister(eq(msg.getRequestId()));
+        verify(timer).cancel();
+
+        // if start publishing again - should use a new token
+        req.startPublishing();
+        QueueToken<PdpMessage> token2 = queue.poll();
+        assertNotNull(token2);
+        assertTrue(token2 != token);
+        assertSame(msg, token2.get());
+    }
+
+    @Test
+    public void testStopPublishingBoolean_FalsePublishing() {
+        req.startPublishing();
+
+        QueueToken<PdpMessage> token = req.stopPublishing(false);
+        assertNotNull(token);
+        assertSame(token, queue.poll());
+
+        // should not be nulled out
+        assertSame(msg, token.get());
+
+        verify(dispatcher).unregister(eq(msg.getRequestId()));
+        verify(timer).cancel();
+
+        // if start publishing again - should use a new token
+        req.startPublishing();
+        QueueToken<PdpMessage> token2 = queue.poll();
+        assertNotNull(token2);
+        assertTrue(token2 != token);
+        assertSame(msg, token2.get());
+    }
+
+    @Test
+    public void testEnqueue() {
+        req.startPublishing();
+
+        // replace the message with a new message
+        PdpStateChange msg2 = new PdpStateChange();
+        req.reconfigure(msg2, null);
+
+        // should still only be one token in the queue
+        QueueToken<PdpMessage> token = queue.poll();
+        assertNull(queue.poll());
+        assertNotNull(token);
+        assertSame(msg2, token.get());
+
+        // force the token to be nulled out
+        req.stopPublishing();
+
+        // enqueue a new message
+        PdpStateChange msg3 = new PdpStateChange();
+        req.reconfigure(msg3, null);
+        req.startPublishing();
+
+        // a new token should have been placed in the queue
+        QueueToken<PdpMessage> token2 = queue.poll();
+        assertTrue(token != token2);
+        assertNull(queue.poll());
+        assertNotNull(token2);
+        assertSame(msg3, token2.get());
+    }
+
+    @Test
+    public void testResetRetryCount_testBumpRetryCount() {
+        req = new MyRequest(new RequestParams().setMaxRetryCount(2).setModifyLock(lock).setPublisher(publisher)
+                        .setResponseDispatcher(dispatcher).setTimers(timers), MY_REQ_NAME, msg);
+        req.setListener(listener);
+
+        assertEquals(0, req.getRetryCount());
+        assertTrue(req.bumpRetryCount());
+        assertTrue(req.bumpRetryCount());
+
+        // limit should now be reached and it should go no further
+        assertFalse(req.bumpRetryCount());
+        assertFalse(req.bumpRetryCount());
+
+        assertEquals(2, req.getRetryCount());
+
+        req.resetRetryCount();
+        assertEquals(0, req.getRetryCount());
+    }
+
+    @Test
+    public void testProcessResponse() {
+        req.startPublishing();
+
+        invokeProcessResponse(response);
+
+        verify(listener).success(PDP1);
+        verify(listener, never()).failure(any(), any());
+    }
+
+    @Test
+    public void testProcessResponse_NotPublishing() {
+        // force registration with the dispatcher - needed by invokeProcessResponse(response)
+        req.startPublishing();
+        req.stopPublishing();
+
+        invokeProcessResponse(response);
+
+        verify(listener, never()).success(any());
+        verify(listener, never()).failure(any(), any());
+    }
+
+    @Test
+    public void testProcessResponse_ResponseFailed() {
+        req.startPublishing();
+
+        response.setName(DIFFERENT);
+
+        invokeProcessResponse(response);
+
+        verify(listener, never()).success(any());
+        verify(listener).failure(DIFFERENT, "PDP name does not match");
+    }
+
+    @Test
+    public void testHandleTimeout() {
+        req.startPublishing();
+
+        // remove it from the queue
+        queue.poll().replaceItem(null);
+
+        invokeTimeoutHandler();
+
+        // count should have been bumped
+        assertEquals(1, req.getRetryCount());
+
+        // should have invoked startPublishing() a second time
+        verify(dispatcher, times(2)).register(eq(msg.getRequestId()), any());
+    }
+
+    @Test
+    public void testHandleTimeout_NotPublishing() {
+        req.startPublishing();
+
+        req.stopPublishing();
+
+        invokeTimeoutHandler();
+
+        // should NOT have invoked startPublishing() a second time
+        verify(dispatcher, times(1)).register(eq(msg.getRequestId()), any());
+        verify(listener, never()).retryCountExhausted();
+    }
+
+    @Test
+    public void testHandleTimeout_RetryExhausted() {
+        req.startPublishing();
+
+        // exhaust the count
+        req.bumpRetryCount();
+        req.bumpRetryCount();
+        req.bumpRetryCount();
+
+        // remove it from the queue
+        queue.poll().replaceItem(null);
+
+        invokeTimeoutHandler();
+
+        // should NOT have invoked startPublishing() a second time
+        verify(dispatcher, times(1)).register(eq(msg.getRequestId()), any());
+
+        verify(listener).retryCountExhausted();
+    }
+
+    @Test
+    public void testCheckResponse_Matched() {
+        req.startPublishing();
+
+        invokeProcessResponse(response);
+
+        verify(listener).success(PDP1);
+        verify(listener, never()).failure(any(), any());
+    }
+
+    @Test
+    public void testCheckResponse_NullName() {
+        req.startPublishing();
+
+        response.setName(null);
+
+        invokeProcessResponse(response);
+
+        verify(listener, never()).success(any());
+        verify(listener).failure(null, "null PDP name");
+    }
+
+    @Test
+    public void testCheckResponse_MismatchedName() {
+        req.startPublishing();
+
+        response.setName(DIFFERENT);
+
+        invokeProcessResponse(response);
+
+        verify(listener, never()).success(any());
+        verify(listener).failure(DIFFERENT, "PDP name does not match");
+    }
+
+    @Test
+    public void testCheckResponse_MismatchedNameWithBroadcast() {
+        msg.setName(null);
+        req.startPublishing();
+
+        response.setName(DIFFERENT);
+
+        invokeProcessResponse(response);
+
+        verify(listener).success(DIFFERENT);
+        verify(listener, never()).failure(any(), any());
+    }
+
+    @Test
+    public void testGetName() {
+        assertEquals(MY_REQ_NAME, req.getName());
+    }
+
+    @Test
+    public void testGetMessage() {
+        assertSame(msg, req.getMessage());
+
+        PdpStateChange msg2 = new PdpStateChange();
+        req.reconfigure(msg2, null);
+        assertSame(msg2, req.getMessage());
+    }
+
+    @Test
+    public void testGetParams() {
+        assertSame(reqParams, req.getParams());
+    }
+
+    private class MyRequest extends RequestImpl {
+
+        public MyRequest(RequestParams params, String name, PdpMessage message) {
+            super(params, name, message);
+        }
+
+        @Override
+        public int getPriority() {
+            return MY_PRIORITY;
+        }
+
+        @Override
+        public boolean isSameContent(Request other) {
+            return false;
+        }
+    }
+}
diff --git a/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/StateChangeDataTest.java b/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/StateChangeDataTest.java
deleted file mode 100644
index 029775fa..00000000
--- a/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/StateChangeDataTest.java
+++ /dev/null
@@ -1,122 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * ONAP PAP
- * ================================================================================
- * Copyright (C) 2019 AT&T Intellectual Property. 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.
- * 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.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.policy.pap.main.comm.msgdata;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertSame;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-import org.junit.Before;
-import org.junit.Test;
-import org.onap.policy.models.pdp.concepts.PdpStateChange;
-import org.onap.policy.models.pdp.concepts.PdpStatus;
-import org.onap.policy.models.pdp.enums.PdpState;
-import org.onap.policy.pap.main.comm.TimerManager;
-import org.onap.policy.pap.main.parameters.PdpModifyRequestMapParams;
-import org.onap.policy.pap.main.parameters.PdpParameters;
-import org.onap.policy.pap.main.parameters.PdpStateChangeParameters;
-
-public class StateChangeDataTest {
-    private static final String MY_NAME = "my-name";
-    private static final String DIFFERENT = "different";
-    private static final PdpState MY_STATE = PdpState.SAFE;
-    private static final PdpState DIFF_STATE = PdpState.TERMINATED;
-    private static final int RETRIES = 1;
-
-    private MyData data;
-    private PdpModifyRequestMapParams params;
-    private TimerManager timers;
-    private PdpStatus response;
-
-    /**
-     * Sets up.
-     */
-    @Before
-    public void setUp() {
-        timers = mock(TimerManager.class);
-        response = new PdpStatus();
-        PdpParameters pdpParams = mock(PdpParameters.class);
-        PdpStateChangeParameters stateParams = mock(PdpStateChangeParameters.class);
-
-        when(stateParams.getMaxRetryCount()).thenReturn(RETRIES);
-        when(pdpParams.getStateChangeParameters()).thenReturn(stateParams);
-
-        params = new PdpModifyRequestMapParams().setParams(pdpParams).setStateChangeTimers(timers);
-
-        response.setName(MY_NAME);
-        response.setState(MY_STATE);
-
-        data = new MyData();
-    }
-
-    @Test
-    public void testGetMaxRetryCount() {
-        assertEquals(RETRIES, data.getMaxRetryCount());
-    }
-
-    @Test
-    public void testGetTimers() {
-        assertSame(timers, data.getTimers());
-    }
-
-    @Test
-    public void testStateChangeCheckResponse() {
-        assertNull(data.checkResponse(response));
-    }
-
-    @Test
-    public void testStateChangeCheckResponse_MismatchedName() {
-        response.setName(DIFFERENT);
-
-        assertEquals("name does not match", data.checkResponse(response));
-    }
-
-    @Test
-    public void testStateChangeCheckResponse_MismatchedState() {
-        response.setState(DIFF_STATE);
-
-        assertEquals("state is TERMINATED, but expected SAFE", data.checkResponse(response));
-    }
-
-    private class MyData extends StateChangeData {
-
-        public MyData() {
-            super(new PdpStateChange(), params);
-
-            PdpStateChange msg = (PdpStateChange) getMessage();
-
-            msg.setName(MY_NAME);
-            msg.setState(MY_STATE);
-        }
-
-        @Override
-        public void mismatch(String reason) {
-            // do nothing
-        }
-
-        @Override
-        public void completed() {
-            // do nothing
-        }
-    }
-}
diff --git a/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/StateChangeReqTest.java b/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/StateChangeReqTest.java
new file mode 100644
index 00000000..35531145
--- /dev/null
+++ b/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/StateChangeReqTest.java
@@ -0,0 +1,113 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP PAP
+ * ================================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. 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.
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.pap.main.comm.msgdata;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.models.pdp.concepts.PdpStateChange;
+import org.onap.policy.models.pdp.concepts.PdpStatus;
+import org.onap.policy.models.pdp.concepts.PdpUpdate;
+import org.onap.policy.pap.main.comm.CommonRequestBase;
+
+public class StateChangeReqTest extends CommonRequestBase {
+
+    private StateChangeReq data;
+    private PdpStatus response;
+    private PdpStateChange msg;
+
+    /**
+     * Sets up.
+     * @throws Exception if an error occurs
+     */
+    @Before
+    public void setUp() throws Exception {
+        super.setUp();
+
+        response = new PdpStatus();
+
+        response.setName(MY_NAME);
+        response.setState(MY_STATE);
+
+        msg = new PdpStateChange();
+        msg.setName(MY_NAME);
+        msg.setState(MY_STATE);
+
+        data = new StateChangeReq(reqParams, MY_REQ_NAME, msg);
+    }
+
+    @Test
+    public void testGetMessage() {
+        assertEquals(MY_REQ_NAME, data.getName());
+        assertSame(msg, data.getMessage());
+    }
+
+    @Test
+    public void testCheckResponse() {
+        assertNull(data.checkResponse(response));
+    }
+
+    @Test
+    public void testCheckResponse_NullName() {
+        response.setName(null);
+
+        assertEquals("null PDP name", data.checkResponse(response));
+    }
+
+    @Test
+    public void testCheckResponse_NullMsgName() {
+        msg.setName(null);
+
+        assertEquals(null, data.checkResponse(response));
+    }
+
+    @Test
+    public void testCheckResponse_MismatchedState() {
+        response.setState(DIFF_STATE);
+
+        assertEquals("state is TERMINATED, but expected SAFE", data.checkResponse(response));
+    }
+
+    @Test
+    public void isSameContent() {
+        PdpStateChange msg2 = new PdpStateChange();
+        msg2.setName("world");
+        msg2.setState(MY_STATE);
+        assertTrue(data.isSameContent(new StateChangeReq(reqParams, MY_REQ_NAME, msg2)));
+
+        // different state
+        msg2.setState(DIFF_STATE);
+        assertFalse(data.isSameContent(new StateChangeReq(reqParams, MY_REQ_NAME, msg2)));
+
+        // different request type
+        assertFalse(data.isSameContent(new UpdateReq(reqParams, MY_REQ_NAME, new PdpUpdate())));
+    }
+
+    @Test
+    public void testGetPriority() {
+        assertTrue(data.getPriority() < new UpdateReq(reqParams, MY_REQ_NAME, new PdpUpdate()).getPriority());
+    }
+}
diff --git a/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/UpdateDataTest.java b/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/UpdateDataTest.java
deleted file mode 100644
index bac85ed0..00000000
--- a/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/UpdateDataTest.java
+++ /dev/null
@@ -1,189 +0,0 @@
-/*
- * ============LICENSE_START=======================================================
- * ONAP PAP
- * ================================================================================
- * Copyright (C) 2019 AT&T Intellectual Property. 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.
- * 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.
- * ============LICENSE_END=========================================================
- */
-
-package org.onap.policy.pap.main.comm.msgdata;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertSame;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-import java.util.stream.Collectors;
-import org.junit.Before;
-import org.junit.Test;
-import org.onap.policy.models.pdp.concepts.PdpStatus;
-import org.onap.policy.models.pdp.concepts.PdpUpdate;
-import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicy;
-import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicyIdentifier;
-import org.onap.policy.pap.main.comm.TimerManager;
-import org.onap.policy.pap.main.parameters.PdpModifyRequestMapParams;
-import org.onap.policy.pap.main.parameters.PdpParameters;
-import org.onap.policy.pap.main.parameters.PdpUpdateParameters;
-
-public class UpdateDataTest {
-    private static final String MY_NAME = "my-name";
-    private static final String DIFFERENT = "different";
-    private static final int RETRIES = 1;
-
-    private MyData data;
-    private PdpModifyRequestMapParams params;
-    private TimerManager timers;
-    private PdpUpdate update;
-    private PdpStatus response;
-
-    /**
-     * Sets up.
-     */
-    @Before
-    public void setUp() {
-        timers = mock(TimerManager.class);
-        response = new PdpStatus();
-        PdpParameters pdpParams = mock(PdpParameters.class);
-        PdpUpdateParameters stateParams = mock(PdpUpdateParameters.class);
-
-        when(stateParams.getMaxRetryCount()).thenReturn(RETRIES);
-        when(pdpParams.getUpdateParameters()).thenReturn(stateParams);
-
-        params = new PdpModifyRequestMapParams().setParams(pdpParams).setUpdateTimers(timers);
-
-        update = makeUpdate();
-
-        response.setName(MY_NAME);
-        response.setPdpGroup(update.getPdpGroup());
-        response.setPdpSubgroup(update.getPdpSubgroup());
-        response.setPolicies(policyToIdent(update.getPolicies()));
-
-        data = new MyData(update);
-    }
-
-    @Test
-    public void testGetMaxRetryCount() {
-        assertEquals(RETRIES, data.getMaxRetryCount());
-    }
-
-    @Test
-    public void testGetTimers() {
-        assertSame(timers, data.getTimers());
-    }
-
-    @Test
-    public void testUpdateCheckResponse() {
-        assertNull(data.checkResponse(response));
-    }
-
-    @Test
-    public void testUpdateDataCheckResponse_MismatchedName() {
-        response.setName(DIFFERENT);
-
-        assertEquals("name does not match", data.checkResponse(response));
-    }
-
-    @Test
-    public void testUpdateDataCheckResponse_MismatchedGroup() {
-        response.setPdpGroup(DIFFERENT);
-
-        assertEquals("group does not match", data.checkResponse(response));
-    }
-
-    @Test
-    public void testUpdateDataCheckResponse_MismatchedSubGroup() {
-        response.setPdpSubgroup(DIFFERENT);
-
-        assertEquals("subgroup does not match", data.checkResponse(response));
-    }
-
-    @Test
-    public void testUpdateDataCheckResponse_MismatchedPolicies() {
-        ArrayList<ToscaPolicy> policies = new ArrayList<>(update.getPolicies());
-        policies.set(0, makePolicy(DIFFERENT, "10.0.0"));
-
-        response.setPolicies(policyToIdent(policies));
-
-        assertEquals("policies do not match", data.checkResponse(response));
-    }
-
-    /**
-     * Converts a list of policies to their corresponding identifiers.
-     *
-     * @param policies policies to be converted
-     * @return a list of policy identifiers
-     */
-    private List<ToscaPolicyIdentifier> policyToIdent(List<ToscaPolicy> policies) {
-        return policies.stream().map(ToscaPolicy::getIdentifier).collect(Collectors.toList());
-    }
-
-    /**
-     * Makes an update message.
-     *
-     * @return a new update message
-     */
-    private PdpUpdate makeUpdate() {
-        PdpUpdate upd = new PdpUpdate();
-
-        upd.setDescription("update-description");
-        upd.setName(MY_NAME);
-        upd.setPdpGroup("group1-a");
-        upd.setPdpSubgroup("sub1-a");
-
-        ToscaPolicy policy1 = makePolicy("policy-1-a", "1.0.0");
-        ToscaPolicy policy2 = makePolicy("policy-2-a", "1.1.0");
-
-        upd.setPolicies(Arrays.asList(policy1, policy2));
-
-        return upd;
-    }
-
-    /**
-     * Creates a new policy.
-     *
-     * @param name policy name
-     * @param version policy version
-     * @return a new policy
-     */
-    private ToscaPolicy makePolicy(String name, String version) {
-        ToscaPolicy policy = new ToscaPolicy();
-
-        policy.setName(name);
-        policy.setVersion(version);
-
-        return policy;
-    }
-
-    private class MyData extends UpdateData {
-
-        public MyData(PdpUpdate message) {
-            super(message, params);
-        }
-
-        @Override
-        public void mismatch(String reason) {
-            // do nothing
-        }
-
-        @Override
-        public void completed() {
-            // do nothing
-        }
-    }
-}
diff --git a/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/UpdateReqTest.java b/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/UpdateReqTest.java
new file mode 100644
index 00000000..156e9c80
--- /dev/null
+++ b/main/src/test/java/org/onap/policy/pap/main/comm/msgdata/UpdateReqTest.java
@@ -0,0 +1,204 @@
+/*
+ * ============LICENSE_START=======================================================
+ * ONAP PAP
+ * ================================================================================
+ * Copyright (C) 2019 AT&T Intellectual Property. 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.
+ * 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.
+ * ============LICENSE_END=========================================================
+ */
+
+package org.onap.policy.pap.main.comm.msgdata;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.stream.Collectors;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.models.pdp.concepts.PdpStateChange;
+import org.onap.policy.models.pdp.concepts.PdpStatus;
+import org.onap.policy.models.pdp.concepts.PdpUpdate;
+import org.onap.policy.models.tosca.authorative.concepts.ToscaPolicy;
+import org.onap.policy.pap.main.comm.CommonRequestBase;
+
+public class UpdateReqTest extends CommonRequestBase {
+
+    private UpdateReq data;
+    private PdpUpdate update;
+    private PdpStatus response;
+
+    /**
+     * Sets up.
+     * @throws Exception if an error occurs
+     */
+    @Before
+    public void setUp() throws Exception {
+        super.setUp();
+
+        response = new PdpStatus();
+
+        update = makeUpdate();
+
+        response.setName(MY_NAME);
+        response.setPdpGroup(update.getPdpGroup());
+        response.setPdpSubgroup(update.getPdpSubgroup());
+        response.setPolicies(
+                        update.getPolicies().stream().map(ToscaPolicy::getIdentifier).collect(Collectors.toList()));
+
+        data = new UpdateReq(reqParams, MY_REQ_NAME, update);
+    }
+
+    @Test
+    public void testGetMessage() {
+        assertEquals(MY_REQ_NAME, data.getName());
+        assertSame(update, data.getMessage());
+    }
+
+    @Test
+    public void testCheckResponse() {
+        assertNull(data.checkResponse(response));
+    }
+
+    @Test
+    public void testCheckResponse_NullName() {
+        response.setName(null);
+
+        assertEquals("null PDP name", data.checkResponse(response));
+    }
+
+    @Test
+    public void testCheckResponse_NullMsgName() {
+        update.setName(null);
+
+        assertEquals(null, data.checkResponse(response));
+    }
+
+    @Test
+    public void testUpdateReqCheckResponse_MismatchedGroup() {
+        response.setPdpGroup(DIFFERENT);
+
+        assertEquals("group does not match", data.checkResponse(response));
+    }
+
+    @Test
+    public void testUpdateReqCheckResponse_MismatchedSubGroup() {
+        response.setPdpSubgroup(DIFFERENT);
+
+        assertEquals("subgroup does not match", data.checkResponse(response));
+    }
+
+    @Test
+    public void testUpdateReqCheckResponse_MismatchedPolicies() {
+        ArrayList<ToscaPolicy> policies = new ArrayList<>(update.getPolicies());
+        policies.set(0, makePolicy(DIFFERENT, "10.0.0"));
+
+        response.setPolicies(policies.stream().map(ToscaPolicy::getIdentifier).collect(Collectors.toList()));
+
+        assertEquals("policies do not match", data.checkResponse(response));
+    }
+
+    @Test
+    public void isSameContent() {
+        PdpUpdate msg2 = new PdpUpdate(update);
+        msg2.setName("world");
+        assertTrue(data.isSameContent(new UpdateReq(reqParams, MY_REQ_NAME, msg2)));
+
+        // different request type
+        assertFalse(data.isSameContent(new StateChangeReq(reqParams, MY_REQ_NAME, new PdpStateChange())));
+    }
+
+    @Test
+    public void isSameContent_BothGroupNamesNull() {
+        PdpUpdate msg2 = new PdpUpdate(update);
+        msg2.setPdpGroup(null);
+        update.setPdpGroup(null);
+        assertTrue(data.isSameContent(new UpdateReq(reqParams, MY_REQ_NAME, msg2)));
+    }
+
+    @Test
+    public void isSameContent_BothSubGroupNamesNull() {
+        PdpUpdate msg2 = new PdpUpdate(update);
+        msg2.setPdpSubgroup(null);
+        update.setPdpSubgroup(null);
+        assertTrue(data.isSameContent(new UpdateReq(reqParams, MY_REQ_NAME, msg2)));
+    }
+
+    @Test
+    public void isSameContent_DiffGroup() {
+        PdpUpdate msg2 = new PdpUpdate(update);
+        msg2.setPdpGroup(null);
+        assertFalse(data.isSameContent(new UpdateReq(reqParams, MY_REQ_NAME, msg2)));
+
+        msg2.setPdpGroup(DIFFERENT);
+        assertFalse(data.isSameContent(new UpdateReq(reqParams, MY_REQ_NAME, msg2)));
+
+        update.setPdpGroup(null);
+        assertFalse(data.isSameContent(new UpdateReq(reqParams, MY_REQ_NAME, msg2)));
+    }
+
+    @Test
+    public void isSameContent_DiffSubGroup() {
+        PdpUpdate msg2 = new PdpUpdate(update);
+        msg2.setPdpSubgroup(null);
+        assertFalse(data.isSameContent(new UpdateReq(reqParams, MY_REQ_NAME, msg2)));
+
+        msg2.setPdpSubgroup(DIFFERENT);
+        assertFalse(data.isSameContent(new UpdateReq(reqParams, MY_REQ_NAME, msg2)));
+
+        update.setPdpSubgroup(null);
+        assertFalse(data.isSameContent(new UpdateReq(reqParams, MY_REQ_NAME, msg2)));
+    }
+
+    @Test
+    public void isSameContent_DiffPolicies() {
+        PdpUpdate msg2 = new PdpUpdate(update);
+
+        ArrayList<ToscaPolicy> policies = new ArrayList<>(update.getPolicies());
+        policies.set(0, makePolicy(DIFFERENT, "10.0.0"));
+        msg2.setPolicies(policies);
+
+        assertFalse(data.isSameContent(new UpdateReq(reqParams, MY_REQ_NAME, msg2)));
+    }
+
+    @Test
+    public void testGetPriority() {
+        assertTrue(data.getPriority() > new StateChangeReq(reqParams, MY_REQ_NAME, new PdpStateChange()).getPriority());
+    }
+
+    /**
+     * Makes an update message.
+     *
+     * @return a new update message
+     */
+    private PdpUpdate makeUpdate() {
+        PdpUpdate upd = new PdpUpdate();
+
+        upd.setDescription("update-description");
+        upd.setName(MY_NAME);
+        upd.setPdpGroup(MY_GROUP);
+        upd.setPdpSubgroup(MY_SUBGROUP);
+
+        ToscaPolicy policy1 = makePolicy("policy-1-a", "1.0.0");
+        ToscaPolicy policy2 = makePolicy("policy-2-a", "1.1.0");
+
+        upd.setPolicies(Arrays.asList(policy1, policy2));
+
+        return upd;
+    }
+}
diff --git a/main/src/test/java/org/onap/policy/pap/main/parameters/TestPdpModifyRequestMapParams.java b/main/src/test/java/org/onap/policy/pap/main/parameters/TestPdpModifyRequestMapParams.java
index 3e691899..b9dde727 100644
--- a/main/src/test/java/org/onap/policy/pap/main/parameters/TestPdpModifyRequestMapParams.java
+++ b/main/src/test/java/org/onap/policy/pap/main/parameters/TestPdpModifyRequestMapParams.java
@@ -53,58 +53,59 @@ public class TestPdpModifyRequestMapParams {
         updTimers = mock(TimerManager.class);
         stateTimers = mock(TimerManager.class);
 
-        params = new PdpModifyRequestMapParams().setModifyLock(lock).setPublisher(pub).setResponseDispatcher(disp);
+        params = new PdpModifyRequestMapParams().setModifyLock(lock).setPublisher(pub).setResponseDispatcher(disp)
+                        .setParams(pdpParams).setStateChangeTimers(stateTimers).setUpdateTimers(updTimers);
     }
 
     @Test
     public void testGettersSetters() {
-        assertSame(params, params.setParams(pdpParams).setStateChangeTimers(stateTimers).setUpdateTimers(updTimers));
-
-        assertSame(pdpParams, params.getParams());
-        assertSame(updTimers, params.getUpdateTimers());
-        assertSame(stateTimers, params.getStateChangeTimers());
-
-        // super class data should also be available
         assertSame(pub, params.getPublisher());
         assertSame(disp, params.getResponseDispatcher());
         assertSame(lock, params.getModifyLock());
+        assertSame(pdpParams, params.getParams());
+        assertSame(updTimers, params.getUpdateTimers());
+        assertSame(stateTimers, params.getStateChangeTimers());
     }
 
     @Test
     public void testValidate() {
         // no exception
-        params.setParams(pdpParams).setStateChangeTimers(stateTimers).setUpdateTimers(updTimers).validate();
+        params.validate();
+    }
+
+    @Test
+    public void testValidate_MissingPublisher() {
+        assertThatIllegalArgumentException().isThrownBy(() -> params.setPublisher(null).validate())
+                        .withMessageContaining("publisher");
+    }
+
+    @Test
+    public void testValidate_MissingDispatcher() {
+        assertThatIllegalArgumentException().isThrownBy(() -> params.setResponseDispatcher(null).validate())
+                        .withMessageContaining("Dispatch");
+    }
+
+    @Test
+    public void testValidate_MissingLock() {
+        assertThatIllegalArgumentException().isThrownBy(() -> params.setModifyLock(null).validate())
+                        .withMessageContaining("Lock");
     }
 
     @Test
     public void testValidate_MissingPdpParams() {
-        assertThatIllegalArgumentException().isThrownBy(
-            () -> params.setStateChangeTimers(stateTimers).setUpdateTimers(updTimers).validate())
+        assertThatIllegalArgumentException().isThrownBy(() -> params.setParams(null).validate())
                         .withMessageContaining("PDP param");
     }
 
     @Test
     public void testValidate_MissingStateChangeTimers() {
-        assertThatIllegalArgumentException().isThrownBy(
-            () -> params.setParams(pdpParams).setUpdateTimers(updTimers).validate())
+        assertThatIllegalArgumentException().isThrownBy(() -> params.setStateChangeTimers(null).validate())
                         .withMessageContaining("state");
     }
 
     @Test
     public void testValidate_MissingUpdateTimers() {
-        assertThatIllegalArgumentException().isThrownBy(
-            () -> params.setParams(pdpParams).setStateChangeTimers(stateTimers).validate())
+        assertThatIllegalArgumentException().isThrownBy(() -> params.setUpdateTimers(null).validate())
                         .withMessageContaining("update");
     }
-
-    @Test
-    public void testValidate_MissingSuperclassData() {
-        // leave out one of the superclass fields
-        assertThatIllegalArgumentException().isThrownBy(
-            () -> new PdpModifyRequestMapParams()
-                        .setPublisher(pub)
-                        .setResponseDispatcher(disp).setParams(pdpParams).setStateChangeTimers(stateTimers)
-                        .setUpdateTimers(updTimers).validate()).withMessageContaining("Lock");
-
-    }
 }
diff --git a/main/src/test/java/org/onap/policy/pap/main/parameters/TestRequestDataParams.java b/main/src/test/java/org/onap/policy/pap/main/parameters/TestRequestParams.java
index 16d247f2..b4855e74 100644
--- a/main/src/test/java/org/onap/policy/pap/main/parameters/TestRequestDataParams.java
+++ b/main/src/test/java/org/onap/policy/pap/main/parameters/TestRequestParams.java
@@ -21,6 +21,7 @@
 package org.onap.policy.pap.main.parameters;
 
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertSame;
 import static org.mockito.Mockito.mock;
 
@@ -29,12 +30,16 @@ import org.junit.Test;
 import org.onap.policy.common.endpoints.listeners.RequestIdDispatcher;
 import org.onap.policy.models.pdp.concepts.PdpStatus;
 import org.onap.policy.pap.main.comm.Publisher;
+import org.onap.policy.pap.main.comm.TimerManager;
 
-public class TestRequestDataParams {
-    private RequestDataParams params;
+public class TestRequestParams {
+    private static final int RETRIES = 1;
+
+    private RequestParams params;
     private Publisher pub;
     private RequestIdDispatcher<PdpStatus> disp;
     private Object lock;
+    private TimerManager timers;
 
     /**
      * Sets up the objects and creates an empty {@link #params}.
@@ -45,8 +50,10 @@ public class TestRequestDataParams {
         pub = mock(Publisher.class);
         disp = mock(RequestIdDispatcher.class);
         lock = new Object();
+        timers = mock(TimerManager.class);
 
-        params = new RequestDataParams();
+        params = new RequestParams().setModifyLock(lock).setPublisher(pub).setResponseDispatcher(disp).setTimers(timers)
+                        .setMaxRetryCount(RETRIES);
     }
 
     @Test
@@ -56,32 +63,37 @@ public class TestRequestDataParams {
         assertSame(pub, params.getPublisher());
         assertSame(disp, params.getResponseDispatcher());
         assertSame(lock, params.getModifyLock());
+        assertSame(timers, params.getTimers());
+        assertEquals(RETRIES, params.getMaxRetryCount());
     }
 
     @Test
     public void testValidate() {
         // no exception
-        params.setModifyLock(lock).setPublisher(pub).setResponseDispatcher(disp).validate();
+        params.validate();
     }
 
     @Test
     public void testValidate_MissingLock() {
-        assertThatIllegalArgumentException().isThrownBy(
-            () -> params.setPublisher(pub).setResponseDispatcher(disp).validate())
-            .withMessageContaining("Lock");
+        assertThatIllegalArgumentException().isThrownBy(() -> params.setModifyLock(null).validate())
+                        .withMessageContaining("Lock");
     }
 
     @Test
     public void testValidate_MissingDispatcher() {
-        assertThatIllegalArgumentException().isThrownBy(
-            () -> params.setModifyLock(lock).setPublisher(pub).validate())
-            .withMessageContaining("Dispatcher");
+        assertThatIllegalArgumentException().isThrownBy(() -> params.setResponseDispatcher(null).validate())
+                        .withMessageContaining("Dispatcher");
     }
 
     @Test
     public void testValidate_MissingPublisher() {
-        assertThatIllegalArgumentException().isThrownBy(
-            () -> params.setModifyLock(lock).setResponseDispatcher(disp).validate())
-            .withMessageContaining("publisher");
+        assertThatIllegalArgumentException().isThrownBy(() -> params.setPublisher(null).validate())
+                        .withMessageContaining("publisher");
+    }
+
+    @Test
+    public void testValidate_MissingTimers() {
+        assertThatIllegalArgumentException().isThrownBy(() -> params.setTimers(null).validate())
+                        .withMessageContaining("timer");
     }
 }