1 files changed, 51 insertions, 134 deletions
diff --git a/java/broker/src/main/java/org/apache/qpid/server/store/MemoryMessageStore.java b/java/broker/src/main/java/org/apache/qpid/server/store/MemoryMessageStore.java
index 546a81a050..59624b7a75 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/store/MemoryMessageStore.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/store/MemoryMessageStore.java
@@ -21,200 +21,117 @@
 package org.apache.qpid.server.store;
 
 import org.apache.commons.configuration.Configuration;
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.AMQException;
 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.logging.actors.CurrentActor;
-import org.apache.qpid.server.logging.messages.ConfigStoreMessages;
-import org.apache.qpid.server.logging.messages.MessageStoreMessages;
 import org.apache.qpid.server.message.EnqueableMessage;
-import org.apache.qpid.server.queue.AMQQueue;
 
-import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
-/** A simple message store that stores the messages in a threadsafe structure in memory. */
-public class MemoryMessageStore implements MessageStore, DurableConfigurationStore
+/** A simple message store that stores the messages in a thread-safe structure in memory. */
+public class MemoryMessageStore extends NullMessageStore
 {
-    private static final Logger _log = Logger.getLogger(MemoryMessageStore.class);
-
-    private static final int DEFAULT_HASHTABLE_CAPACITY = 50000;
-
-    private static final String HASHTABLE_CAPACITY_CONFIG = "hashtable-capacity";
-
-
     private final AtomicLong _messageId = new AtomicLong(1);
-    private AtomicBoolean _closed = new AtomicBoolean(false);
-    private LogSubject _logSubject;
+    private final AtomicBoolean _closed = new AtomicBoolean(false);
 
     private static final Transaction IN_MEMORY_TRANSACTION = new Transaction()
     {
+        @Override
+        public StoreFuture commitTranAsync() throws AMQStoreException
+        {
+            return StoreFuture.IMMEDIATE_FUTURE;
+        }
+
+        @Override
         public void enqueueMessage(TransactionLogResource queue, EnqueableMessage message) throws AMQStoreException
         {
         }
 
+        @Override
         public void dequeueMessage(TransactionLogResource  queue, EnqueableMessage message) throws AMQStoreException
         {
         }
 
+        @Override
         public void commitTran() throws AMQStoreException
         {
         }
 
-        public StoreFuture commitTranAsync() throws AMQStoreException
+        @Override
+        public void abortTran() throws AMQStoreException
         {
-            return IMMEDIATE_FUTURE;
         }
 
-        public void abortTran() throws AMQStoreException
+        @Override
+        public void removeXid(long format, byte[] globalId, byte[] branchId)
         {
         }
 
-    };
-
-    public void configureConfigStore(String name, ConfigurationRecoveryHandler handler, Configuration configuration, LogSubject logSubject) throws Exception
-    {
-        _logSubject = logSubject;
-        CurrentActor.get().message(_logSubject, ConfigStoreMessages.CREATED(this.getClass().getName()));
-
-
-    }
-
-    public void configureMessageStore(String name,
-                                      MessageStoreRecoveryHandler recoveryHandler,
-                                      Configuration config,
-                                      LogSubject logSubject) throws Exception
-    {
-        if(_logSubject == null)
+        @Override
+        public void recordXid(long format, byte[] globalId, byte[] branchId, Record[] enqueues, Record[] dequeues)
         {
-            _logSubject = logSubject;
         }
-        int hashtableCapacity = config.getInt(name + "." + HASHTABLE_CAPACITY_CONFIG, DEFAULT_HASHTABLE_CAPACITY);
-        _log.info("Using capacity " + hashtableCapacity + " for hash tables");
-        CurrentActor.get().message(_logSubject, MessageStoreMessages.CREATED(this.getClass().getName()));
-    }
-
-    public void close() throws Exception
-    {
-        _closed.getAndSet(true);
-        CurrentActor.get().message(_logSubject,MessageStoreMessages.CLOSED());
-
-    }
-
-    public StoredMessage addMessage(StorableMessageMetaData metaData)
-    {
-        final long id = _messageId.getAndIncrement();
-        StoredMemoryMessage message = new StoredMemoryMessage(id, metaData);
-
-        return message;
-    }
-
-
-    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
-    {
-        // Not requred to do anything
-    }
+    };
 
-    public void createQueue(AMQQueue queue, FieldTable arguments) throws AMQStoreException
-    {
-        // Not required to do anything
-    }
+    private final StateManager _stateManager;
+    private final EventManager _eventManager = new EventManager();
 
-    public void removeQueue(final AMQQueue queue) throws AMQStoreException
-    {
-        // Not required to do anything
-    }
-    
-    public void updateQueue(final AMQQueue queue) throws AMQStoreException
+    public MemoryMessageStore()
     {
-        // Not required to do anything
+        _stateManager = new StateManager(_eventManager);
     }
 
-    public void createBrokerLink(final BrokerLink link) throws AMQStoreException
+    @Override
+    public void configureConfigStore(String name, ConfigurationRecoveryHandler recoveryHandler, Configuration config) throws Exception
     {
-
+        _stateManager.attainState(State.CONFIGURING);
     }
 
-    public void deleteBrokerLink(final BrokerLink link) throws AMQStoreException
+    @Override
+    public void configureMessageStore(String name, MessageStoreRecoveryHandler recoveryHandler, TransactionLogRecoveryHandler tlogRecoveryHandler, Configuration config) throws Exception
     {
-
+        _stateManager.attainState(State.CONFIGURED);
     }
 
-    public void createBridge(final Bridge bridge) throws AMQStoreException
+    @Override
+    public void activate() throws Exception
     {
-
+        _stateManager.attainState(State.RECOVERING);
+        
+        _stateManager.attainState(State.ACTIVE);
     }
 
-    public void deleteBridge(final Bridge bridge) throws AMQStoreException
+    @Override
+    public StoredMessage addMessage(StorableMessageMetaData metaData)
     {
+        final long id = _messageId.getAndIncrement();
+        StoredMemoryMessage message = new StoredMemoryMessage(id, metaData);
 
+        return message;
     }
 
-    public void configureTransactionLog(String name,
-                                        TransactionLogRecoveryHandler recoveryHandler,
-                                        Configuration storeConfiguration,
-                                        LogSubject logSubject) throws Exception
-    {
-        //To change body of implemented methods use File | Settings | File Templates.
-    }
-
+    @Override
     public Transaction newTransaction()
     {
         return IN_MEMORY_TRANSACTION;
     }
 
-
-    public List<AMQQueue> createQueues() throws AMQException
+    @Override
+    public boolean isPersistent()
     {
-        return null;
+        return false;
     }
 
-    public Long getNewMessageId()
+    @Override
+    public void close() throws Exception
     {
-        return _messageId.getAndIncrement();
+        _stateManager.attainState(State.CLOSING);
+        _closed.getAndSet(true);
+        _stateManager.attainState(State.CLOSED);
     }
 
-    public boolean isPersistent()
+    @Override
+    public void addEventListener(EventListener eventListener, Event... events)
     {
-        return false;
-    }
-
-    private void checkNotClosed() throws MessageStoreClosedException
-     {
-        if (_closed.get())
-        {
-            throw new MessageStoreClosedException();
-        }
+        _eventManager.addEventListener(eventListener, events);
     }
-
-
 }