I am commiting the patch supplied by Arnaud Simon. This patch contains support for dtx.

Currently there is one test case failing. I will try to fix it, if not Arnuad will provide a patch soon


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

7 files changed, 18 insertions, 821 deletions

diff --git a/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java b/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java
index 3ee8277eba..e69de29bb2 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/ack/TxAckTest.java
@@ -1,232 +0,0 @@
-/*
- *
- * 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.ack;
-
-import junit.framework.TestCase;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.BasicPublishBody;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.ContentHeaderBody;
-import org.apache.qpid.framing.AMQFrameDecodingException;
-import org.apache.qpid.framing.abstraction.MessagePublishInfo;
-import org.apache.qpid.server.RequiredDeliveryException;
-import org.apache.qpid.server.queue.AMQMessage;
-import org.apache.qpid.server.store.TestableMemoryMessageStore;
-import org.apache.qpid.server.store.StoreContext;
-import org.apache.qpid.server.txn.NonTransactionalContext;
-import org.apache.qpid.server.txn.TransactionalContext;
-
-import java.util.*;
-
-public class TxAckTest extends TestCase
-{
-    private Scenario individual;
-    private Scenario multiple;
-    private Scenario combined;
-
-    protected void setUp() throws Exception
-    {
-        super.setUp();
-
-        //ack only 5th msg
-        individual = new Scenario(10, Arrays.asList(5l), Arrays.asList(1l, 2l, 3l, 4l, 6l, 7l, 8l, 9l, 10l));
-        individual.update(5, false);
-
-        //ack all up to and including 5th msg
-        multiple = new Scenario(10, Arrays.asList(1l, 2l, 3l, 4l, 5l), Arrays.asList(6l, 7l, 8l, 9l, 10l));
-        multiple.update(5, true);
-
-        //leave only 8th and 9th unacked
-        combined = new Scenario(10, Arrays.asList(1l, 2l, 3l, 4l, 5l, 6l, 7l, 10l), Arrays.asList(8l, 9l));
-        combined.update(3, false);
-        combined.update(5, true);
-        combined.update(7, true);
-        combined.update(2, true);//should be ignored
-        combined.update(1, false);//should be ignored
-        combined.update(10, false);
-    }
-
-    public void testPrepare() throws AMQException
-    {
-        individual.prepare();
-        multiple.prepare();
-        combined.prepare();
-    }
-
-    public void testUndoPrepare() throws AMQException
-    {
-        individual.undoPrepare();
-        multiple.undoPrepare();
-        combined.undoPrepare();
-    }
-
-    public void testCommit() throws AMQException
-    {
-        individual.commit();
-        multiple.commit();
-        combined.commit();
-    }
-
-    public static junit.framework.Test suite()
-    {
-        return new junit.framework.TestSuite(TxAckTest.class);
-    }
-
-    private class Scenario
-    {
-        private final UnacknowledgedMessageMap _map = new UnacknowledgedMessageMapImpl(5000);
-        private final TxAck _op = new TxAck(_map);
-        private final List<Long> _acked;
-        private final List<Long> _unacked;
-        private StoreContext _storeContext = new StoreContext();
-
-        Scenario(int messageCount, List<Long> acked, List<Long> unacked)
-        {
-            TransactionalContext txnContext = new NonTransactionalContext(new TestableMemoryMessageStore(),
-                                                                          _storeContext, null,
-                                                                          new LinkedList<RequiredDeliveryException>(),
-                                                                          new HashSet<Long>());
-            for (int i = 0; i < messageCount; i++)
-            {
-                long deliveryTag = i + 1;
-
-                MessagePublishInfo info = new MessagePublishInfo()
-                {
-
-                    public AMQShortString getExchange()
-                    {
-                        return null;
-                    }
-
-                    public boolean isImmediate()
-                    {
-                        return false;
-                    }
-
-                    public boolean isMandatory()
-                    {
-                        return false;
-                    }
-
-                    public AMQShortString getRoutingKey()
-                    {
-                        return null;
-                    }
-                };
-
-                TestMessage message = new TestMessage(deliveryTag, i, info, txnContext);
-                _map.add(deliveryTag, new UnacknowledgedMessage(null, message, null, deliveryTag));
-            }
-            _acked = acked;
-            _unacked = unacked;
-        }
-
-        void update(long deliverytag, boolean multiple)
-        {
-            _op.update(deliverytag, multiple);
-        }
-
-        private void assertCount(List<Long> tags, int expected)
-        {
-            for (long tag : tags)
-            {
-                UnacknowledgedMessage u = _map.get(tag);
-                assertTrue("Message not found for tag " + tag, u != null);
-                ((TestMessage) u.message).assertCountEquals(expected);
-            }
-        }
-
-        void prepare() throws AMQException
-        {
-            _op.consolidate();
-            _op.prepare(_storeContext);
-
-            assertCount(_acked, -1);
-            assertCount(_unacked, 0);
-
-        }
-
-        void undoPrepare()
-        {
-            _op.consolidate();
-            _op.undoPrepare();
-
-            assertCount(_acked, 1);
-            assertCount(_unacked, 0);
-        }
-
-        void commit()
-        {
-            _op.consolidate();
-            _op.commit(_storeContext);
-
-            //check acked messages are removed from map
-            Set<Long> keys = new HashSet<Long>(_map.getDeliveryTags());
-            keys.retainAll(_acked);
-            assertTrue("Expected messages with following tags to have been removed from map: " + keys, keys.isEmpty());
-            //check unacked messages are still in map
-            keys = new HashSet<Long>(_unacked);
-            keys.removeAll(_map.getDeliveryTags());
-            assertTrue("Expected messages with following tags to still be in map: " + keys, keys.isEmpty());
-        }
-    }
-
-    private class TestMessage extends AMQMessage
-    {
-        private final long _tag;
-        private int _count;
-
-        TestMessage(long tag, long messageId, MessagePublishInfo publishBody, TransactionalContext txnContext)
-        {
-            super(messageId, publishBody, txnContext);
-            try
-            {
-                setContentHeaderBody(new ContentHeaderBody()
-                {
-                    public int getSize()
-                    {
-                        return 1;
-                    }
-                });
-            }
-            catch (AMQException e)
-            {
-                // won't happen
-            }
-            _tag = tag;
-        }
-
-        public void incrementReference()
-        {
-            _count++;
-        }
-
-        public void decrementReference(StoreContext context)
-        {
-            _count--;
-        }
-
-        void assertCountEquals(int expected)
-        {
-            assertEquals("Wrong count for message with tag " + _tag, expected, _count);
-        }
-    }
-}
diff --git a/java/systests/src/main/java/org/apache/qpid/server/channel/MaxChannelsTest.java b/java/systests/src/main/java/org/apache/qpid/server/channel/MaxChannelsTest.java
index 5ddccb8a7b..e69de29bb2 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/channel/MaxChannelsTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/channel/MaxChannelsTest.java
@@ -1,75 +0,0 @@
-/*
- *
- * 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.virtualhost.VirtualHost;
-import org.apache.qpid.server.registry.IApplicationRegistry;
-import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.exchange.ExchangeRegistry;
-import org.apache.qpid.server.queue.QueueRegistry;
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.store.MessageStore;
-import org.apache.qpid.server.store.SkeletonMessageStore;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.protocol.AMQConstant;
-import org.apache.qpid.framing.AMQShortString;
-
-import javax.management.JMException;
-
-/** Test class to test MBean operations for AMQMinaProtocolSession. */
-public class MaxChannelsTest extends TestCase
-{
-//    private MessageStore _messageStore = new SkeletonMessageStore();
-
-    public void testChannels() throws Exception
-    {
-        IApplicationRegistry appRegistry = ApplicationRegistry.getInstance();
-        AMQMinaProtocolSession _protocolSession = new AMQMinaProtocolSession(new MockIoSession(),
-                                                                             appRegistry.getVirtualHostRegistry(),
-                                                                             new AMQCodecFactory(true),
-                                                                             null);
-        _protocolSession.setVirtualHost(appRegistry.getVirtualHostRegistry().getVirtualHost("test"));
-
-        // check the channel count is correct
-        int channelCount = _protocolSession.getChannels().size();
-        assertEquals("Initial channel count wrong", 0, channelCount);
-
-        long maxChannels = 10L;
-        _protocolSession.setMaximumNumberOfChannels(maxChannels);
-        assertEquals("Number of channels not correctly set.", new Long(maxChannels), _protocolSession.getMaximumNumberOfChannels());
-
-
-        try
-        {
-            for (long currentChannel = 0L; currentChannel < maxChannels; currentChannel++)
-            {
-                _protocolSession.addChannel(new AMQChannel(_protocolSession, (int) currentChannel, null, null));
-            }
-        }
-        catch (AMQException e)
-        {
-            assertEquals("Wrong exception recevied.", e.getErrorCode(), AMQConstant.NOT_ALLOWED);
-        }
-        assertEquals("Maximum number of channels not set.", new Long(maxChannels), new Long(_protocolSession.getChannels().size()));
-    }
-
-}
diff --git a/java/systests/src/main/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java b/java/systests/src/main/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
index ff5517bdd5..7d2420e223 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
@@ -28,9 +28,9 @@ import org.apache.qpid.server.queue.AMQMessage;
 import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.queue.MessageHandleFactory;
 import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.messageStore.MessageStore;
 import org.apache.qpid.server.store.SkeletonMessageStore;
