Merge of trunk up to rev 489403

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

12 files changed, 330 insertions, 35 deletions

diff --git a/java/systests/src/test/java/org/apache/qpid/server/ack/TxAckTest.java b/java/systests/src/test/java/org/apache/qpid/server/ack/TxAckTest.java
index a0765f6924..258bcecc41 100644
--- a/java/systests/src/test/java/org/apache/qpid/server/ack/TxAckTest.java
+++ b/java/systests/src/test/java/org/apache/qpid/server/ack/TxAckTest.java
@@ -95,7 +95,8 @@ public class TxAckTest extends TestCase
         Scenario(int messageCount, List<Long> acked, List<Long> unacked)
         {
             TransactionalContext txnContext = new NonTransactionalContext(new TestableMemoryMessageStore(), null,
-                                                                          new LinkedList<RequiredDeliveryException>());
+                                                                          new LinkedList<RequiredDeliveryException>(),
+                                                                          new HashSet<Long>());
             for(int i = 0; i < messageCount; i++)
             {
                 long deliveryTag = i + 1;
diff --git a/java/systests/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java b/java/systests/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
index 991a098678..5909ac048b 100644
--- a/java/systests/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
+++ b/java/systests/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
@@ -202,7 +202,8 @@ public class AbstractHeadersExchangeTestBase extends TestCase
         private static MessageStore _messageStore = new SkeletonMessageStore();
 
         private static TransactionalContext _txnContext = new NonTransactionalContext(_messageStore, null,
-                                                                         new LinkedList<RequiredDeliveryException>());
+                                                                         new LinkedList<RequiredDeliveryException>(),
+                                                                         new HashSet<Long>());
 
         Message(String id, String... headers) throws AMQException
         {
diff --git a/java/systests/src/test/java/org/apache/qpid/server/exchange/ExchangeMBeanTest.java b/java/systests/src/test/java/org/apache/qpid/server/exchange/ExchangeMBeanTest.java
new file mode 100644
index 0000000000..c8a87a0a0e
--- /dev/null
+++ b/java/systests/src/test/java/org/apache/qpid/server/exchange/ExchangeMBeanTest.java
@@ -0,0 +1,124 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.exchange;
+
+import junit.framework.TestCase;
+import org.apache.qpid.server.management.ManagedObject;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+
+import javax.management.openmbean.CompositeData;
+import javax.management.openmbean.TabularData;
+import java.util.ArrayList;
+
+/**
+ * Unit test class for testing different Exchange MBean operations
+ */
+public class ExchangeMBeanTest  extends TestCase
+{
+    private AMQQueue _queue;
+    private QueueRegistry _queueRegistry;
+
+    /**
+     * Test for direct exchange mbean
+     * @throws Exception
+     */
+    public void testDirectExchangeMBean() throws Exception
+    {
+        DestNameExchange exchange = new DestNameExchange();
+        exchange.initialise("amq.direct", false, 0, true);
+        ManagedObject managedObj = exchange.getManagedObject();
+        ManagedExchange mbean = (ManagedExchange)managedObj;
+
+        mbean.createNewBinding(_queue.getName(), "binding1");
+        mbean.createNewBinding(_queue.getName(), "binding2");
+
+        TabularData data = mbean.bindings();
+        ArrayList<CompositeData> list = new ArrayList<CompositeData>(data.values());
+        assertTrue(list.size() == 2);
+
+        // test general exchange properties
+        assertEquals(mbean.getName(), "amq.direct");
+        assertEquals(mbean.getExchangeType(), "direct");
+        assertTrue(mbean.getTicketNo() == 0);
+        assertTrue(!mbean.isDurable());
+        assertTrue(mbean.isAutoDelete());
+    }
+
+    /**
+     * Test for "topic" exchange mbean
+     * @throws Exception
+     */
+    public void testTopicExchangeMBean() throws Exception
+    {
+        DestWildExchange exchange = new DestWildExchange();
+        exchange.initialise("amq.topic", false, 0, true);
+        ManagedObject managedObj = exchange.getManagedObject();
+        ManagedExchange mbean = (ManagedExchange)managedObj;
+
+        mbean.createNewBinding(_queue.getName(), "binding1");
+        mbean.createNewBinding(_queue.getName(), "binding2");
+
+        TabularData data = mbean.bindings();
+        ArrayList<CompositeData> list = new ArrayList<CompositeData>(data.values());
+        assertTrue(list.size() == 2);
+
+        // test general exchange properties
+        assertEquals(mbean.getName(), "amq.topic");
+        assertEquals(mbean.getExchangeType(), "topic");
+        assertTrue(mbean.getTicketNo() == 0);
+        assertTrue(!mbean.isDurable());
+        assertTrue(mbean.isAutoDelete());
+    }
+
+    /**
+     * Test for "Headers" exchange mbean
+     * @throws Exception
+     */
+    public void testHeadersExchangeMBean() throws Exception
+    {
+        HeadersExchange exchange = new HeadersExchange();
+        exchange.initialise("amq.headers", false, 0, true);
+        ManagedObject managedObj = exchange.getManagedObject();
+        ManagedExchange mbean = (ManagedExchange)managedObj;
+
+        mbean.createNewBinding(_queue.getName(), "key1=binding1,key2=binding2");
+        mbean.createNewBinding(_queue.getName(), "key3=binding3");
+
+        TabularData data = mbean.bindings();
+        ArrayList<CompositeData> list = new ArrayList<CompositeData>(data.values());
+        assertTrue(list.size() == 2);
+
+        // test general exchange properties
+        assertEquals(mbean.getName(), "amq.headers");
+        assertEquals(mbean.getExchangeType(), "headers");
+        assertTrue(mbean.getTicketNo() == 0);
+        assertTrue(!mbean.isDurable());
+        assertTrue(mbean.isAutoDelete());
+    }
+
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _queueRegistry = ApplicationRegistry.getInstance().getQueueRegistry();
+        _queue = new AMQQueue("testQueue", false, "ExchangeMBeanTest", false, _queueRegistry);
+        _queueRegistry.registerQueue(_queue);
+    }
+}
diff --git a/java/systests/src/test/java/org/apache/qpid/server/exchange/ReturnUnroutableMandatoryMessageTest.java b/java/systests/src/test/java/org/apache/qpid/server/exchange/ReturnUnroutableMandatoryMessageTest.java
index ec6a82cc29..b125bc1d4c 100644
--- a/java/systests/src/test/java/org/apache/qpid/server/exchange/ReturnUnroutableMandatoryMessageTest.java
+++ b/java/systests/src/test/java/org/apache/qpid/server/exchange/ReturnUnroutableMandatoryMessageTest.java
@@ -6,6 +6,7 @@ import org.apache.qpid.test.VMBrokerSetup;
 import org.apache.qpid.server.registry.ApplicationRegistry;

 import org.apache.qpid.server.util.TestApplicationRegistry;

 import org.apache.qpid.client.*;

