Refactor request map

Refactored to facilitate the addition of broadcast requests in the
future.
Added disable-PDP capability.
Updated some comments.
Add junit tests.
Only treat null group as broadcast - empty strings are not broadcasts.
Added more logging.

Change-Id: I6f62b8755d20769c333f06752e9d63ac9ed83b57
Issue-ID: POLICY-1542
Signed-off-by: Jim Hahn <jrh3@att.com>

12 files changed, 1885 insertions, 1270 deletions

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");
     }
 }