-import org.apache.qpid.server.store.MemoryMessageStore;
+import org.apache.qpid.server.messageStore.MemoryMessageStore;
 import org.apache.qpid.server.store.StoreContext;
 import org.apache.qpid.server.txn.NonTransactionalContext;
 import org.apache.qpid.server.txn.TransactionalContext;
@@ -261,7 +261,7 @@ public class AbstractHeadersExchangeTestBase extends TestCase
      */
     static class Message extends AMQMessage
     {
-        private static MessageStore _messageStore = new SkeletonMessageStore();
+        private static MessageStore _messageStore = new MemoryMessageStore();
 
         private static StoreContext _storeContext = new StoreContext();
 
diff --git a/java/systests/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java b/java/systests/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java
index de214e6e4f..6f14956cc4 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java
@@ -21,25 +21,30 @@ 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.txn.MemoryTransactionManager;
+import org.apache.qpid.server.txn.TransactionManager;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 import org.apache.qpid.server.registry.IApplicationRegistry;
 import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.exchange.ExchangeRegistry;
 import org.apache.qpid.server.queue.QueueRegistry;
 import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.messageStore.MessageStore;
+import org.apache.qpid.server.messageStore.MemoryMessageStore;
 import org.apache.qpid.server.store.SkeletonMessageStore;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.framing.AMQShortString;
 
 import javax.management.JMException;
 
+
 /**
  * Test class to test MBean operations for AMQMinaProtocolSession.
  */
 public class AMQProtocolSessionMBeanTest   extends TestCase
 {
-    private MessageStore _messageStore = new SkeletonMessageStore();
+    private MessageStore _messageStore = new MemoryMessageStore();
+    private TransactionManager _txm = new MemoryTransactionManager();
     private AMQMinaProtocolSession _protocolSession;
     private AMQChannel _channel;
     private AMQProtocolSessionMBean _mbean;
@@ -54,7 +59,7 @@ public class AMQProtocolSessionMBeanTest   extends TestCase
                                                                    new AMQShortString("test"),
                                                                    true,
                                                                    _protocolSession.getVirtualHost());
-        AMQChannel channel = new AMQChannel(_protocolSession,2, _messageStore, null);
+        AMQChannel channel = new AMQChannel(_protocolSession,2,_txm, _messageStore, null);
         channel.setDefaultQueue(queue);
         _protocolSession.addChannel(channel);
         channelCount = _mbean.channels().size();
@@ -65,7 +70,7 @@ public class AMQProtocolSessionMBeanTest   extends TestCase
         assertTrue(_mbean.getMaximumNumberOfChannels() == 1000L);
 
         // check APIs
-        AMQChannel channel3 = new AMQChannel(_protocolSession,3, _messageStore, null);
+        AMQChannel channel3 = new AMQChannel(_protocolSession,3,_txm,  _messageStore, null);
         channel3.setLocalTransactional();
         _protocolSession.addChannel(channel3);
         _mbean.rollbackTransactions(2);
@@ -85,14 +90,14 @@ public class AMQProtocolSessionMBeanTest   extends TestCase
         }
 
         // check if closing of session works