+import org.apache.qpid.client.transport.TransportConnection;

 import org.apache.qpid.url.AMQBindingURL;

 import org.apache.qpid.url.BindingURL;

 import org.apache.qpid.exchange.ExchangeDefaults;

@@ -38,12 +39,14 @@ public class ReturnUnroutableMandatoryMessageTest extends TestCase implements Ex
     protected void setUp() throws Exception

     {

         super.setUp();

+        TransportConnection.createVMBroker(1);

         ApplicationRegistry.initialise(new TestApplicationRegistry(), 1);

     }

 

     protected void tearDown() throws Exception

     {

         super.tearDown();

+        TransportConnection.killAllVMBrokers();

     }

 

     /**

@@ -56,13 +59,14 @@ public class ReturnUnroutableMandatoryMessageTest extends TestCase implements Ex
         _bouncedMessageList.clear();

         Connection con = new AMQConnection("vm://:1", "guest", "guest", "consumer1", "/test");

 

+

         AMQSession consumerSession = (AMQSession) con.createSession(false, Session.CLIENT_ACKNOWLEDGE);

 

-        AMQHeadersExchange queue = new AMQHeadersExchange(new AMQBindingURL(ExchangeDefaults.HEADERS_EXCHANGE_CLASS+"://"+ExchangeDefaults.HEADERS_EXCHANGE_NAME+"/test/queue1?"+ BindingURL.OPTION_ROUTING_KEY+"='F0000=1'"));

+        AMQHeadersExchange queue = new AMQHeadersExchange(new AMQBindingURL(ExchangeDefaults.HEADERS_EXCHANGE_CLASS + "://" + ExchangeDefaults.HEADERS_EXCHANGE_NAME + "/test/queue1?" + BindingURL.OPTION_ROUTING_KEY + "='F0000=1'"));

         FieldTable ft = new PropertyFieldTable();

-        ft.setString("F1000","1");

-        MessageConsumer consumer = consumerSession.createConsumer(queue, AMQSession.DEFAULT_PREFETCH_LOW_MARK, AMQSession.DEFAULT_PREFETCH_HIGH_MARK, false, false, (String)null, ft);

-        

+        ft.setString("F1000", "1");

+        MessageConsumer consumer = consumerSession.createConsumer(queue, AMQSession.DEFAULT_PREFETCH_LOW_MARK, AMQSession.DEFAULT_PREFETCH_HIGH_MARK, false, false, (String) null, ft);

+

         //force synch to ensure the consumer has resulted in a bound queue

         ((AMQSession) consumerSession).declareExchangeSynch(ExchangeDefaults.HEADERS_EXCHANGE_NAME, ExchangeDefaults.HEADERS_EXCHANGE_CLASS);

 

@@ -76,49 +80,45 @@ public class ReturnUnroutableMandatoryMessageTest extends TestCase implements Ex
         con2.start();

 

 

-        MessageProducer nonMandatoryProducer = producerSession.createProducer(queue,false,false);

+        MessageProducer nonMandatoryProducer = producerSession.createProducer(queue, false, false);

         MessageProducer mandatoryProducer = producerSession.createProducer(queue);

 

-

         // First test - should neither be bounced nor routed

         _logger.info("Sending non-routable non-mandatory message");

-        TextMessage msg1 =  producerSession.createTextMessage("msg1");

+        TextMessage msg1 = producerSession.createTextMessage("msg1");

         nonMandatoryProducer.send(msg1);

 

         // Second test - should be bounced

         _logger.info("Sending non-routable mandatory message");

-        TextMessage msg2 =  producerSession.createTextMessage("msg2");

+        TextMessage msg2 = producerSession.createTextMessage("msg2");

         mandatoryProducer.send(msg2);

 

         // Third test - should be routed

         _logger.info("Sending routable message");

-        TextMessage msg3 =  producerSession.createTextMessage("msg3");

-        msg3.setStringProperty("F1000","1");

+        TextMessage msg3 = producerSession.createTextMessage("msg3");

+        msg3.setStringProperty("F1000", "1");

         mandatoryProducer.send(msg3);

 

 

-

         _logger.info("Starting consumer connection");

         con.start();

         TextMessage tm = (TextMessage) consumer.receive(1000L);

 

-        assertTrue("No message routed to receiver",tm != null);

-        assertTrue("Wrong message routed to receiver: "+tm.getText(),"msg3".equals(tm.getText()));

+        assertTrue("No message routed to receiver", tm != null);

+        assertTrue("Wrong message routed to receiver: " + tm.getText(), "msg3".equals(tm.getText()));

 

         try

         {

             Thread.sleep(1000L);

         }

-        catch(InterruptedException e)

+        catch (InterruptedException e)

         {

             ;

         }

 

-        assertTrue("Wrong number of messages bounced (expect 1): "+_bouncedMessageList.size(),_bouncedMessageList.size()==1);

+        assertTrue("Wrong number of messages bounced (expect 1): " + _bouncedMessageList.size(), _bouncedMessageList.size() == 1);

         Message m = _bouncedMessageList.get(0);

-        assertTrue("Wrong message bounced: "+m.toString(),m.toString().contains("msg2"));

-

-

+        assertTrue("Wrong message bounced: " + m.toString(), m.toString().contains("msg2"));

 

 

         con.close();

@@ -129,18 +129,23 @@ public class ReturnUnroutableMandatoryMessageTest extends TestCase implements Ex
 

     public static junit.framework.Test suite()

     {

-        return new VMBrokerSetup(new junit.framework.TestSuite(ReturnUnroutableMandatoryMessageTest.class));

+        return new junit.framework.TestSuite(ReturnUnroutableMandatoryMessageTest.class);

     }

 

     public void onException(JMSException jmsException)

     {

-        _logger.warn("Caught exception on producer: ",jmsException);

+

         Exception linkedException = jmsException.getLinkedException();

-        if(linkedException instanceof AMQNoRouteException)

+        if (linkedException instanceof AMQNoRouteException)

         {

             AMQNoRouteException noRoute = (AMQNoRouteException) linkedException;

             Message bounced = (Message) noRoute.getUndeliveredMessage();

             _bouncedMessageList.add(bounced);

+            _logger.info("Caught expected NoRouteException");

+        }

+        else

+        {

+            _logger.warn("Caught exception on producer: ", jmsException);

         }

     }

 }

diff --git a/java/systests/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java b/java/systests/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java
new file mode 100644
index 0000000000..c2ac099855
--- /dev/null
+++ b/java/systests/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java
@@ -0,0 +1,110 @@
+/*
+ *
+ * Copyright (c) 2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.apache.qpid.server.protocol;
+
+import junit.framework.TestCase;
+import org.apache.mina.common.IoSession;
+import org.apache.qpid.codec.AMQCodecFactory;
+import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.exchange.DefaultExchangeFactory;
+import org.apache.qpid.server.exchange.DefaultExchangeRegistry;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.queue.DefaultQueueRegistry;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.store.SkeletonMessageStore;
+import org.apache.qpid.AMQException;
+
+import javax.management.JMException;
+
+/**
+ * Test class to test MBean operations for AMQMinaProtocolSession.
+ */
+public class AMQProtocolSessionMBeanTest   extends TestCase
+{
+    private IoSession _mockIOSession;
+    private MessageStore _messageStore = new SkeletonMessageStore();
+    private AMQMinaProtocolSession _protocolSession;
+    private AMQChannel _channel;
+    private QueueRegistry _queueRegistry;
+    private ExchangeRegistry _exchangeRegistry;
+    private AMQProtocolSessionMBean _mbean;
+
+    public void testChannels() throws Exception
+    {
+        // check the channel count is correct
+        int channelCount = _mbean.channels().size();
+        assertTrue(channelCount == 1);
+        _protocolSession.addChannel(new AMQChannel(2, _messageStore, null));
+        channelCount = _mbean.channels().size();
+        assertTrue(channelCount == 2);
+
+        // general properties test
+        _mbean.setMaximumNumberOfChannels(1000L);
+        assertTrue(_mbean.getMaximumNumberOfChannels() == 1000L);
+
+        // check APIs
+        AMQChannel channel3 = new AMQChannel(3, _messageStore, null);
+        channel3.setLocalTransactional();
+        _protocolSession.addChannel(channel3);
+        _mbean.rollbackTransactions(2);
+        _mbean.rollbackTransactions(3);
+        _mbean.commitTransactions(2);
+        _mbean.commitTransactions(3);
+
+        // This should throw exception, because the channel does't exist
+        try
+        {
+            _mbean.commitTransactions(4);
+            fail();
+        }
+        catch (JMException ex)
+        {
+            System.out.println("expected exception is thrown :" + ex.getMessage());     
+        }
+
+        // check if closing of session works
+        _protocolSession.addChannel(new AMQChannel(5, _messageStore, null));
+        _mbean.closeConnection();
+        try
+        {
+            channelCount = _mbean.channels().size();
+            assertTrue(channelCount == 0);
+            // session is now closed so adding another channel should throw an exception
+            _protocolSession.addChannel(new AMQChannel(6, _messageStore, null));
+            fail();
+        }
+        catch(AMQException ex)
+        {
+            System.out.println("expected exception is thrown :" + ex.getMessage());
+        }
+    }
+    
+    @Override
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        _channel = new AMQChannel(1, _messageStore, null);
+        _queueRegistry = new DefaultQueueRegistry();
+        _exchangeRegistry = new DefaultExchangeRegistry(new DefaultExchangeFactory());
+        _mockIOSession = new MockIoSession();
+        _protocolSession = new AMQMinaProtocolSession(_mockIOSession, _queueRegistry, _exchangeRegistry, new AMQCodecFactory(true));
+        _protocolSession.addChannel(_channel);
+        _mbean = (AMQProtocolSessionMBean)_protocolSession.getManagedObject();
+    }
+}
diff --git a/java/systests/src/test/java/org/apache/qpid/server/protocol/MockIoSession.java b/java/systests/src/test/java/org/apache/qpid/server/protocol/MockIoSession.java
index 81dea32a76..cf6366b513 100644
--- a/java/systests/src/test/java/org/apache/qpid/server/protocol/MockIoSession.java
+++ b/java/systests/src/test/java/org/apache/qpid/server/protocol/MockIoSession.java
@@ -25,6 +25,7 @@ import org.apache.mina.common.support.DefaultCloseFuture;
 import org.apache.mina.common.support.DefaultWriteFuture;
 
 import java.net.SocketAddress;
