QPID-3917: Refactor VirtualHost/MessageStore implementations to be ready for BDB-HA

Applied patch from Andrew MacBean <andymacbean@gmail.com> and myself.

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

24 files changed, 554 insertions, 579 deletions

diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/AMQBrokerManagerMBeanTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/AMQBrokerManagerMBeanTest.java
index d34d1bbef3..5c500771c2 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/AMQBrokerManagerMBeanTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/AMQBrokerManagerMBeanTest.java
@@ -37,6 +37,7 @@ import org.apache.qpid.server.queue.QueueRegistry;
 import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.registry.IApplicationRegistry;
 import org.apache.qpid.server.store.TestableMemoryMessageStore;
+import org.apache.qpid.server.store.TestableMemoryMessageStoreFactory;
 import org.apache.qpid.server.util.TestApplicationRegistry;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 import org.apache.qpid.server.virtualhost.VirtualHostImpl;
@@ -165,7 +166,7 @@ public class AMQBrokerManagerMBeanTest extends QpidTestCase
 
         XMLConfiguration configXml = new XMLConfiguration();
         configXml.addProperty("virtualhosts.virtualhost(-1).name", "test");
-        configXml.addProperty("virtualhosts.virtualhost(-1).test.store.class", TestableMemoryMessageStore.class.getName());
+        configXml.addProperty("virtualhosts.virtualhost(-1).test.store.factoryclass", TestableMemoryMessageStoreFactory.class.getName());
 
         ServerConfiguration configuration = new ServerConfiguration(configXml);
 
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/configuration/VirtualHostConfigurationTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/configuration/VirtualHostConfigurationTest.java
index 50e7f0588b..c4c93acfb6 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/configuration/VirtualHostConfigurationTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/configuration/VirtualHostConfigurationTest.java
@@ -27,6 +27,7 @@ import org.apache.qpid.server.queue.AMQPriorityQueue;
 import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.store.TestableMemoryMessageStore;
+import org.apache.qpid.server.store.TestableMemoryMessageStoreFactory;
 import org.apache.qpid.server.util.InternalBrokerBaseCase;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 
@@ -161,7 +162,7 @@ public class VirtualHostConfigurationTest extends InternalBrokerBaseCase
         getConfigXml().addProperty("virtualhosts.virtualhost." + getName() + "Extra.queues(-1).queue(-1).name", "r2d2");
         getConfigXml().addProperty("virtualhosts.virtualhost." + getName() + "Extra.queues.queue.r2d2.deadLetterQueues", "true");
         getConfigXml().addProperty("virtualhosts.virtualhost." + getName() + "Extra.queues(-1).queue(-1).name", "c3p0");
-        getConfigXml().addProperty("virtualhosts.virtualhost." + getName() + "Extra.store.class", TestableMemoryMessageStore.class.getName());
+        getConfigXml().addProperty("virtualhosts.virtualhost." + getName() + "Extra.store.factoryclass", TestableMemoryMessageStoreFactory.class.getName());
 
         // Start the broker now.
         super.createBroker();
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
index 488f251b0a..b6ee95a1cb 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/exchange/AbstractHeadersExchangeTestBase.java
@@ -20,8 +20,18 @@
  */
 package org.apache.qpid.server.exchange;
 
-import org.apache.log4j.Logger;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
 
+import org.apache.log4j.Logger;
 import org.apache.qpid.AMQException;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.BasicContentHeaderProperties;
@@ -31,7 +41,6 @@ import org.apache.qpid.framing.FieldTable;
 import org.apache.qpid.framing.FieldTableFactory;
 import org.apache.qpid.framing.abstraction.MessagePublishInfo;
 import org.apache.qpid.server.binding.Binding;
-import org.apache.qpid.server.binding.BindingFactory;
 import org.apache.qpid.server.message.AMQMessage;
 import org.apache.qpid.server.message.AMQMessageHeader;
 import org.apache.qpid.server.message.MessageMetaData;
@@ -44,23 +53,10 @@ import org.apache.qpid.server.queue.MockStoredMessage;
 import org.apache.qpid.server.queue.QueueEntry;
 import org.apache.qpid.server.queue.SimpleAMQQueue;
 import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.store.DurableConfigurationStore;
-import org.apache.qpid.server.store.MemoryMessageStore;
 import org.apache.qpid.server.store.StoredMessage;
 import org.apache.qpid.server.subscription.Subscription;
 import org.apache.qpid.server.util.InternalBrokerBaseCase;
 
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicLong;
-
 public class AbstractHeadersExchangeTestBase extends InternalBrokerBaseCase
 {
     private static final Logger _log = Logger.getLogger(AbstractHeadersExchangeTestBase.class);
@@ -68,24 +64,6 @@ public class AbstractHeadersExchangeTestBase extends InternalBrokerBaseCase
     private final HeadersExchange exchange = new HeadersExchange();
     protected final Set<TestQueue> queues = new HashSet<TestQueue>();
 
-
-
-    /**
-     * Not used in this test, just there to stub out the routing calls
-     */
-    private MemoryMessageStore _store = new MemoryMessageStore();
-
-
-    private BindingFactory bindingFactory = new BindingFactory(new DurableConfigurationStore.Source()
-                                                        {
-
-                                                            public DurableConfigurationStore getMessageStore()
-                                                            {
-                                                                return _store;
-                                                            }
-                                                        },
-                                                        exchange);
-
     private int count;
 
     public void testDoNothing()
@@ -103,7 +81,6 @@ public class AbstractHeadersExchangeTestBase extends InternalBrokerBaseCase
     protected void unbind(TestQueue queue, String... bindings) throws AMQException
     {
         String queueName = queue.getName();
-        //TODO - check this
         exchange.onUnbind(new Binding(null,queueName, queue, exchange, getHeadersMap(bindings)));
     }
     
@@ -538,12 +515,6 @@ public class AbstractHeadersExchangeTestBase extends InternalBrokerBaseCase
             }
 
 