-        _protocolSession.addChannel(new AMQChannel(_protocolSession,5, _messageStore, null));
+        _protocolSession.addChannel(new AMQChannel(_protocolSession,5, _txm, _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(_protocolSession,6, _messageStore, null));
+            _protocolSession.addChannel(new AMQChannel(_protocolSession,6, _txm, _messageStore, null));
             fail();
         }
         catch(AMQException ex)
@@ -112,7 +117,7 @@ public class AMQProtocolSessionMBeanTest   extends TestCase
                                                       new AMQCodecFactory(true),
                                                       null);
         _protocolSession.setVirtualHost(appRegistry.getVirtualHostRegistry().getVirtualHost("test"));
-        _channel = new AMQChannel(_protocolSession,1, _messageStore, null);
+        _channel = new AMQChannel(_protocolSession,1, _txm, _messageStore, null);
         _protocolSession.addChannel(_channel);
         _mbean = (AMQProtocolSessionMBean)_protocolSession.getManagedObject();
     }
diff --git a/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java b/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java
index ae2209c629..e69de29bb2 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/queue/AckTest.java
@@ -1,356 +0,0 @@
-/*
- *
- * 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 junit.framework.TestCase;
-import org.apache.log4j.Logger;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.BasicContentHeaderProperties;
-import org.apache.qpid.framing.BasicPublishBody;
-import org.apache.qpid.framing.ContentHeaderBody;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.abstraction.MessagePublishInfo;
-import org.apache.qpid.server.AMQChannel;
-import org.apache.qpid.server.RequiredDeliveryException;
-import org.apache.qpid.server.ack.UnacknowledgedMessage;
-import org.apache.qpid.server.ack.UnacknowledgedMessageMap;
-import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.store.TestableMemoryMessageStore;
-import org.apache.qpid.server.store.StoreContext;
-import org.apache.qpid.server.txn.NonTransactionalContext;
-import org.apache.qpid.server.txn.TransactionalContext;
-import org.apache.qpid.server.util.TestApplicationRegistry;
-import org.apache.qpid.server.util.NullApplicationRegistry;
-
-import java.util.LinkedList;
-import java.util.Set;
-import java.util.HashSet;
-
-/**
- * Tests that acknowledgements are handled correctly.
- */
-public class AckTest extends TestCase
-{
-    private static final Logger _log = Logger.getLogger(AckTest.class);
-
-    private SubscriptionImpl _subscription;
-
-    private MockProtocolSession _protocolSession;
-
-    private TestableMemoryMessageStore _messageStore;
-
-    private StoreContext _storeContext = new StoreContext();
-
-    private AMQChannel _channel;
-
-    private SubscriptionSet _subscriptionManager;
-
-    private AMQQueue _queue;
-
-    private static final AMQShortString DEFAULT_CONSUMER_TAG = new AMQShortString("conTag");
-
-    public AckTest() throws Exception
-    {
-        ApplicationRegistry.initialise(new NullApplicationRegistry());
-    }
-
-    protected void setUp() throws Exception
-    {
-        super.setUp();
-        _messageStore = new TestableMemoryMessageStore();
-        _protocolSession = new MockProtocolSession(_messageStore);
-        _channel = new AMQChannel(_protocolSession,5, _messageStore, null/*dont need exchange registry*/);
-
-        _protocolSession.addChannel(_channel);
-        _subscriptionManager = new SubscriptionSet();
-        _queue = new AMQQueue(new AMQShortString("myQ"), false, new AMQShortString("guest"), true, ApplicationRegistry.getInstance().getVirtualHostRegistry().getVirtualHost("test"), _subscriptionManager);
-    }
-
-    private void publishMessages(int count) throws AMQException
-    {
-        publishMessages(count, false);
-    }
-
-    private void publishMessages(int count, boolean persistent) throws AMQException
-    {
-        TransactionalContext txnContext = new NonTransactionalContext(_messageStore, _storeContext, null,
-                                                                      new LinkedList<RequiredDeliveryException>(),
-                                                                      new HashSet<Long>());
-        MessageHandleFactory factory = new MessageHandleFactory();
-        for (int i = 1; i <= count; i++)
-        {
-            // AMQP version change: Hardwire the version to 0-8 (major=8, minor=0)
-            // TODO: Establish some way to determine the version for the test.
-            MessagePublishInfo publishBody = new MessagePublishInfo()
-            {
-
-                public AMQShortString getExchange()
-                {
-                    return new AMQShortString("someExchange");
-                }
-
-                public boolean isImmediate()
-                {
-                    return false;
-                }
-
-                public boolean isMandatory()
-                {
-                    return false;
-                }
-
-                public AMQShortString getRoutingKey()
-                {
-                    return new AMQShortString("rk");
-                }
-            };
-            AMQMessage msg = new AMQMessage(_messageStore.getNewMessageId(), publishBody, txnContext);
-            if (persistent)
-            {
-                BasicContentHeaderProperties b = new BasicContentHeaderProperties();
-                //This is DeliveryMode.PERSISTENT
-                b.setDeliveryMode((byte) 2);
-                ContentHeaderBody cb = new ContentHeaderBody();
-                cb.properties = b;
-                msg.setContentHeaderBody(cb);
-            }
-            else
-            {
-                msg.setContentHeaderBody(new ContentHeaderBody());
-            }
-            // we increment the reference here since we are not delivering the messaging to any queues, which is where
-            // the reference is normally incremented. The test is easier to construct if we have direct access to the
-            // subscription
-            msg.incrementReference();
-            msg.routingComplete(_messageStore, _storeContext, factory);
-            // we manually send the message to the subscription
-            _subscription.send(msg, _queue);
-        }
-    }
-
-    /**
-     * Tests that the acknowledgements are correctly associated with a channel and
-     * order is preserved when acks are enabled
-     */
-    public void testAckChannelAssociationTest() throws AMQException
-    {
-        _subscription = new SubscriptionImpl(5, _protocolSession, DEFAULT_CONSUMER_TAG, true);
-        final int msgCount = 10;
-        publishMessages(msgCount, true);
-
-        UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap();
-        assertTrue(map.size() == msgCount);
-        assertTrue(_messageStore.getMessageMetaDataMap().size() == msgCount);
-
-        Set<Long> deliveryTagSet = map.getDeliveryTags();
-        int i = 1;
-        for (long deliveryTag : deliveryTagSet)
-        {
-            assertTrue(deliveryTag == i);
-            i++;
-            UnacknowledgedMessage unackedMsg = map.get(deliveryTag);
-            assertTrue(unackedMsg.queue == _queue);
-        }
-
-        assertTrue(map.size() == msgCount);
-        assertTrue(_messageStore.getMessageMetaDataMap().size() == msgCount);
-    }
-
-    /**
-     * Tests that in no-ack mode no messages are retained
-     */
-    public void testNoAckMode() throws AMQException
-    {
-        // false arg means no acks expected
-        _subscription = new SubscriptionImpl(5, _protocolSession, DEFAULT_CONSUMER_TAG, false);
-        final int msgCount = 10;
-        publishMessages(msgCount);
-
-        UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap();
-        assertTrue(map.size() == 0);
-        assertTrue(_messageStore.getMessageMetaDataMap().size() == 0);
-    }
-
-    /**
-     * Tests that a single acknowledgement is handled correctly (i.e multiple flag not
-     * set case)
-     */
-    public void testSingleAckReceivedTest() throws AMQException
-    {
-        _subscription = new SubscriptionImpl(5, _protocolSession, DEFAULT_CONSUMER_TAG, true);
-        final int msgCount = 10;
-        publishMessages(msgCount);
-
-        _channel.acknowledgeMessage(5, false);
-        UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap();
-        assertTrue(map.size() == msgCount - 1);
-
-        Set<Long> deliveryTagSet = map.getDeliveryTags();
-        int i = 1;
-        for (long deliveryTag : deliveryTagSet)
-        {
-            assertTrue(deliveryTag == i);
-            UnacknowledgedMessage unackedMsg = map.get(deliveryTag);
-            assertTrue(unackedMsg.queue == _queue);
-            // 5 is the delivery tag of the message that *should* be removed
-            if (++i == 5)
-            {
-                ++i;
-            }
-        }
-    }
-
-    /**
-     * Tests that a single acknowledgement is handled correctly (i.e multiple flag not
-     * set case)
-     */
-    public void testMultiAckReceivedTest() throws AMQException
-    {
-        _subscription = new SubscriptionImpl(5, _protocolSession, DEFAULT_CONSUMER_TAG, true);
-        final int msgCount = 10;
-        publishMessages(msgCount);
-
-        _channel.acknowledgeMessage(5, true);
-        UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap();
-        assertTrue(map.size() == 5);
-
-        Set<Long> deliveryTagSet = map.getDeliveryTags();
-        int i = 1;
-        for (long deliveryTag : deliveryTagSet)
-        {
-            assertTrue(deliveryTag == i + 5);
-            UnacknowledgedMessage unackedMsg = map.get(deliveryTag);
-            assertTrue(unackedMsg.queue == _queue);
-            ++i;
-        }
-    }
-
-    /**
-     * Tests that a multiple acknowledgement is handled correctly. When ack'ing all pending msgs.
-     */
-    public void testMultiAckAllReceivedTest() throws AMQException
-    {
-        _subscription = new SubscriptionImpl(5, _protocolSession, DEFAULT_CONSUMER_TAG, true);
-        final int msgCount = 10;
-        publishMessages(msgCount);
-
-        _channel.acknowledgeMessage(0, true);
-        UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap();
-        assertTrue(map.size() == 0);
-
-        Set<Long> deliveryTagSet = map.getDeliveryTags();
-        int i = 1;
-        for (long deliveryTag : deliveryTagSet)
-        {
-            assertTrue(deliveryTag == i + 5);
-            UnacknowledgedMessage unackedMsg = map.get(deliveryTag);
-            assertTrue(unackedMsg.queue == _queue);
-            ++i;
-        }
-    }
-
-    public void testPrefetchHighLow() throws AMQException
-    {
-        int lowMark = 5;
-        int highMark = 10;
-
-        _subscription = new SubscriptionImpl(5, _protocolSession, DEFAULT_CONSUMER_TAG, true);
-        _channel.setPrefetchLowMarkCount(lowMark);
-        _channel.setPrefetchHighMarkCount(highMark);
-
-        assertTrue(_channel.getPrefetchLowMarkCount() == lowMark);
-        assertTrue(_channel.getPrefetchHighMarkCount() == highMark);
-
-        publishMessages(highMark);
-
-        // at this point we should have sent out only highMark messages
-        // which have not bee received so will be queued up in the channel
-        // which should be suspended
-        assertTrue(_subscription.isSuspended());
-        UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap();
-        assertTrue(map.size() == highMark);
-
-        //acknowledge messages so we are just above lowMark
-        _channel.acknowledgeMessage(lowMark - 1, true);
-
-        //we should still be suspended
-        assertTrue(_subscription.isSuspended());
-        assertTrue(map.size() == lowMark + 1);
-
-        //acknowledge one more message
-        _channel.acknowledgeMessage(lowMark, true);
-
-        //and suspension should be lifted
-        assertTrue(!_subscription.isSuspended());
-
-        //pubilsh more msgs so we are just below the limit
-        publishMessages(lowMark - 1);
-
-        //we should not be suspended
-        assertTrue(!_subscription.isSuspended());
-
-        //acknowledge all messages
-        _channel.acknowledgeMessage(0, true);
-        try
-        {
-            Thread.sleep(3000);
-        }
-        catch (InterruptedException e)
-        {
-            _log.error("Error: " + e, e);
-        }
-        //map will be empty
-        assertTrue(map.size() == 0);
-    }
-
-    public void testPrefetch() throws AMQException
-    {
-        _subscription = new SubscriptionImpl(5, _protocolSession, DEFAULT_CONSUMER_TAG, true);
-        _channel.setPrefetchCount(5);
-
-        assertTrue(_channel.getPrefetchCount() == 5);
-
-        final int msgCount = 5;
-        publishMessages(msgCount);
-
-        // at this point we should have sent out only 5 messages with a further 5 queued
-        // up in the channel which should now be suspended
-        assertTrue(_subscription.isSuspended());
-        UnacknowledgedMessageMap map = _channel.getUnacknowledgedMessageMap();
-        assertTrue(map.size() == 5);
-        _channel.acknowledgeMessage(5, true);
-        assertTrue(!_subscription.isSuspended());
-        try
-        {
-            Thread.sleep(3000);
-        }
-        catch (InterruptedException e)
-        {
-            _log.error("Error: " + e, e);
-        }
-        assertTrue(map.size() == 0);
-    }
-
-    public static junit.framework.Test suite()
-    {
-        return new junit.framework.TestSuite(AckTest.class);
-    }
-}
diff --git a/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java b/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java
index 88272023e8..4c3bf2a3ea 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/queue/MessageTestHelper.java
@@ -24,7 +24,8 @@ import org.apache.qpid.framing.BasicPublishBody;
 import org.apache.qpid.framing.ContentHeaderBody;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.abstraction.MessagePublishInfo;