+import java.net.InetSocketAddress;
 import java.util.Set;
 
 public class MockIoSession implements IoSession
@@ -151,7 +152,7 @@ public class MockIoSession implements IoSession
 
     public SocketAddress getRemoteAddress()
     {
-        return null;  //To change body of implemented methods use File | Settings | File Templates.
+        return new InetSocketAddress("127.0.0.1", 1234);  //To change body of implemented methods use File | Settings | File Templates.
     }
 
     public SocketAddress getLocalAddress()
diff --git a/java/systests/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java b/java/systests/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
index 4fcc691a2f..562452d729 100644
--- a/java/systests/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
+++ b/java/systests/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
@@ -30,18 +30,20 @@ import org.apache.qpid.server.store.SkeletonMessageStore;
 
 import javax.management.JMException;
 import java.util.LinkedList;
+import java.util.HashSet;
 
 /**
  * Test class to test AMQQueueMBean attribtues and operations
  */
-public class AMQQueueMBeanTest  extends TestCase
+public class AMQQueueMBeanTest extends TestCase
 {
     private AMQQueue _queue;
     private AMQQueueMBean _queueMBean;
     private QueueRegistry _queueRegistry;
     private MessageStore _messageStore = new SkeletonMessageStore();
     private TransactionalContext _transactionalContext = new NonTransactionalContext(_messageStore, null,
-                                                                                     new LinkedList<RequiredDeliveryException>()); 
+                                                                                     new LinkedList<RequiredDeliveryException>(),
+                                                                                     new HashSet<Long>()); 
     private MockProtocolSession _protocolSession;
     private AMQChannel _channel;
 