-            public AMQMessage getUnderlyingMessage()
-            {
-                return Message.this;
-            }
-
-
             public ContentHeaderBody getContentHeader()
             {
                 try
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/messages/MessageStoreMessagesTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/messages/MessageStoreMessagesTest.java
index cc032a0430..3377573b9d 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/messages/MessageStoreMessagesTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/messages/MessageStoreMessagesTest.java
@@ -29,12 +29,10 @@ public class MessageStoreMessagesTest extends AbstractTestMessages
 {
     public void testMessageStoreCreated()
     {
-        String name = "DerbyMessageStore";
-
-        _logMessage = MessageStoreMessages.CREATED(name);
+        _logMessage = MessageStoreMessages.CREATED();
         List<Object> log = performLog();
 
-        String[] expected = {"Created :", name};
+        String[] expected = {"Created"};
 
         validateLogMessage(log, "MST-1001", expected);
     }
@@ -70,56 +68,4 @@ public class MessageStoreMessagesTest extends AbstractTestMessages
 
         validateLogMessage(log, "MST-1004", expected);
     }
-/*
-    public void testMessageStoreRecoveryStart_withQueue()
-    {
-        String queueName = "testQueue";
-
-        _logMessage = MessageStoreMessages.RECOVERY_START(queueName, true);
-        List<Object> log = performLog();
-
-        String[] expected = {"Recovery Start :", queueName};
-
-        validateLogMessage(log, "MST-1004", expected);
-    }
-
-    public void testMessageStoreRecovered()
-    {
-        String queueName = "testQueue";
-        Integer messasgeCount = 2000;
-
-        _logMessage = MessageStoreMessages.MST_RECOVERED(messasgeCount, queueName);
-        List<Object> log = performLog();
-
-        // Here we use MessageFormat to ensure the messasgeCount of 2000 is
-        // reformated for display as '2,000'
-        String[] expected = {"Recovered ",
-                             MessageFormat.format("{0,number}", messasgeCount),
-                             "messages for queue", queueName};
-
-        validateLogMessage(log, "MST-1005", expected);
-    }
-
-    public void testMessageStoreRecoveryComplete()
-    {
-        _logMessage = MessageStoreMessages.MST_RECOVERY_COMPLETE(null,false);
-        List<Object> log = performLog();
-
-        String[] expected = {"Recovery Complete"};
-
-        validateLogMessage(log, "MST-1006", expected);
-    }
-
-    public void testMessageStoreRecoveryComplete_withQueue()
-    {
-        String queueName = "testQueue";
-
-        _logMessage = MessageStoreMessages.MST_RECOVERY_COMPLETE(queueName, true);
-        List<Object> log = performLog();
-
-        String[] expected = {"Recovery Complete :", queueName};
-
-        validateLogMessage(log, "MST-1006", expected);
-    }
-    */
 }
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/MessageStoreLogSubjectTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/MessageStoreLogSubjectTest.java
index 158fb667a9..c62b24c3b9 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/MessageStoreLogSubjectTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/logging/subjects/MessageStoreLogSubjectTest.java
@@ -37,13 +37,13 @@ public class MessageStoreLogSubjectTest extends AbstractTestLogSubject
         _testVhost = ApplicationRegistry.getInstance().getVirtualHostRegistry().
                 getVirtualHost("test");
 
-        _subject = new MessageStoreLogSubject(_testVhost, _testVhost.getMessageStore());
+        _subject = new MessageStoreLogSubject(_testVhost, _testVhost.getMessageStore().getClass().getSimpleName());
     }
 
     /**
      * Validate that the logged Subject  message is as expected:
      * MESSAGE [Blank][vh(/test)/ms(MemoryMessageStore)] <Log Message>
-     * @param message the message whos format needs validation
+     * @param message the message who's format needs validation
      */
     @Override
     protected void validateLogStatement(String message)
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java
index 71d5211470..fe9bcc57a6 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/protocol/AMQProtocolSessionMBeanTest.java
@@ -30,7 +30,7 @@ import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.queue.AMQQueueFactory;
 import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.store.MessageStore;
-import org.apache.qpid.server.store.SkeletonMessageStore;
+import org.apache.qpid.server.store.TestableMemoryMessageStore;
 import org.apache.qpid.server.util.InternalBrokerBaseCase;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 
@@ -45,7 +45,7 @@ public class AMQProtocolSessionMBeanTest extends InternalBrokerBaseCase
     /** Used for debugging. */
     private static final Logger log = Logger.getLogger(AMQProtocolSessionMBeanTest.class);
 
-    private MessageStore _messageStore = new SkeletonMessageStore();
+    private MessageStore _messageStore = new TestableMemoryMessageStore();
     private AMQProtocolEngine _protocolSession;
     private AMQChannel _channel;
     private AMQProtocolSessionMBean _mbean;
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueFactoryTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueFactoryTest.java
index 337ff194c3..2e3ff90df9 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueFactoryTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueFactoryTest.java
@@ -35,6 +35,7 @@ import org.apache.qpid.server.logging.actors.CurrentActor;
 import org.apache.qpid.server.logging.actors.TestLogActor;
 import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.store.TestableMemoryMessageStore;
+import org.apache.qpid.server.store.TestableMemoryMessageStoreFactory;
 import org.apache.qpid.server.util.TestApplicationRegistry;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 import org.apache.qpid.test.utils.QpidTestCase;
@@ -53,7 +54,7 @@ public class AMQQueueFactoryTest extends QpidTestCase
 
         XMLConfiguration configXml = new XMLConfiguration();
         configXml.addProperty("virtualhosts.virtualhost(-1).name", getName());
