QPID-1385 : Extracted annonymous class from AMQChannel.resend(boolean), Added new Mock Objects to the broke to allow direct testing of the new class.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@707344 13f79535-47bb-0310-9956-ffa450edef68

8 files changed, 827 insertions, 62 deletions

diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
index 0e2645689c..0fd2b5c83a 100644
--- a/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
@@ -22,9 +22,7 @@ package org.apache.qpid.server;
 
 import org.apache.log4j.Logger;
 import org.apache.qpid.AMQException;
-import org.apache.qpid.configuration.Configured;
 import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.BasicContentHeaderProperties;
 import org.apache.qpid.framing.ContentBody;
 import org.apache.qpid.framing.ContentHeaderBody;
 import org.apache.qpid.framing.FieldTable;
@@ -60,7 +58,6 @@ import java.util.Map;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
 
 public class AMQChannel
 {
@@ -594,60 +591,9 @@ public class AMQChannel
         // Process the Unacked-Map.
         // Marking messages who still have a consumer for to be resent
         // and those that don't to be requeued.
+        _unacknowledgedMessageMap.visit(new ExtractResendAndRequeue(_unacknowledgedMessageMap, msgToRequeue,
+                                                                    msgToResend, requeue, _storeContext));
 
-        _unacknowledgedMessageMap.visit(new UnacknowledgedMessageMap.Visitor()
-        {
-            public boolean callback(final long deliveryTag, QueueEntry message) throws AMQException
-            {
-
-                AMQMessage msg = message.getMessage();
-                msg.setRedelivered(true);
-                final Subscription subscription = message.getDeliveredSubscription();
-                if (subscription != null)
-                {
-                    // Consumer exists
-                    if (!subscription.isClosed())
-                    {
-                        msgToResend.put(deliveryTag, message);
-                    }
-                    else // consumer has gone
-                    {
-                        msgToRequeue.put(deliveryTag, message);
-                    }
-                }
-                else
-                {
-                    // Message has no consumer tag, so was "delivered" to a GET
-                    // or consumer no longer registered
-                    // cannot resend, so re-queue.
-                    if (!message.isQueueDeleted())
-                    {
-                        if (requeue)
-                        {
-                            msgToRequeue.put(deliveryTag, message);
-                        }
-                        else
-                        {
-                            _log.info("No DeadLetter Queue and requeue not requested so dropping message:" + message);
-                        }
-                    }
-                    else
-                    {
-                        message.discard(_storeContext);
-                        _log.warn("Message.queue is null and no DeadLetter Queue so dropping message:" + message);
-                    }
-                }
-
-                // false means continue processing
-                return false;
-            }
-
-            public void visitComplete()
-            {
-            }
-        });
-
-        _unacknowledgedMessageMap.clear();
 
         // Process Messages to Resend
         if (_log.isDebugEnabled())
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/ExtractResendAndRequeue.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/ExtractResendAndRequeue.java
new file mode 100644
index 0000000000..4f5461bc0e
--- /dev/null
+++ b/qpid/java/broker/src/main/java/org/apache/qpid/server/ExtractResendAndRequeue.java
@@ -0,0 +1,110 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ *
+ */
+package org.apache.qpid.server;
+
+import org.apache.qpid.server.ack.UnacknowledgedMessageMap;
+import org.apache.qpid.server.ack.UnacknowledgedMessageMapImpl;
+import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.store.StoreContext;
+import org.apache.qpid.AMQException;
+import org.apache.log4j.Logger;
+
+import java.util.Map;
+
+public class ExtractResendAndRequeue implements UnacknowledgedMessageMap.Visitor
+{
+    private static final Logger _log = Logger.getLogger(ExtractResendAndRequeue.class);
+
+    private Map<Long, QueueEntry> _msgToRequeue;
+    private Map<Long, QueueEntry> _msgToResend;
+    private boolean _requeueIfUnabletoResend;
+    private StoreContext _storeContext;
+    private UnacknowledgedMessageMap _unacknowledgedMessageMap;
+
+    public ExtractResendAndRequeue(UnacknowledgedMessageMap unacknowledgedMessageMap, 
+                                   Map<Long, QueueEntry> msgToRequeue,
+                                   Map<Long, QueueEntry> msgToResend,
+                                   boolean requeueIfUnabletoResend,
+                                   StoreContext storeContext)
+    {
+        _unacknowledgedMessageMap = unacknowledgedMessageMap;
+        _msgToRequeue = msgToRequeue;
+        _msgToResend = msgToResend;
+        _requeueIfUnabletoResend = requeueIfUnabletoResend;
+        _storeContext = storeContext;
+    }
+
+    @Override
+    public boolean callback(final long deliveryTag, QueueEntry message) throws AMQException
+    {
+
+        AMQMessage msg = message.getMessage();
+        msg.setRedelivered(true);
+        final Subscription subscription = message.getDeliveredSubscription();
+        if (subscription != null)
+        {
+            // Consumer exists
+            if (!subscription.isClosed())
+            {
+                _msgToResend.put(deliveryTag, message);
+            }
+            else // consumer has gone
+            {
+                _msgToRequeue.put(deliveryTag, message);
+            }
+        }
+        else
+        {
+            // Message has no consumer tag, so was "delivered" to a GET
+            // or consumer no longer registered
+            // cannot resend, so re-queue.
+            if (!message.isQueueDeleted())
+            {
+                if (_requeueIfUnabletoResend)
+                {
+                    _msgToRequeue.put(deliveryTag, message);
+                }
+                else
+                {
+                    message.discard(_storeContext);
+                    _log.info("No DeadLetter Queue and requeue not requested so dropping message:" + message);
+                }
+            }
+            else
+            {
+                message.discard(_storeContext);
+                _log.warn("Message.queue is null and no DeadLetter Queue so dropping message:" + message);
+            }
+        }
+
+        // false means continue processing
+        return false;
+    }
+
+    @Override
+    public void visitComplete()
+    {
+        _unacknowledgedMessageMap.clear();
+    }
+
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/ExtractResendAndRequeueTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/ExtractResendAndRequeueTest.java
new file mode 100644
index 0000000000..5fbf9484f7
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/ExtractResendAndRequeueTest.java
@@ -0,0 +1,255 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ *
+ */
+package org.apache.qpid.server;
+
+import junit.framework.TestCase;
+import org.apache.qpid.server.ack.UnacknowledgedMessageMapImpl;
+import org.apache.qpid.server.queue.MockQueueEntry;
+import org.apache.qpid.server.queue.QueueEntry;
+import org.apache.qpid.server.queue.SimpleQueueEntryList;
+import org.apache.qpid.server.queue.MockAMQMessage;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.MockAMQQueue;
+import org.apache.qpid.server.queue.AMQMessage;
+import org.apache.qpid.server.queue.QueueEntryIterator;
+import org.apache.qpid.server.store.StoreContext;
+import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.subscription.MockSubscription;
+import org.apache.qpid.AMQException;
+
+import java.util.Map;
+import java.util.LinkedHashMap;
+import java.util.LinkedList;
+import java.util.Iterator;
+
+/**
+ * QPID-1385 : Race condition between added to unacked map and resending due to a rollback.
+ *
+ * In AMQChannel _unackedMap.clear() was done after the visit. This meant that the clear was not in the same
+ * synchronized block as as the preparation to resend.
+ *
+ * This clearing/prep for resend was done as a result of the rollback call. HOWEVER, the delivery thread was still
+ * in the process of sending messages to the client. It is therefore possible that a message could block on the
+ * _unackedMap lock waiting for the visit to compelete so that it can add the new message to the unackedMap....
+ * which is then cleared by the resend/rollback thread.
+ *
+ * This problem was encountered by the testSend2ThenRollback test.
+ *
+ * To try and increase the chance of the race condition occuring this test will send multiple messages so that the
+ * delivery thread will be in progress while the rollback method is called. Hopefully this will cause the
+ * deliveryTag to be lost
+ */
+public class ExtractResendAndRequeueTest extends TestCase
+{
+
+    UnacknowledgedMessageMapImpl _unacknowledgedMessageMap;
+    private static final int INITIAL_MSG_COUNT = 10;
+    private AMQQueue _queue = new MockAMQQueue();
+    private LinkedList<QueueEntry> _referenceList = new LinkedList<QueueEntry>();
+
+    @Override
+    public void setUp() throws AMQException
+    {
+        _unacknowledgedMessageMap = new UnacknowledgedMessageMapImpl(100);
+
+        long id = 0;
+        SimpleQueueEntryList list = new SimpleQueueEntryList(_queue);
+
+        // Add initial messages to QueueEntryList
+        for (int count = 0; count < INITIAL_MSG_COUNT; count++)
+        {
+            AMQMessage msg = new MockAMQMessage(id);
+
+            list.add(msg);
+
+            //Increment ID;
+            id++;
+        }
+
+        // Iterate through the QueueEntryList and add entries to unacknowledgeMessageMap and referecenList
+        QueueEntryIterator queueEntries = list.iterator();
+        while(queueEntries.advance())
+        {
+            QueueEntry entry = queueEntries.getNode();
+            _unacknowledgedMessageMap.add(entry.getMessage().getMessageId(), entry);
+
+            // Store the entry for future inspection
+            _referenceList.add(entry);
+        }
+
+        assertEquals("Map does not contain correct setup data", INITIAL_MSG_COUNT, _unacknowledgedMessageMap.size());
+    }
+
+    /**
+     * Helper method to create a new subscription and aquire the given messages.
+     *
+     * @param messageList The messages to aquire
+     *
+     * @return Subscription that performed the aquire
+     */
+    private Subscription createSubscriptionAndAquireMessages(LinkedList<QueueEntry> messageList)
+    {
+        Subscription subscription = new MockSubscription();
+
+        // Aquire messages in subscription
+        for (QueueEntry entry : messageList)
+        {
+            entry.acquire(subscription);
+        }
+
+        return subscription;
+    }
+
+    /**
+     * This is the normal consumer rollback method.
+     *
+     * An active consumer that has aquired messages expects those messasges to be reset when rollback is requested.
+     *
+     * This test validates that the msgToResend map includes all the messages and none are left behind.
+     *
+     * @throws AMQException the visit interface throws this
+     */
+    public void testResend() throws AMQException
+    {
+        //We don't need the subscription object here.
+        createSubscriptionAndAquireMessages(_referenceList);
+
+        final Map<Long, QueueEntry> msgToRequeue = new LinkedHashMap<Long, QueueEntry>();
+        final Map<Long, QueueEntry> msgToResend = new LinkedHashMap<Long, QueueEntry>();
+
+        // requeueIfUnabletoResend doesn't matter here.
+        _unacknowledgedMessageMap.visit(new ExtractResendAndRequeue(_unacknowledgedMessageMap, msgToRequeue,
+                                                                    msgToResend, true, new StoreContext()));
+
+        assertEquals("Message count for resend not correct.", INITIAL_MSG_COUNT, msgToResend.size());
+        assertEquals("Message count for requeue not correct.", 0, msgToRequeue.size());
+        assertEquals("Map was not emptied", 0, _unacknowledgedMessageMap.size());
+    }
+
+    /**
+     * This is the normal consumer close method.
+     *
+     * When a consumer that has aquired messages expects closes the messages that it has aquired should be removed from
+     * the unacknowledgeMap and placed in msgToRequeue
+     *
+     * This test validates that the msgToRequeue map includes all the messages and none are left behind.
+     *
+     * @throws AMQException the visit interface throws this
+     */
+    public void testRequeueDueToSubscriptionClosure() throws AMQException
+    {
+        Subscription subscription = createSubscriptionAndAquireMessages(_referenceList);
+
+        // Close subscription
+        subscription.close();
+
+        final Map<Long, QueueEntry> msgToRequeue = new LinkedHashMap<Long, QueueEntry>();
+        final Map<Long, QueueEntry> msgToResend = new LinkedHashMap<Long, QueueEntry>();
+
+        // requeueIfUnabletoResend doesn't matter here.
+        _unacknowledgedMessageMap.visit(new ExtractResendAndRequeue(_unacknowledgedMessageMap, msgToRequeue,
+                                                                    msgToResend, true, new StoreContext()));
+
+        assertEquals("Message count for resend not correct.", 0, msgToResend.size());
+        assertEquals("Message count for requeue not correct.", INITIAL_MSG_COUNT, msgToRequeue.size());
+        assertEquals("Map was not emptied", 0, _unacknowledgedMessageMap.size());
+    }
+
+    /**
+     * If the subscription is null, due to message being retrieved via a GET, And we request that messages are requeued
+     * requeueIfUnabletoResend(set to true) then all messages should be sent to the msgToRequeue map.
+     *
+     * @throws AMQException the visit interface throws this
+     */
+
+    public void testRequeueDueToMessageHavingNoConsumerTag() throws AMQException
+    {
+        final Map<Long, QueueEntry> msgToRequeue = new LinkedHashMap<Long, QueueEntry>();
+        final Map<Long, QueueEntry> msgToResend = new LinkedHashMap<Long, QueueEntry>();
+
+        // requeueIfUnabletoResend = true so all messages should go to msgToRequeue
+        _unacknowledgedMessageMap.visit(new ExtractResendAndRequeue(_unacknowledgedMessageMap, msgToRequeue,
+                                                                    msgToResend, true, new StoreContext()));
+
+        assertEquals("Message count for resend not correct.", 0, msgToResend.size());
+        assertEquals("Message count for requeue not correct.", INITIAL_MSG_COUNT, msgToRequeue.size());
+        assertEquals("Map was not emptied", 0, _unacknowledgedMessageMap.size());
+    }
+
+    /**
+     * If the subscription is null, due to message being retrieved via a GET, And we request that we don't
+     * requeueIfUnabletoResend(set to false) then all messages should be dropped as we do not have a dead letter queue.
+     *
+     * @throws AMQException the visit interface throws this
+     */
+
+    public void testDrop() throws AMQException
+    {
+        final Map<Long, QueueEntry> msgToRequeue = new LinkedHashMap<Long, QueueEntry>();
+        final Map<Long, QueueEntry> msgToResend = new LinkedHashMap<Long, QueueEntry>();
+
+        // requeueIfUnabletoResend = false so all messages should be dropped all maps should be empty
+        _unacknowledgedMessageMap.visit(new ExtractResendAndRequeue(_unacknowledgedMessageMap, msgToRequeue,
+                                                                    msgToResend, false, new StoreContext()));
+
+        assertEquals("Message count for resend not correct.", 0, msgToResend.size());
+        assertEquals("Message count for requeue not correct.", 0, msgToRequeue.size());
+        assertEquals("Map was not emptied", 0, _unacknowledgedMessageMap.size());
+
+
+        for (QueueEntry entry : _referenceList)
+        {
+            assertTrue("Message was not discarded", entry.isDeleted());
+        }
+
+    }
+
+    /**
+     * If the subscription is null, due to message being retrieved via a GET, AND the queue upon which the message was
+     * delivered has been deleted then it is not possible to requeue. Currently we simply discar the message but in the
+     * future we may wish to dead letter the message.
+     *
+     * Validate that at the end of the visit all Maps are empty and all messages are marked as deleted
+     *
+     * @throws AMQException the visit interface throws this
+     */
+    public void testDiscard() throws AMQException
+    {
+        final Map<Long, QueueEntry> msgToRequeue = new LinkedHashMap<Long, QueueEntry>();
+        final Map<Long, QueueEntry> msgToResend = new LinkedHashMap<Long, QueueEntry>();
+
+        _queue.delete();
+
+        // requeueIfUnabletoResend : value doesn't matter here as queue has been deleted
+        _unacknowledgedMessageMap.visit(new ExtractResendAndRequeue(_unacknowledgedMessageMap, msgToRequeue,
+                                                                    msgToResend, false, new StoreContext()));
+
+        assertEquals("Message count for resend not correct.", 0, msgToResend.size());
+        assertEquals("Message count for requeue not correct.", 0, msgToRequeue.size());
+        assertEquals("Map was not emptied", 0, _unacknowledgedMessageMap.size());
+
+        for (QueueEntry entry : _referenceList)
+        {
+            assertTrue("Message was not discarded", entry.isDeleted());
+        }
+    }
+
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessage.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessage.java
new file mode 100644
index 0000000000..355ba6a362
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessage.java
@@ -0,0 +1,49 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ *
+ */
+package org.apache.qpid.server.queue;
+
+import org.apache.qpid.server.store.StoreContext;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.abstraction.MessagePublishInfo;
+
+public class MockAMQMessage extends AMQMessage
+{
+    public MockAMQMessage(long messageId)
+            throws AMQException
+    {
+       super(new MockAMQMessageHandle(messageId) ,
+                (StoreContext)null,
+                (MessagePublishInfo)new MockMessagePublishInfo());
+    }
+
+    protected MockAMQMessage(AMQMessage msg)
+            throws AMQException
+    {
+        super(msg);
+    }
+
+
+    @Override
+    public long getSize()
+    {
+        return 0l;
+    }
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessageHandle.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessageHandle.java
new file mode 100644
index 0000000000..bdb0707c27
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQMessageHandle.java
@@ -0,0 +1,37 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ *
+ */
+package org.apache.qpid.server.queue;
+
+import org.apache.qpid.server.store.StoreContext;
+
+public class MockAMQMessageHandle extends InMemoryMessageHandle
+{
+    public MockAMQMessageHandle(final Long messageId)
+    {
+        super(messageId);
+    }
+
+    @Override
+    public long getBodySize(StoreContext store)
+    {
+      return 0l;
+    }
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
new file mode 100644
index 0000000000..cecb430574
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
@@ -0,0 +1,313 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ *
+ */
+package org.apache.qpid.server.queue;
+
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.framing.FieldTable;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.store.StoreContext;
+import org.apache.qpid.server.management.ManagedObject;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.AMQException;
+import org.apache.commons.configuration.Configuration;
+
+import java.util.List;
+import java.util.Set;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.LinkedList;
+
+public class MockAMQQueue implements AMQQueue
+{
+    private boolean _deleted = false;
+
+    public AMQShortString getName()
+    {
+        return null;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public boolean isDurable()
+    {
+        return false;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public boolean isAutoDelete()
+    {
+        return false;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public AMQShortString getOwner()
+    {
+        return null;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public VirtualHost getVirtualHost()
+    {
+        return null;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void bind(Exchange exchange, AMQShortString routingKey, FieldTable arguments) throws AMQException
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void unBind(Exchange exchange, AMQShortString routingKey, FieldTable arguments) throws AMQException
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public List<ExchangeBinding> getExchangeBindings()
+    {
+        return null;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void registerSubscription(Subscription subscription, boolean exclusive) throws AMQException
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void unregisterSubscription(Subscription subscription) throws AMQException
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public int getConsumerCount()
+    {
+        return 0;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public int getActiveConsumerCount()
+    {
+        return 0;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public boolean isUnused()
+    {
+        return false;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public boolean isEmpty()
+    {
+        return false;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public int getMessageCount()
+    {
+        return 0;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public int getUndeliveredMessageCount()
+    {
+        return 0;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public long getQueueDepth()
+    {
+        return 0;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public long getReceivedMessageCount()
+    {
+        return 0;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public long getOldestMessageArrivalTime()
+    {
+        return 0;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public boolean isDeleted()
+    {
+        return _deleted;
+    }
+
+    public int delete() throws AMQException
+    {
+       return 0;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public QueueEntry enqueue(StoreContext storeContext, AMQMessage message) throws AMQException
+    {
+        return null;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void requeue(StoreContext storeContext, QueueEntry entry) throws AMQException
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void dequeue(StoreContext storeContext, QueueEntry entry) throws FailedDequeueException
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public boolean resend(QueueEntry entry, Subscription subscription) throws AMQException
+    {
+        return false;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void addQueueDeleteTask(Task task)
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public List<QueueEntry> getMessagesOnTheQueue()
+    {
+        return null;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public List<QueueEntry> getMessagesOnTheQueue(long fromMessageId, long toMessageId)
+    {
+        return null;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public List<Long> getMessagesOnTheQueue(int num)
+    {
+        return null;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public List<Long> getMessagesOnTheQueue(int num, int offest)
+    {
+        return null;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public QueueEntry getMessageOnTheQueue(long messageId)
+    {
+        return null;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void moveMessagesToAnotherQueue(long fromMessageId, long toMessageId, String queueName, StoreContext storeContext)
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void copyMessagesToAnotherQueue(long fromMessageId, long toMessageId, String queueName, StoreContext storeContext)
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void removeMessagesFromQueue(long fromMessageId, long toMessageId, StoreContext storeContext)
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public long getMaximumMessageSize()
+    {
+        return 0;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void setMaximumMessageSize(long value)
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public long getMaximumMessageCount()
+    {
+        return 0;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void setMaximumMessageCount(long value)
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public long getMaximumQueueDepth()
+    {
+        return 0;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void setMaximumQueueDepth(long value)
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public long getMaximumMessageAge()
+    {
+        return 0;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void setMaximumMessageAge(long maximumMessageAge)
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public long getMinimumAlertRepeatGap()
+    {
+        return 0;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void deleteMessageFromTop(StoreContext storeContext) throws AMQException
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public long clearQueue(StoreContext storeContext) throws AMQException
+    {
+        return 0;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void removeExpiredIfNoSubscribers() throws AMQException
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public Set<NotificationCheck> getNotificationChecks()
+    {
+        return null;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void flushSubscription(Subscription sub) throws AMQException
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void deliverAsync(Subscription sub)
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void deliverAsync()
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void stop()
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void configure(Configuration virtualHostDefaultQueueConfiguration)
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public ManagedObject getManagedObject()
+    {
+        return null;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public int compareTo(AMQQueue o)
+    {
+        return 0;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockMessagePublishInfo.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockMessagePublishInfo.java
new file mode 100644
index 0000000000..5a5ffaa14d
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/MockMessagePublishInfo.java
@@ -0,0 +1,52 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.
+ *
+ */
+package org.apache.qpid.server.queue;
+
+import org.apache.qpid.framing.abstraction.MessagePublishInfo;
+import org.apache.qpid.framing.AMQShortString;
+
+public class MockMessagePublishInfo implements MessagePublishInfo
+{
+    public AMQShortString getExchange()
+    {
+        return null;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public void setExchange(AMQShortString exchange)
+    {
+        //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public boolean isImmediate()
+    {
+        return false;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public boolean isMandatory()
+    {
+        return false;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
+    public AMQShortString getRoutingKey()
+    {
+        return null;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java
index db2f8a57ad..4f0213ce8a 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java
@@ -33,10 +33,10 @@ import org.apache.qpid.server.queue.QueueEntry.SubscriptionAcquiredState;
 public class MockSubscription implements Subscription
 {
 
-    private boolean closed = false;
+    private boolean _closed = false;
     private AMQShortString tag = new AMQShortString("mocktag");
     private AMQQueue queue = null;
-    private StateListener listener = null;
+    private StateListener _listener = null;
     private QueueEntry lastSeen = null;
     private State _state = State.ACTIVE;
     private ArrayList<QueueEntry> messages = new ArrayList<QueueEntry>();
@@ -44,8 +44,11 @@ public class MockSubscription implements Subscription
     @Override
     public void close()
     {
-        closed = true;
-        listener.stateChange(this, _state, State.CLOSED);
+        _closed = true;
+        if (_listener != null)
+        {
+            _listener.stateChange(this, _state, State.CLOSED);
+        }
         _state = State.CLOSED;
     }
 
@@ -117,7 +120,7 @@ public class MockSubscription implements Subscription
     @Override
     public boolean isClosed()
     {
-        return closed;
+        return _closed;
     }
 
     @Override
@@ -174,7 +177,7 @@ public class MockSubscription implements Subscription
     @Override
     public void setStateListener(StateListener listener)
     {
-        this.listener = listener;
+        this._listener = listener;
     }
 
     public State getState()