@@ -68,14 +70,14 @@ public class AMQQueueMBeanTest  extends TestCase
         assertFalse(mgr.hasActiveSubscribers());
         assertTrue(_queueMBean.getActiveConsumerCount() == 0);
 
-         _channel = new AMQChannel(1, _messageStore, null);
+        _channel = new AMQChannel(1, _messageStore, null);
         _protocolSession = new MockProtocolSession(_messageStore);
         _protocolSession.addChannel(_channel);
 
-        _queue.registerProtocolSession(_protocolSession, 1, "test", false);
+        _queue.registerProtocolSession(_protocolSession, 1, "test", false, null);
         assertTrue(_queueMBean.getActiveConsumerCount() == 1);
 
-        SubscriptionSet _subscribers = (SubscriptionSet)mgr;
+        SubscriptionSet _subscribers = (SubscriptionSet) mgr;
         SubscriptionTestHelper s1 = new SubscriptionTestHelper("S1");
         SubscriptionTestHelper s2 = new SubscriptionTestHelper("S2");
         _subscribers.addSubscriber(s1);
@@ -167,7 +169,7 @@ public class AMQQueueMBeanTest  extends TestCase
         super.setUp();
         _queueRegistry = new DefaultQueueRegistry();
         _queue = new AMQQueue("testQueue", false, "AMQueueMBeanTest", false, _queueRegistry);
