QPID-1764 : Add a BaseTransactionLog that takes care of handling persistent message references so that the underlying TransactionLog need not worry about that.

Updated MemoryMS to use this even to ensure that the code is exercised.
To ensure that the new BaseTransactionLog was correctly used when used by a TransactionLog. The configure() method now returns an Object(TransactionLog) that is the newly configured TL. Existing tests and code where the original TL reference was used have been changed to use the output of the configure() call.
NOTE: the return type should be changed to TransactionLog but until we have completely split the TransactionLog and RoutingTable implementations then this is not possible.

The implementation also includes a number of items from the Flow To Disk review:
- The old get* Methods have been removed from the TransactionLog interface.
- Rollback should now rollback enqueues. (No test provided)
- StoreContext now has enqueue/dequeue methods that track the messageId/Queue pairing
- The linked list per message has been reduced to a link list per message that is enqueued on multiple queues. Messages that exist on only one queue have no additional overhead.

- Optimisation also included to:
Include message delete in 'dequeue transaction' where the message was only ever enqueued on a single queue.
All other message deletes are peformed as part of an asynchrounous commit.

The asynchrounous commit is setup via the StoreContext, which has had some work done to move it towards becomming a Qpid Transaction Object where all operations are performed against rather than going via the TransactionLog.

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

1 files changed, 32 insertions, 20 deletions

diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/server/store/SlowMessageStore.java b/qpid/java/systests/src/main/java/org/apache/qpid/server/store/SlowMessageStore.java
index c7c2c8b292..662f04b3c9 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/server/store/SlowMessageStore.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/server/store/SlowMessageStore.java
@@ -36,6 +36,7 @@ import org.apache.qpid.server.routing.RoutingTable;
 
 import java.util.HashMap;
 import java.util.Iterator;
+import java.util.ArrayList;
 
 public class SlowMessageStore implements TransactionLog, RoutingTable
 {
@@ -50,7 +51,7 @@ public class SlowMessageStore implements TransactionLog, RoutingTable
     private static final String POST = "post";
     private String DEFAULT_DELAY = "default";
 
-    public void configure(VirtualHost virtualHost, String base, VirtualHostConfiguration config) throws Exception
+    public Object configure(VirtualHost virtualHost, String base, VirtualHostConfiguration config) throws Exception
     {
         _logger.warn("Starting SlowMessageStore on Virtualhost:" + virtualHost.getName());
         Configuration delays = config.getStoreConfiguration().subset(DELAYS);
@@ -81,7 +82,11 @@ public class SlowMessageStore implements TransactionLog, RoutingTable
                 _realTransactionLog = (TransactionLog) o;
             }
         }
-        _realTransactionLog.configure(virtualHost, base , config);
+
+        // The call to configure may return a new transaction log 
+        _realTransactionLog = (TransactionLog) _realTransactionLog.configure(virtualHost, base , config);
+        
+        return this;
     }
 
     private void configureDelays(Configuration config)
@@ -205,10 +210,10 @@ public class SlowMessageStore implements TransactionLog, RoutingTable
         doPostDelay("removeQueue");
     }
 
-    public void enqueueMessage(StoreContext context, AMQQueue queue, Long messageId) throws AMQException
+    public void enqueueMessage(StoreContext context, ArrayList<AMQQueue> queues, Long messageId) throws AMQException
     {
         doPreDelay("enqueueMessage");
-        _realTransactionLog.enqueueMessage(context, queue, messageId);
+        _realTransactionLog.enqueueMessage(context, queues, messageId);
         doPostDelay("enqueueMessage");
     }
 
@@ -219,6 +224,13 @@ public class SlowMessageStore implements TransactionLog, RoutingTable
         doPostDelay("dequeueMessage");
     }
 
+    public void removeMessage(StoreContext context, Long messageId) throws AMQException
+    {
+        doPreDelay("dequeueMessage");
+        _realTransactionLog.removeMessage(context, messageId);
+        doPostDelay("dequeueMessage");
+    }
+
     public void beginTran(StoreContext context) throws AMQException
     {
         doPreDelay("beginTran");
@@ -262,22 +274,22 @@ public class SlowMessageStore implements TransactionLog, RoutingTable
         doPostDelay("storeMessageMetaData");
     }
 
-    public MessageMetaData getMessageMetaData(StoreContext context, Long messageId) throws AMQException
-    {
-        doPreDelay("getMessageMetaData");
-        MessageMetaData mmd = _realTransactionLog.getMessageMetaData(context, messageId);
-        doPostDelay("getMessageMetaData");
-        return mmd;
-    }
-
-    public ContentChunk getContentBodyChunk(StoreContext context, Long messageId, int index) throws AMQException
-    {
-        doPreDelay("getContentBodyChunk");
-        ContentChunk c = _realTransactionLog.getContentBodyChunk(context, messageId, index);
-        doPostDelay("getContentBodyChunk");
-        return c;
-    }
-
+//    public MessageMetaData getMessageMetaData(StoreContext context, Long messageId) throws AMQException
+//    {
+//        doPreDelay("getMessageMetaData");
+//        MessageMetaData mmd = _realTransactionLog.getMessageMetaData(context, messageId);
+//        doPostDelay("getMessageMetaData");
+//        return mmd;
+//    }
+//
+//    public ContentChunk getContentBodyChunk(StoreContext context, Long messageId, int index) throws AMQException
+//    {
+//        doPreDelay("getContentBodyChunk");
+//        ContentChunk c = _realTransactionLog.getContentBodyChunk(context, messageId, index);
+//        doPostDelay("getContentBodyChunk");
+//        return c;
+//    }
+//
     public boolean isPersistent()
     {
         return _realTransactionLog.isPersistent();