-import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.messageStore.MessageStore;
+import org.apache.qpid.server.messageStore.MemoryMessageStore;
 import org.apache.qpid.server.store.SkeletonMessageStore;
 import org.apache.qpid.server.store.StoreContext;
 import org.apache.qpid.server.registry.ApplicationRegistry;
@@ -42,7 +43,7 @@ import java.util.HashSet;
 
 class MessageTestHelper extends TestCase
 {
-    private final MessageStore _messageStore = new SkeletonMessageStore();
+    private final MessageStore _messageStore = new MemoryMessageStore();
 
     private final StoreContext _storeContext = new StoreContext();
 
diff --git a/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java b/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java
index ab6d9742e4..e69de29bb2 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/store/TestReferenceCounting.java
@@ -1,146 +0,0 @@
-/*
- *
- * 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.store;
-
-import junit.framework.TestCase;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.framing.BasicContentHeaderProperties;
-import org.apache.qpid.framing.BasicPublishBody;
-import org.apache.qpid.framing.ContentHeaderBody;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.abstraction.MessagePublishInfo;
-import org.apache.qpid.server.queue.AMQMessage;
-import org.apache.qpid.server.queue.MessageHandleFactory;
-import org.apache.qpid.server.txn.NonTransactionalContext;
-
-/**
- * Tests that reference counting works correctly with AMQMessage and the message store
- */
-public class TestReferenceCounting extends TestCase
-{
-    private TestableMemoryMessageStore _store;
-
-    private StoreContext _storeContext = new StoreContext();
-
-    protected void setUp() throws Exception
-    {
-        super.setUp();
-        _store = new TestableMemoryMessageStore();
-    }
-
-    /**
-     * Check that when the reference count is decremented the message removes itself from the store
-     */
-    public void testMessageGetsRemoved() throws AMQException
-    {
-        createPersistentContentHeader();
-
-        MessagePublishInfo info = new MessagePublishInfo()
-        {
-
-            public AMQShortString getExchange()
-            {
-                return null;
-            }
-
-            public boolean isImmediate()
-            {
-                return false;
-            }
-
-            public boolean isMandatory()
-            {
-                return false;
-            }
-
-            public AMQShortString getRoutingKey()
-            {
-                return null;
-            }
-        };
-
-        AMQMessage message = new AMQMessage(_store.getNewMessageId(), info,
-                                            new NonTransactionalContext(_store, _storeContext, null, null, null),
-                                            createPersistentContentHeader());
-        message = message.takeReference();
-
-        // we call routing complete to set up the handle
-        message.routingComplete(_store, _storeContext, new MessageHandleFactory());
-        assertTrue(_store.getMessageMetaDataMap().size() == 1);
-        message.decrementReference(_storeContext);
-        assertTrue(_store.getMessageMetaDataMap().size() == 0);
-    }
-
-    private ContentHeaderBody createPersistentContentHeader()
-    {
-        ContentHeaderBody chb = new ContentHeaderBody();
-        BasicContentHeaderProperties bchp = new BasicContentHeaderProperties();
-        bchp.setDeliveryMode((byte)2);
-        chb.properties = bchp;
-        return chb;
-    }
-
-    public void testMessageRemains() throws AMQException
-    {
-
-        MessagePublishInfo info = new MessagePublishInfo()
-        {
-
-            public AMQShortString getExchange()
-            {
-                return null;
-            }
-
-            public boolean isImmediate()
-            {
-                return false;
-            }
-
-            public boolean isMandatory()
-            {
-                return false;
-            }
-
-            public AMQShortString getRoutingKey()
-            {
-                return null;
-            }
-        };
-
-        AMQMessage message = new AMQMessage(_store.getNewMessageId(),
-                                            info,
-                                            new NonTransactionalContext(_store, _storeContext, null, null, null),
-                                            createPersistentContentHeader());
-        
-        message = message.takeReference();
-        // we call routing complete to set up the handle
-        message.routingComplete(_store, _storeContext, new MessageHandleFactory());
-        assertTrue(_store.getMessageMetaDataMap().size() == 1);
-        message = message.takeReference();
-        message.decrementReference(_storeContext);
-        assertTrue(_store.getMessageMetaDataMap().size() == 1);
-    }
-
-    public static junit.framework.Test suite()
-    {
-        return new junit.framework.TestSuite(TestReferenceCounting.class);
-    }
-}