-        _queueMBean = new AMQQueueMBean(_queue);   
+        _queueMBean = new AMQQueueMBean(_queue);
     }
 
     private void sendMessages(int messageCount) throws AMQException
@@ -175,7 +177,8 @@ public class AMQQueueMBeanTest  extends TestCase
         AMQMessage[] messages = new AMQMessage[messageCount];
         for (int i = 0; i < messages.length; i++)
         {
-            messages[i] = message(false);;
+            messages[i] = message(false);
+            ;
         }
         for (int i = 0; i < messageCount; i++)
         {
diff --git a/java/systests/src/test/java/org/apache/qpid/server/queue/AckTest.java b/java/systests/src/test/java/org/apache/qpid/server/queue/AckTest.java
index 1911d38cd2..d4ea728e95 100644
--- a/java/systests/src/test/java/org/apache/qpid/server/queue/AckTest.java
+++ b/java/systests/src/test/java/org/apache/qpid/server/queue/AckTest.java
@@ -38,6 +38,7 @@ import org.apache.qpid.server.util.TestApplicationRegistry;
 
 import java.util.LinkedList;
 import java.util.Set;
+import java.util.HashSet;
 
 /**
  * Tests that acknowledgements are handled correctly.
@@ -82,7 +83,8 @@ public class AckTest extends TestCase
     private void publishMessages(int count, boolean persistent) throws AMQException
     {
         TransactionalContext txnContext = new NonTransactionalContext(_messageStore, null,
-                                                                      new LinkedList<RequiredDeliveryException>());
+                                                                      new LinkedList<RequiredDeliveryException>(),
+                                                                      new HashSet<Long>());
         MessageHandleFactory factory = new MessageHandleFactory();
         for (int i = 1; i <= count; i++)
         {
diff --git a/java/systests/src/test/java/org/apache/qpid/server/queue/MessageTestHelper.java b/java/systests/src/test/java/org/apache/qpid/server/queue/MessageTestHelper.java
index 8cf84e0dcf..7843d8a182 100644
--- a/java/systests/src/test/java/org/apache/qpid/server/queue/MessageTestHelper.java
+++ b/java/systests/src/test/java/org/apache/qpid/server/queue/MessageTestHelper.java
@@ -34,13 +34,15 @@ import org.apache.qpid.AMQException;
 import junit.framework.TestCase;
 
 import java.util.LinkedList;
+import java.util.HashSet;
 
 class MessageTestHelper extends TestCase
 {
     private final MessageStore _messageStore = new SkeletonMessageStore();
 
     private final TransactionalContext _txnContext = new NonTransactionalContext(_messageStore, null,
-                                                                                 new LinkedList<RequiredDeliveryException>());
+                                                                                 new LinkedList<RequiredDeliveryException>(),
+                                                                                 new HashSet<Long>());
 
     MessageTestHelper() throws Exception
     {
diff --git a/java/systests/src/test/java/org/apache/qpid/server/queue/MockProtocolSession.java b/java/systests/src/test/java/org/apache/qpid/server/queue/MockProtocolSession.java
index bb8fd5bc19..87e5c43932 100644
--- a/java/systests/src/test/java/org/apache/qpid/server/queue/MockProtocolSession.java
+++ b/java/systests/src/test/java/org/apache/qpid/server/queue/MockProtocolSession.java
@@ -22,6 +22,7 @@ package org.apache.qpid.server.queue;
 
 import org.apache.qpid.AMQException;
 import org.apache.qpid.framing.AMQDataBlock;
+import org.apache.qpid.framing.FieldTable;
 import org.apache.qpid.server.AMQChannel;
 import org.apache.qpid.server.protocol.AMQProtocolSession;
 import org.apache.qpid.server.store.MessageStore;
@@ -121,4 +122,13 @@ public class MockProtocolSession implements AMQProtocolSession
     public void setSaslServer(SaslServer saslServer)
     {
     }
+
+    public FieldTable getClientProperties()
+    {
+        return null;
+    }
+
+    public void setClientProperties(FieldTable clientProperties)
+    {
+    }
 }
diff --git a/java/systests/src/test/java/org/apache/qpid/server/queue/SubscriptionTestHelper.java b/java/systests/src/test/java/org/apache/qpid/server/queue/SubscriptionTestHelper.java
index 2773c810d2..fea3c93280 100644
--- a/java/systests/src/test/java/org/apache/qpid/server/queue/SubscriptionTestHelper.java
+++ b/java/systests/src/test/java/org/apache/qpid/server/queue/SubscriptionTestHelper.java
@@ -22,6 +22,7 @@ package org.apache.qpid.server.queue;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Queue;
 
 public class SubscriptionTestHelper implements Subscription
 {
@@ -70,6 +71,41 @@ public class SubscriptionTestHelper implements Subscription
     {
     }
 
+    public boolean hasFilters()
+    {
+        return false;
+    }
+
+    public boolean hasInterest(AMQMessage msg)
+    {
+        return true;
+    }
+
+    public Queue<AMQMessage> getPreDeliveryQueue()
+    {
+        return null;
+    }
+
+    public void enqueueForPreDelivery(AMQMessage msg)
+    {
+        //no-op
+    }
+
+    public boolean isAutoClose()
+    {
+        return false;
+    }
+
+    public void close()
+    {
+        //no-op
+    }
+
+    public boolean isBrowser()
+    {
+        return false;  //To change body of implemented methods use File | Settings | File Templates.
+    }
+
     public int hashCode()
     {
         return key.hashCode();
diff --git a/java/systests/src/test/java/org/apache/qpid/server/store/TestReferenceCounting.java b/java/systests/src/test/java/org/apache/qpid/server/store/TestReferenceCounting.java
index 97c9becf18..a40a9bf12f 100644
--- a/java/systests/src/test/java/org/apache/qpid/server/store/TestReferenceCounting.java
+++ b/java/systests/src/test/java/org/apache/qpid/server/store/TestReferenceCounting.java
@@ -49,7 +49,7 @@ public class TestReferenceCounting extends TestCase
     {
         createPersistentContentHeader();
         AMQMessage message = new AMQMessage(_store.getNewMessageId(), new BasicPublishBody(),
-                                            new NonTransactionalContext(_store, null, null),
+                                            new NonTransactionalContext(_store, null, null, null),
                                             createPersistentContentHeader());
         message.incrementReference();
         // we call routing complete to set up the handle
@@ -71,7 +71,7 @@ public class TestReferenceCounting extends TestCase
     public void testMessageRemains() throws AMQException
     {
         AMQMessage message = new AMQMessage(_store.getNewMessageId(), new BasicPublishBody(),
-                                            new NonTransactionalContext(_store, null, null),
+                                            new NonTransactionalContext(_store, null, null, null),
                                             createPersistentContentHeader());
         message.incrementReference();
         // we call routing complete to set up the handle