-        configXml.addProperty("virtualhosts.virtualhost(-1)."+getName()+".store.class", TestableMemoryMessageStore.class.getName());
+        configXml.addProperty("virtualhosts.virtualhost(-1)."+getName()+".store.factoryclass", TestableMemoryMessageStoreFactory.class.getName());
 
         ServerConfiguration configuration = new ServerConfiguration(configXml);
 
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
index 45933e7064..d588cdd42c 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AMQQueueMBeanTest.java
@@ -157,7 +157,7 @@ public class AMQQueueMBeanTest extends InternalBrokerBaseCase
     private void verifyBrokerState()
     {
 
-        TestableMemoryMessageStore store = (TestableMemoryMessageStore) getVirtualHost().getMessageStore();
+        TestableMemoryMessageStore store = (TestableMemoryMessageStore) getVirtualHost().getMessageStore().getUnderlyingStore();
 
         // Unlike MessageReturnTest there is no need for a delay as there this thread does the clean up.
 
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AckTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AckTest.java
index 273f0dc018..409b9fd92e 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AckTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/AckTest.java
@@ -20,7 +20,6 @@
  */
 package org.apache.qpid.server.queue;
 
-import org.apache.log4j.Logger;
 
 import org.apache.qpid.AMQException;
 import org.apache.qpid.framing.AMQShortString;
@@ -37,7 +36,7 @@ import org.apache.qpid.server.protocol.AMQProtocolSession;
 import org.apache.qpid.server.protocol.InternalTestProtocolSession;
 import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.store.StoredMessage;
-import org.apache.qpid.server.store.TestMemoryMessageStore;
+import org.apache.qpid.server.store.TestableMemoryMessageStore;
 import org.apache.qpid.server.subscription.Subscription;
 import org.apache.qpid.server.subscription.SubscriptionFactoryImpl;
 import org.apache.qpid.server.txn.AutoCommitTransaction;
@@ -53,13 +52,11 @@ import java.util.Set;
  */
 public class AckTest extends InternalBrokerBaseCase
 {
-    private static final Logger _log = Logger.getLogger(AckTest.class);
-
     private Subscription _subscription;
 
     private AMQProtocolSession _protocolSession;
 
-    private TestMemoryMessageStore _messageStore;
+    private TestableMemoryMessageStore _messageStore;
 
     private AMQChannel _channel;
 
@@ -73,7 +70,7 @@ public class AckTest extends InternalBrokerBaseCase
     {
         super.setUp();
         _virtualHost = ApplicationRegistry.getInstance().getVirtualHostRegistry().getVirtualHost("test");
-        _messageStore = new TestMemoryMessageStore();
+        _messageStore = new TestableMemoryMessageStore();
         _protocolSession = new InternalTestProtocolSession(_virtualHost);
         _channel = new AMQChannel(_protocolSession,5, _messageStore /*dont need exchange registry*/);
 
@@ -180,7 +177,7 @@ public class AckTest extends InternalBrokerBaseCase
         }
         catch (InterruptedException e)
         {
-            e.printStackTrace();  //TODO.
+            Thread.currentThread().interrupt();
         }
 
     }
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java
index a8676bf4c2..7c3098298e 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java
@@ -42,6 +42,7 @@ import org.apache.qpid.server.queue.SimpleAMQQueue.QueueEntryFilter;
 import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.store.StoredMessage;
 import org.apache.qpid.server.store.TestableMemoryMessageStore;
+import org.apache.qpid.server.store.TestableMemoryMessageStoreFactory;
 import org.apache.qpid.server.subscription.MockSubscription;
 import org.apache.qpid.server.subscription.Subscription;
 import org.apache.qpid.server.txn.AutoCommitTransaction;
@@ -105,9 +106,9 @@ public class SimpleAMQQueueTest extends InternalBrokerBaseCase
         ApplicationRegistry applicationRegistry = (ApplicationRegistry)ApplicationRegistry.getInstance();
 
         PropertiesConfiguration env = new PropertiesConfiguration();
-        VirtualHostConfiguration vHostConfig = new VirtualHostConfiguration(getClass().getName(), env);
-        vHostConfig.setMessageStoreClass(TestableMemoryMessageStore.class.getName());
-        _virtualHost = new VirtualHostImpl(ApplicationRegistry.getInstance(), vHostConfig);
+        final VirtualHostConfiguration vhostConfig = new VirtualHostConfiguration(getClass().getName(), env);
+        vhostConfig.setMessageStoreFactoryClass(TestableMemoryMessageStoreFactory.class.getName());
+        _virtualHost = new VirtualHostImpl(ApplicationRegistry.getInstance(), vhostConfig);
         applicationRegistry.getVirtualHostRegistry().registerVirtualHost(_virtualHost);
 
         _queue = (SimpleAMQQueue) AMQQueueFactory.createAMQQueueImpl(_qname, false, _owner, false, false, _virtualHost, _arguments);
@@ -635,7 +636,7 @@ public class SimpleAMQQueueTest extends InternalBrokerBaseCase
 
         qs.add(_queue);
         MessageMetaData metaData = msg.headersReceived(System.currentTimeMillis());
-        TestableMemoryMessageStore store = (TestableMemoryMessageStore) _virtualHost.getMessageStore();
+        TestableMemoryMessageStore store = (TestableMemoryMessageStore) _virtualHost.getMessageStore().getUnderlyingStore();
         StoredMessage handle = store.addMessage(metaData);
         msg.setStoredMessage(handle);
 
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/store/EventManagerTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/EventManagerTest.java
new file mode 100644
index 0000000000..2be79c5839
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/EventManagerTest.java
@@ -0,0 +1,72 @@
+/*
+ * 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 static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyZeroInteractions;
+import static org.apache.qpid.server.store.Event.AFTER_ACTIVATE;
+import static org.apache.qpid.server.store.Event.BEFORE_ACTIVATE;
+import junit.framework.TestCase;
+
+public class EventManagerTest extends TestCase
+{
+    private EventManager _eventManager = new EventManager();
+    private EventListener _mockListener = mock(EventListener.class);
+
+    public void testEventListenerFires()
+    {
+        _eventManager.addEventListener(_mockListener, BEFORE_ACTIVATE);
+        _eventManager.notifyEvent(BEFORE_ACTIVATE);
+        verify(_mockListener).event(BEFORE_ACTIVATE);
+    }
+
+    public void testEventListenerDoesntFire()
+    {
+        _eventManager.addEventListener(_mockListener, BEFORE_ACTIVATE);
+        _eventManager.notifyEvent(AFTER_ACTIVATE);
+        verifyZeroInteractions(_mockListener);
+    }
+
+    public void testEventListenerFiresMulitpleTimes()
+    {
+        _eventManager.addEventListener(_mockListener, BEFORE_ACTIVATE);
+        _eventManager.addEventListener(_mockListener, AFTER_ACTIVATE);
+
+        _eventManager.notifyEvent(BEFORE_ACTIVATE);
+        verify(_mockListener).event(BEFORE_ACTIVATE);
+
+        _eventManager.notifyEvent(AFTER_ACTIVATE);
+        verify(_mockListener).event(AFTER_ACTIVATE);
+    }
+
+    public void testMultipleListenersFireForSameEvent()
+    {
+        final EventListener mockListener1 = mock(EventListener.class);
+        final EventListener mockListener2 = mock(EventListener.class);
+
+        _eventManager.addEventListener(mockListener1, BEFORE_ACTIVATE);
+        _eventManager.addEventListener(mockListener2, BEFORE_ACTIVATE);
+        _eventManager.notifyEvent(BEFORE_ACTIVATE);
+
+        verify(mockListener1).event(BEFORE_ACTIVATE);
+        verify(mockListener2).event(BEFORE_ACTIVATE);
+    }
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTest.java
index 755d61a260..c589bd108b 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/MessageStoreTest.java
@@ -101,7 +101,7 @@ public class MessageStoreTest extends InternalBrokerBaseCase
         String storePath = System.getProperty("QPID_WORK") + "/" + getName();
         
         _config = new PropertiesConfiguration();
-        _config.addProperty("store.class", getTestProfileMessageStoreClassName());
+        _config.addProperty("store.factoryclass", getTestProfileMessageStoreFactoryClassName());
         _config.addProperty("store.environment-path", storePath);
 
         cleanup(new File(storePath));
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java
index 2ffa157ca8..4aa023a25c 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/ReferenceCountingTest.java
@@ -34,12 +34,12 @@ import org.apache.qpid.test.utils.QpidTestCase;
  */
 public class ReferenceCountingTest extends QpidTestCase
 {
-    private TestMemoryMessageStore _store;
+    private TestableMemoryMessageStore _store;
 
 
     protected void setUp() throws Exception
     {
-        _store = new TestMemoryMessageStore();
+        _store = new TestableMemoryMessageStore();
     }
 
     /**
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/store/SkeletonMessageStore.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/SkeletonMessageStore.java
deleted file mode 100644
index 38d3fb78fc..0000000000
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/store/SkeletonMessageStore.java
+++ /dev/null
@@ -1,179 +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 org.apache.commons.configuration.Configuration;
-
-import org.apache.qpid.AMQStoreException;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.server.exchange.Exchange;
-import org.apache.qpid.server.federation.Bridge;
-import org.apache.qpid.server.federation.BrokerLink;
-import org.apache.qpid.server.logging.LogSubject;
-import org.apache.qpid.server.message.EnqueableMessage;
-import org.apache.qpid.server.queue.AMQQueue;
-
-/**
- * A message store that does nothing. Designed to be used in tests that do not want to use any message store
- * functionality.
- */
-public class SkeletonMessageStore implements MessageStore
-{
-    public void configureConfigStore(String name,
-                          ConfigurationRecoveryHandler recoveryHandler,
-                          Configuration config,
-                          LogSubject logSubject) throws Exception
-    {
-    }
-
-    public void configureMessageStore(String name,
-                                      MessageStoreRecoveryHandler recoveryHandler,
-                                      TransactionLogRecoveryHandler tlogRecoveryHandler,
-                                      Configuration config, LogSubject logSubject) throws Exception
-    {
-    }
-
-    public void close() throws Exception
-    {
-    }
-
-    public <M extends StorableMessageMetaData> StoredMessage<M> addMessage(M metaData)
-    {
-        return null;
-    }
-
-
-    public void createExchange(Exchange exchange) throws AMQStoreException
-    {
-
-    }
-
-    public void removeExchange(Exchange exchange) throws AMQStoreException
-    {
-
-    }
-
-    public void bindQueue(Exchange exchange, AMQShortString routingKey, AMQQueue queue, FieldTable args) throws AMQStoreException
-    {
-
-    }
-
-    public void unbindQueue(Exchange exchange, AMQShortString routingKey, AMQQueue queue, FieldTable args) throws AMQStoreException
-    {
-
-    }
-
-    public void createQueue(AMQQueue queue) throws AMQStoreException
-    {
-    }
-
-    public void createQueue(AMQQueue queue, FieldTable arguments) throws AMQStoreException
-    {
-    }
-
-    public boolean isPersistent()
-    {
-        return false;
-    }
-
-    public void removeQueue(final AMQQueue queue) throws AMQStoreException
-    {
-
-    }
-
-    public Transaction newTransaction()
-    {
-        return new Transaction()
-        {
-
-            public void enqueueMessage(TransactionLogResource queue, EnqueableMessage message) throws AMQStoreException
-            {
-
-            }
-
-            public void dequeueMessage(TransactionLogResource  queue, EnqueableMessage message) throws AMQStoreException
-            {
-
-            }
-
-            public void commitTran() throws AMQStoreException
-            {
-
-            }
-
-            public StoreFuture commitTranAsync() throws AMQStoreException
-            {
-                return new StoreFuture()
-                            {
-                                public boolean isComplete()
-                                {
-                                    return true;
-                                }
-
-                                public void waitForCompletion()
-                                {
-
-                                }
-                            };
-            }
-
-            public void abortTran() throws AMQStoreException
-            {
-
-            }
-
-            public void removeXid(long format, byte[] globalId, byte[] branchId)
-            {
-            }
-
-            public void recordXid(long format, byte[] globalId, byte[] branchId, Record[] enqueues, Record[] dequeues)
-            {
-            }
-        };
-    }
-
-    public void updateQueue(AMQQueue queue) throws AMQStoreException
-    {
-
-    }
-
-    @Override
-    public void createBrokerLink(BrokerLink link) throws AMQStoreException
-    {
-    }
-
-    @Override
-    public void deleteBrokerLink(BrokerLink link) throws AMQStoreException
-    {
-    }
-
-    @Override
-    public void createBridge(Bridge bridge) throws AMQStoreException
-    {
-    }
-
-    @Override
-    public void deleteBridge(Bridge bridge) throws AMQStoreException
-    {
-    }
-
-}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/store/StateManagerTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/StateManagerTest.java
new file mode 100644
index 0000000000..b09dcbbdf3
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/StateManagerTest.java
@@ -0,0 +1,89 @@
+/*
+ *
+ * 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;
+
+public class StateManagerTest extends TestCase
+{
+
+    private StateManager _manager = new StateManager();
+
+    public void testInitialState()
+    {
+        assertEquals(State.INITIAL, _manager.getState());
+    }
+
+    public void testStateTransitionAllowed()
+    {
+        assertEquals(State.INITIAL, _manager.getState());
+
+        _manager.stateTransition(State.INITIAL, State.ACTIVE);
+        assertEquals(State.ACTIVE, _manager.getState());
+    }
+
+    public void testStateTransitionDisallowed()
+    {
+        assertEquals(State.INITIAL, _manager.getState());
+
+        try
+        {
+            _manager.stateTransition(State.ACTIVE, State.CLOSING);
+            fail("Exception not thrown");
+        }
+        catch (IllegalStateException e)
+        {
+            // PASS
+        }
+        assertEquals(State.INITIAL, _manager.getState());
+    }
+
+    public void testIsInState()
+    {
+        assertEquals(State.INITIAL, _manager.getState());
+        assertFalse(_manager.isInState(State.ACTIVE));
+        assertTrue(_manager.isInState(State.INITIAL));
+    }
+
+    public void testIsNotInState()
+    {
+        assertEquals(State.INITIAL, _manager.getState());
+        assertTrue(_manager.isNotInState(State.ACTIVE));
+        assertFalse(_manager.isNotInState(State.INITIAL));
+    }
+
+    public void testCheckInState()
+    {
+        assertEquals(State.INITIAL, _manager.getState());
+
+        try
+        {
+            _manager.checkInState(State.ACTIVE);
+            fail("Exception not thrown");
+        }
+        catch (IllegalStateException e)
+        {
+            // PASS
+        }
+        assertEquals(State.INITIAL, _manager.getState());
+    }
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/store/TestMemoryMessageStore.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/TestMemoryMessageStore.java
deleted file mode 100644
index 8a261b3b86..0000000000
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/store/TestMemoryMessageStore.java
+++ /dev/null
@@ -1,98 +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 java.nio.ByteBuffer;
-import java.util.concurrent.atomic.AtomicInteger;
-
-/**
- * Adds some extra methods to the memory message store for testing purposes.
- */
-public class TestMemoryMessageStore extends MemoryMessageStore
-{
-    private AtomicInteger _messageCount = new AtomicInteger(0);
-
-
-    public TestMemoryMessageStore()
-    {
-    }
-
-    @Override
-    public StoredMessage addMessage(StorableMessageMetaData metaData)
-    {
-        return new TestableStoredMessage(super.addMessage(metaData));
-    }
-
-    public int getMessageCount()
-    {
-        return _messageCount.get();
-    }
-
-    private class TestableStoredMessage implements StoredMessage
-    {
-        private final StoredMessage _storedMessage;
-
-        public TestableStoredMessage(StoredMessage storedMessage)
-        {
-            _messageCount.incrementAndGet();
-            _storedMessage = storedMessage;
-        }
-
-        public StorableMessageMetaData getMetaData()
-        {
-            return _storedMessage.getMetaData();
-        }
-
-        public long getMessageNumber()
-        {
-            return _storedMessage.getMessageNumber();
-        }
-
-        public void addContent(int offsetInMessage, ByteBuffer src)
-        {
-            _storedMessage.addContent(offsetInMessage, src);
-        }
-
-        public int getContent(int offsetInMessage, ByteBuffer dst)
-        {
-            return _storedMessage.getContent(offsetInMessage, dst);
-        }
-
-
-        public ByteBuffer getContent(int offsetInMessage, int size)
-        {
-            return _storedMessage.getContent(offsetInMessage, size);
-        }
-
-        public StoreFuture flushToStore()
-        {
-            return _storedMessage.flushToStore();
-        }
-
-        public void remove()
-        {
-            _storedMessage.remove();
-            _messageCount.decrementAndGet();
-        }
-
-    }
-    
-}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/store/TestableMemoryMessageStore.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/TestableMemoryMessageStore.java
index 104e06d29a..210408f490 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/store/TestableMemoryMessageStore.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/TestableMemoryMessageStore.java
@@ -26,6 +26,7 @@ import org.apache.qpid.server.queue.AMQQueue;
 
 import java.nio.ByteBuffer;
 import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicInteger;
 
 /**
@@ -33,26 +34,8 @@ import java.util.concurrent.atomic.AtomicInteger;
  */
 public class TestableMemoryMessageStore extends MemoryMessageStore
 {
-
-    private MemoryMessageStore _mms = null;
-    private HashMap<Long, AMQQueue> _messages = new HashMap<Long, AMQQueue>();
-    private AtomicInteger _messageCount = new AtomicInteger(0);
-
-    public TestableMemoryMessageStore(MemoryMessageStore mms)
-    {
-        _mms = mms;
-    }
-
-    public TestableMemoryMessageStore()
-    {
-
-    }
-
-    @Override
-    public void close() throws Exception
-    {
-        // Not required to do anything
-    }
+    private final Map<Long, AMQQueue> _messages = new HashMap<Long, AMQQueue>();
+    private final AtomicInteger _messageCount = new AtomicInteger(0);
 
     @Override
     public StoredMessage addMessage(StorableMessageMetaData metaData)
@@ -65,36 +48,34 @@ public class TestableMemoryMessageStore extends MemoryMessageStore
         return _messageCount.get();
     }
 
+    public Map<Long, AMQQueue> getMessages()
+    {
+        return _messages;
+    }
+
     private class TestableTransaction implements Transaction
     {
+        @Override
         public void enqueueMessage(TransactionLogResource queue, EnqueableMessage message) throws AMQStoreException
         {
             getMessages().put(message.getMessageNumber(), (AMQQueue)queue);
         }
 
+        @Override
         public void dequeueMessage(TransactionLogResource queue, EnqueableMessage message) throws AMQStoreException
         {
             getMessages().remove(message.getMessageNumber());
         }
 
+        @Override
         public void commitTran() throws AMQStoreException
         {
         }
 
+        @Override
         public StoreFuture commitTranAsync() throws AMQStoreException
         {
-            return new StoreFuture()
-                    {
-                        public boolean isComplete()
-                        {
-                            return true;
-                        }
-
-                        public void waitForCompletion()
-                        {
-
-                        }
-                    };
+            return StoreFuture.IMMEDIATE_FUTURE;
         }
 
         public void abortTran() throws AMQStoreException
@@ -117,10 +98,6 @@ public class TestableMemoryMessageStore extends MemoryMessageStore
         return new TestableTransaction();
     }
 
-    public HashMap<Long, AMQQueue> getMessages()
-    {
-        return _messages;
-    }
 
     private class TestableStoredMessage implements StoredMessage
     {
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/store/TestableMemoryMessageStoreFactory.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/TestableMemoryMessageStoreFactory.java
new file mode 100644
index 0000000000..a737836ed5
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/TestableMemoryMessageStoreFactory.java
@@ -0,0 +1,41 @@
+/*
+ * 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 org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.store.decorators.EventDecorator;
+import org.apache.qpid.server.store.decorators.OperationalLoggingDecorator;
+
+public class TestableMemoryMessageStoreFactory implements MessageStoreFactory
+{
+
+    @Override
+    public MessageStore createMessageStore(LogSubject logSubject)
+    {
+        return new OperationalLoggingDecorator(new EventDecorator(new TestableMemoryMessageStore()), logSubject);
+    }
+
+    @Override
+    public String getStoreClassName()
+    {
+        return TestableMemoryMessageStore.class.getSimpleName();
+    }
+
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/store/decorators/EventDecoratorTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/decorators/EventDecoratorTest.java
new file mode 100644
index 0000000000..7038b8710b
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/decorators/EventDecoratorTest.java
@@ -0,0 +1,79 @@
+/*
+ * 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.decorators;
+
+import static org.mockito.Mockito.*;
+
+import org.apache.qpid.server.store.Event;
+import org.apache.qpid.server.store.EventListener;
+import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.store.decorators.EventDecorator;
+import org.mockito.InOrder;
+
+import junit.framework.TestCase;
+
+public class EventDecoratorTest extends TestCase
+{
+    private MessageStore _mockStore = mock(MessageStore.class);
+    private EventListener _mockListener = mock(EventListener.class);
+
+    private EventDecorator _eventDecorator = new EventDecorator(_mockStore);
+    private InOrder _orderMock = inOrder(_mockListener, _mockStore);
+
+    public void testBeforeActivateDecoration() throws Exception
+    {
+        _eventDecorator.addEventListener(_mockListener, Event.BEFORE_ACTIVATE);
+        _eventDecorator.activate();
+
+        _orderMock.verify(_mockListener).event(Event.BEFORE_ACTIVATE);
+        _orderMock.verify(_mockStore).activate();
+    }
+
+    public void testAfterActivateDecoration() throws Exception
+    {
+        _eventDecorator.addEventListener(_mockListener, Event.AFTER_ACTIVATE);
+        _eventDecorator.activate();
+
+        _orderMock.verify(_mockStore).activate();
+        _orderMock.verify(_mockListener).event(Event.AFTER_ACTIVATE);
+    }
+
+    public void testBeforeAfterActivateDecoration() throws Exception
+    {
+        _eventDecorator.addEventListener(_mockListener, Event.BEFORE_ACTIVATE);
+        _eventDecorator.addEventListener(_mockListener, Event.AFTER_ACTIVATE);
+        _eventDecorator.activate();
+
+        _orderMock.verify(_mockListener).event(Event.BEFORE_ACTIVATE);
+        _orderMock.verify(_mockStore).activate();
+        _orderMock.verify(_mockListener).event(Event.AFTER_ACTIVATE);
+    }
+
+    public void testBeforeAfterCloseDecoration() throws Exception
+    {
+        _eventDecorator.addEventListener(_mockListener, Event.BEFORE_CLOSE);
+        _eventDecorator.addEventListener(_mockListener, Event.AFTER_CLOSE);
+        _eventDecorator.close();
+
+        _orderMock.verify(_mockListener).event(Event.BEFORE_CLOSE);
+        _orderMock.verify(_mockStore).close();
+        _orderMock.verify(_mockListener).event(Event.AFTER_CLOSE);
+    }
+}
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/store/decorators/OperationalLoggingDecoratorTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/decorators/OperationalLoggingDecoratorTest.java
new file mode 100644
index 0000000000..cf06d3ab72
--- /dev/null
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/store/decorators/OperationalLoggingDecoratorTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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.decorators;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.argThat;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import static org.apache.qpid.server.store.MessageStoreConstants.ENVIRONMENT_PATH_PROPERTY;
+
+import junit.framework.TestCase;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.qpid.server.logging.LogActor;
+import org.apache.qpid.server.logging.LogMessage;
+import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.store.ConfigurationRecoveryHandler;
+import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.server.store.MessageStoreRecoveryHandler;
+import org.apache.qpid.server.store.TransactionLogRecoveryHandler;
+import org.apache.qpid.server.store.decorators.OperationalLoggingDecorator;
+import org.hamcrest.Description;
+import org.mockito.ArgumentMatcher;
+import org.mockito.InOrder;
+
+public class OperationalLoggingDecoratorTest extends TestCase
+{
+    private MessageStore _messageStore = mock(MessageStore.class);
+    private LogActor _mockActor = mock(LogActor.class);
+    private LogSubject _mockLogSubject = mock(LogSubject.class);
+    private OperationalLoggingDecorator _operationalLoggingDecorator = new OperationalLoggingDecorator(_messageStore, _mockLogSubject);
+    private InOrder _inOrder = inOrder(_mockActor, _messageStore);
+
+    protected void setUp() throws Exception
+    {
+        super.setUp();
+        CurrentActor.set(_mockActor);
+    }
+
+    public void testConfigureMessageStore() throws Exception
+    {
+        _operationalLoggingDecorator.configureMessageStore(null,null,null,null);
+
+        _inOrder.verify(_mockActor).message(eq(_mockLogSubject), matchesLogMessage("MST-1001 : Created"));
+        _inOrder.verify(_mockActor).message(eq(_mockLogSubject), matchesLogMessage("TXN-1001 : Created"));
+        _inOrder.verify(_messageStore).configureMessageStore(anyString(), any(MessageStoreRecoveryHandler.class), any(TransactionLogRecoveryHandler.class), any(Configuration.class));
+    }
+
+    public void testConfigureMessageStoreWithStoreLocation() throws Exception
+    {
+        final String storeLocation = "/my/store/location";
+        Configuration mockConfig = mock(Configuration.class);
+        when(mockConfig.getString(ENVIRONMENT_PATH_PROPERTY)).thenReturn(storeLocation);
+
+        _operationalLoggingDecorator.configureMessageStore(null,null,null, mockConfig);
+
+        _inOrder.verify(_mockActor).message(eq(_mockLogSubject), matchesLogMessage("MST-1001 : Created"));
+        _inOrder.verify(_mockActor).message(eq(_mockLogSubject), matchesLogMessage("TXN-1001 : Created"));
+        _inOrder.verify(_mockActor).message(eq(_mockLogSubject), matchesLogMessage("MST-1002 : Store location : " + storeLocation));
+        _inOrder.verify(_messageStore).configureMessageStore(anyString(), any(MessageStoreRecoveryHandler.class), any(TransactionLogRecoveryHandler.class), any(Configuration.class));
+    }
+
+    public void testConfigureConfigStore() throws Exception
+    {
+        _operationalLoggingDecorator.configureConfigStore(null,null,null);
+
+        _inOrder.verify(_mockActor).message(eq(_mockLogSubject), matchesLogMessage("CFG-1001 : Created"));
+        _inOrder.verify(_messageStore).configureConfigStore(anyString(), any(ConfigurationRecoveryHandler.class), any(Configuration.class));
+    }
+
+    public void testActivate() throws Exception
+    {
+        _operationalLoggingDecorator.activate();
+
+        _inOrder.verify(_mockActor).message(eq(_mockLogSubject), matchesLogMessage("MST-1004 : Recovery Start"));
+        _inOrder.verify(_messageStore).activate();
+        _inOrder.verify(_mockActor).message(eq(_mockLogSubject), matchesLogMessage("MST-1006 : Recovery Complete"));
+    }
+
+    public void testClose() throws Exception
+    {
+        _operationalLoggingDecorator.close();
+
+        _inOrder.verify(_mockActor).message(eq(_mockLogSubject), matchesLogMessage("MST-1003 : Closed"));
+        _inOrder.verify(_messageStore).close();
+    }
+
+    @Override
+    protected void tearDown() throws Exception
+    {
+        super.tearDown();
+        CurrentActor.remove();
+    }
+
+    private LogMessage matchesLogMessage(String expectedLogMessage)
+    {
+        return argThat(new LogMessageArgumentMatcher(expectedLogMessage));
+    }
+
+    private final class LogMessageArgumentMatcher extends ArgumentMatcher<LogMessage>
+    {
+        private final String _expectedText;
+        private String _description = null;
+;
+        public LogMessageArgumentMatcher(String _expectedLogMessage)
+        {
+            this._expectedText = _expectedLogMessage;
+        }
+
+        @Override
+        public boolean matches(Object item)
+        {
+            LogMessage logMessage = (LogMessage) item;
+            final String actualText = logMessage.toString();
+            if (actualText.equals(_expectedText))
+            {
+                return true;
+            }
+            else
+            {
+                _description  = "Expected <" + _expectedText + "> but got <" + actualText + ">";
+                return false;
+            }
+        }
+
+        @Override
+        public void describeTo(Description description)
+        {
+            if (description != null)
+            {
+                description.appendText(" : "+ _description);
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/txn/MockStoreTransaction.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/txn/MockStoreTransaction.java
index e9b7ceacc5..af49238998 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/txn/MockStoreTransaction.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/txn/MockStoreTransaction.java
@@ -20,26 +20,13 @@
  */
 package org.apache.qpid.server.txn;
 
-import org.apache.commons.configuration.Configuration;
 import org.apache.commons.lang.NotImplementedException;
-
 import org.apache.qpid.AMQStoreException;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.framing.FieldTable;
-import org.apache.qpid.server.exchange.Exchange;
-import org.apache.qpid.server.federation.Bridge;
-import org.apache.qpid.server.federation.BrokerLink;
-import org.apache.qpid.server.logging.LogSubject;
 import org.apache.qpid.server.message.EnqueableMessage;
-import org.apache.qpid.server.queue.AMQQueue;
-import org.apache.qpid.server.store.ConfigurationRecoveryHandler;
 import org.apache.qpid.server.store.MessageStore;
-import org.apache.qpid.server.store.MessageStoreRecoveryHandler;
-import org.apache.qpid.server.store.StorableMessageMetaData;
+import org.apache.qpid.server.store.NullMessageStore;
 import org.apache.qpid.server.store.StoreFuture;
-import org.apache.qpid.server.store.StoredMessage;
 import org.apache.qpid.server.store.Transaction;
-import org.apache.qpid.server.store.TransactionLogRecoveryHandler;
 import org.apache.qpid.server.store.TransactionLogResource;
 
 /**
@@ -129,111 +116,14 @@ class MockStoreTransaction implements Transaction
 
     public static MessageStore createTestTransactionLog(final MockStoreTransaction storeTransaction)
     {
-        return new MessageStore()
+        return new NullMessageStore()
         {
-            public void configureMessageStore(final String name,
-                                              final MessageStoreRecoveryHandler recoveryHandler,
-                                              TransactionLogRecoveryHandler tlogRecoveryHandler,
-                                              final Configuration config, final LogSubject logSubject) throws Exception
-            {
-            }
-
-            public void close() throws Exception
-            {
-            }
-
-            public <T extends StorableMessageMetaData> StoredMessage<T> addMessage(final T metaData)
-            {
-                return null;
-            }
-
-            public boolean isPersistent()
-            {
-                return false;
-            }
-
+            @Override
             public Transaction newTransaction()
             {
                 storeTransaction.setState(TransactionState.STARTED);
                 return storeTransaction;
             }
-
-            @Override
-            public void configureConfigStore(String name,
-                    ConfigurationRecoveryHandler recoveryHandler,
-                    Configuration config, LogSubject logSubject)
-                    throws Exception
-            {
-            }
-
-            @Override
-            public void createExchange(Exchange exchange)
-                    throws AMQStoreException
-            {
-            }
-
-            @Override
-            public void removeExchange(Exchange exchange)
-                    throws AMQStoreException
-            {
-            }
-
-            @Override
-            public void bindQueue(Exchange exchange, AMQShortString routingKey,
-                    AMQQueue queue, FieldTable args) throws AMQStoreException
-            {
-            }
-
-            @Override
-            public void unbindQueue(Exchange exchange,
-                    AMQShortString routingKey, AMQQueue queue, FieldTable args)
-                    throws AMQStoreException
-            {
-            }
-
-            @Override
-            public void createQueue(AMQQueue queue) throws AMQStoreException
-            {
-            }
-
-            @Override
-            public void createQueue(AMQQueue queue, FieldTable arguments)
-                    throws AMQStoreException
-            {
-            }
-
-            @Override
-            public void removeQueue(AMQQueue queue) throws AMQStoreException
-            {
-            }
-
-            @Override
-            public void updateQueue(AMQQueue queue) throws AMQStoreException
-            {
-            }
-
-            @Override
-            public void createBrokerLink(BrokerLink link)
-                    throws AMQStoreException
-            {
-            }
-
-            @Override
-            public void deleteBrokerLink(BrokerLink link)
-                    throws AMQStoreException
-            {
-            }
-
-            @Override
-            public void createBridge(Bridge bridge) throws AMQStoreException
-            {
-            }
-
-            @Override
-            public void deleteBridge(Bridge bridge) throws AMQStoreException
-            {
-            }
-            
-        };
+       };
     }
 }
\ No newline at end of file
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java
index 9df0aec545..6b48d55fae 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/util/InternalBrokerBaseCase.java
@@ -43,6 +43,7 @@ import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.registry.IApplicationRegistry;
 import org.apache.qpid.server.store.MessageStore;
 import org.apache.qpid.server.store.TestableMemoryMessageStore;
+import org.apache.qpid.server.store.TestableMemoryMessageStoreFactory;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 import org.apache.qpid.test.utils.QpidTestCase;
 
@@ -65,10 +66,10 @@ public class InternalBrokerBaseCase extends QpidTestCase
         super.setUp();
 
         _configXml.addProperty("virtualhosts.virtualhost.name", "test");
-        _configXml.addProperty("virtualhosts.virtualhost.test.store.class", TestableMemoryMessageStore.class.getName());
+        _configXml.addProperty("virtualhosts.virtualhost.test.store.factoryclass", TestableMemoryMessageStoreFactory.class.getName());
 
         _configXml.addProperty("virtualhosts.virtualhost(-1).name", getName());
-        _configXml.addProperty("virtualhosts.virtualhost(-1)."+getName()+".store.class", TestableMemoryMessageStore.class.getName());
+        _configXml.addProperty("virtualhosts.virtualhost(-1)."+getName()+".store.factoryclass", TestableMemoryMessageStoreFactory.class.getName());
 
         createBroker();
     }
@@ -97,7 +98,7 @@ public class InternalBrokerBaseCase extends QpidTestCase
         _virtualHost.getBindingFactory().addBinding(QUEUE_NAME.toString(), _queue, defaultExchange, null);
 
         _virtualHost = _registry.getVirtualHostRegistry().getVirtualHost("test");
-        _messageStore = _virtualHost.getMessageStore();
+        _messageStore = _virtualHost.getMessageStore().getUnderlyingStore();
 
         _queue = AMQQueueFactory.createAMQQueueImpl(new AMQShortString(getName()), false, new AMQShortString("testowner"),
                                                     false, false, _virtualHost, null);
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/virtualhost/MockVirtualHost.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/virtualhost/MockVirtualHost.java
index cccf02c9f3..f27dc33dc3 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/virtualhost/MockVirtualHost.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/virtualhost/MockVirtualHost.java
@@ -275,4 +275,10 @@ public class MockVirtualHost implements VirtualHost
     {
 
     }
+
+    @Override
+    public State getState()
+    {
+        return State.ACTIVE;
+    }
 }
\ No newline at end of file
diff --git a/qpid/java/broker/src/test/java/org/apache/qpid/server/virtualhost/VirtualHostImplTest.java b/qpid/java/broker/src/test/java/org/apache/qpid/server/virtualhost/VirtualHostImplTest.java
index df7b4da426..87eb0f9d16 100644
--- a/qpid/java/broker/src/test/java/org/apache/qpid/server/virtualhost/VirtualHostImplTest.java
+++ b/qpid/java/broker/src/test/java/org/apache/qpid/server/virtualhost/VirtualHostImplTest.java
@@ -27,7 +27,7 @@ import org.apache.qpid.server.configuration.ServerConfiguration;
 import org.apache.qpid.server.exchange.Exchange;
 import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.registry.ApplicationRegistry;
-import org.apache.qpid.server.store.TestableMemoryMessageStore;
+import org.apache.qpid.server.store.MemoryMessageStoreFactory;
 import org.apache.qpid.server.util.TestApplicationRegistry;
 import org.apache.qpid.test.utils.QpidTestCase;
 
@@ -68,31 +68,6 @@ public class VirtualHostImplTest extends QpidTestCase
         customBindingTestImpl(new String[0]);
     }
 
-    private void customBindingTestImpl(final String[] routingKeys) throws Exception
-    {
-        String exchangeName = getName() +".direct";
-        String vhostName = getName();
-        String queueName = getName();
-
-        File config = writeConfigFile(vhostName, queueName, exchangeName, false, routingKeys);
-        VirtualHost vhost = createVirtualHost(vhostName, config);
-        assertNotNull("virtualhost should exist", vhost);
-
-        AMQQueue queue = vhost.getQueueRegistry().getQueue(queueName);
-        assertNotNull("queue should exist", queue);
-
-        Exchange defaultExch = vhost.getExchangeRegistry().getDefaultExchange();
-        assertTrue("queue should have been bound to default exchange with its name", defaultExch.isBound(queueName, queue));
-
-        Exchange exch = vhost.getExchangeRegistry().getExchange(exchangeName);
-        assertTrue("queue should have been bound to " + exchangeName + " with its name", exch.isBound(queueName, queue));
-
-        for(String key: routingKeys)
-        {
-            assertTrue("queue should have been bound to " + exchangeName + " with key " + key, exch.isBound(key, queue));
-        }
-    }
-
     /**
      * Tests that specifying custom routing keys for a queue in the configuration file results in failure
      * to create the vhost (since this is illegal, only queue names are used with the default exchange)
@@ -106,12 +81,32 @@ public class VirtualHostImplTest extends QpidTestCase
             createVirtualHost(getName(), config);
             fail("virtualhost creation should have failed due to illegal configuration");
         }
-        catch (ConfigurationException e)
+        catch (RuntimeException e)
         {
+            assertEquals(ConfigurationException.class, e.getCause().getClass());
             //expected
         }
     }
 
+    public void testVirtualHostBecomesActive() throws Exception
+    {
+        File config = writeConfigFile(getName(), getName(), getName() +".direct", false, new String[0]);
+        VirtualHost vhost = createVirtualHost(getName(), config);
+        assertNotNull(vhost);
+        assertEquals(State.ACTIVE, vhost.getState());
+    }
+
+    public void testVirtualHostBecomesStoppedOnClose() throws Exception
+    {
+        File config = writeConfigFile(getName(), getName(), getName() +".direct", false, new String[0]);
+        VirtualHost vhost = createVirtualHost(getName(), config);
+        assertNotNull(vhost);
+        assertEquals(State.ACTIVE, vhost.getState());
+        vhost.close();
+        assertEquals(State.STOPPED, vhost.getState());
+        assertEquals(0, vhost.getHouseKeepingActiveCount());
+    }
+
     /**
      * Tests that specifying an unknown exchange to bind the queue to results in failure to create the vhost
      */
@@ -124,12 +119,39 @@ public class VirtualHostImplTest extends QpidTestCase
             createVirtualHost(getName(), config);
             fail("virtualhost creation should have failed due to illegal configuration");
         }
-        catch (ConfigurationException e)
+        catch (RuntimeException e)
         {
+            assertEquals(ConfigurationException.class, e.getCause().getClass());
             //expected
         }
     }
 
+    private void customBindingTestImpl(final String[] routingKeys) throws Exception
+    {
+        String exchangeName = getName() +".direct";
+        String vhostName = getName();
+        String queueName = getName();
+
+        File config = writeConfigFile(vhostName, queueName, exchangeName, false, routingKeys);
+        VirtualHost vhost = createVirtualHost(vhostName, config);
+        assertNotNull("virtualhost should exist", vhost);
+
+        AMQQueue queue = vhost.getQueueRegistry().getQueue(queueName);
+        assertNotNull("queue should exist", queue);
+
+        Exchange defaultExch = vhost.getExchangeRegistry().getDefaultExchange();
+        assertTrue("queue should have been bound to default exchange with its name", defaultExch.isBound(queueName, queue));
+
+        Exchange exch = vhost.getExchangeRegistry().getExchange(exchangeName);
+        assertTrue("queue should have been bound to " + exchangeName + " with its name", exch.isBound(queueName, queue));
+
+        for(String key: routingKeys)
+        {
+            assertTrue("queue should have been bound to " + exchangeName + " with key " + key, exch.isBound(key, queue));
+        }
+    }
+
+
     private VirtualHost createVirtualHost(String vhostName, File config) throws Exception
     {
         _configuration = new ServerConfiguration(new XMLConfiguration(config));
@@ -167,11 +189,11 @@ public class VirtualHostImplTest extends QpidTestCase
             writer.write("<virtualhosts>");
             writer.write("  <default>" + vhostName + "</default>");
             writer.write("  <virtualhost>");
-            writer.write("      <store>");
-            writer.write("          <class>" + TestableMemoryMessageStore.class.getName() + "</class>");
-            writer.write("      </store>");
             writer.write("      <name>" + vhostName + "</name>");
             writer.write("      <" + vhostName + ">");
+            writer.write("              <store>");
+            writer.write("                <factoryclass>" + MemoryMessageStoreFactory.class.getName() + "</factoryclass>");
+            writer.write("              </store>");
             if(exchangeName != null && !dontDeclare)
             {
                 writer.write("          <exchanges>");