qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ritch...@apache.org
Subject svn commit: r764850 - in /qpid/trunk/qpid/java: broker/ systests/src/main/java/org/apache/qpid/server/store/SlowMessageStore.java
Date Tue, 14 Apr 2009 15:54:16 GMT
Author: ritchiem
Date: Tue Apr 14 15:54:16 2009
New Revision: 764850

URL: http://svn.apache.org/viewvc?rev=764850&view=rev
Log:
QPID-1807 : Add 0.5-fix broker and update SlowMessageStore to use MessageStores rather than
TransactionLogs

Added:
    qpid/trunk/qpid/java/broker/   (props changed)
      - copied from r764833, qpid/branches/0.5-fix/qpid/java/broker/
Modified:
    qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/store/SlowMessageStore.java

Propchange: qpid/trunk/qpid/java/broker/
------------------------------------------------------------------------------
--- svn:ignore (added)
+++ svn:ignore Tue Apr 14 15:54:16 2009
@@ -0,0 +1,9 @@
+*.iml
+intellijclasses
+log
+target
+release
+qpid-broker.ipr
+qpid-broker.iws
+.classpath
+.project

Propchange: qpid/trunk/qpid/java/broker/
------------------------------------------------------------------------------
--- svn:mergeinfo (added)
+++ svn:mergeinfo Tue Apr 14 15:54:16 2009
@@ -0,0 +1 @@
+/qpid/trunk/qpid/java/broker:742626,743015,743028-743029,743304,743306,743311,743357,744113,747363,747367,747369-747370,747376,747783,747868-747870,747875,748561,748591,748641,748680,748686,749149,749282,749285,749315,749340,749572,753219-753220,753253,754934,754958,755256,757258,757270,758730,759097,760919,761721,762365,762992,763959,764109,764140,764790

Modified: qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/store/SlowMessageStore.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/store/SlowMessageStore.java?rev=764850&r1=764849&r2=764850&view=diff
==============================================================================
--- qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/store/SlowMessageStore.java
(original)
+++ qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/store/SlowMessageStore.java
Tue Apr 14 15:54:16 2009
@@ -31,62 +31,54 @@
 import org.apache.qpid.server.exchange.Exchange;
 import org.apache.qpid.server.queue.AMQQueue;
 import org.apache.qpid.server.queue.MessageMetaData;
-import org.apache.qpid.server.transactionlog.TransactionLog;
-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
+public class SlowMessageStore implements MessageStore
 {
     private static final Logger _logger = Logger.getLogger(SlowMessageStore.class);
     private static final String DELAYS = "delays";
     private HashMap<String, Long> _preDelays = new HashMap<String, Long>();
     private HashMap<String, Long> _postDelays = new HashMap<String, Long>();
     private long _defaultDelay = 0L;
-    private TransactionLog _realTransactionLog = new MemoryMessageStore();
-    private RoutingTable _realRoutingTable = (RoutingTable)_realTransactionLog;
+    private MessageStore _realStore = new MemoryMessageStore();
     private static final String PRE = "pre";
     private static final String POST = "post";
     private String DEFAULT_DELAY = "default";
 
-    public Object configure(VirtualHost virtualHost, String base, VirtualHostConfiguration
config) throws Exception
+    public void configure(VirtualHost virtualHost, String base, VirtualHostConfiguration
config) throws Exception
     {
-        _logger.warn("Starting SlowMessageStore on Virtualhost:" + virtualHost.getName());
+        _logger.info("Starting SlowMessageStore on Virtualhost:" + virtualHost.getName());
         Configuration delays = config.getStoreConfiguration().subset(DELAYS);
 
         configureDelays(delays);
 
-        String transactionLogClass = config.getTransactionLogClass();
+        String messageStoreClass = config.getStoreConfiguration().getString("realStore");
 
         if (delays.containsKey(DEFAULT_DELAY))
         {
             _defaultDelay = delays.getLong(DEFAULT_DELAY);
-            _logger.warn("Delay is:" + _defaultDelay);
         }
 
-        if (transactionLogClass != null)
+        if (messageStoreClass != null)
         {
-            Class clazz = Class.forName(transactionLogClass);
-            if (clazz != this.getClass())
-            {
+            Class clazz = Class.forName(messageStoreClass);
 
-                Object o = clazz.newInstance();
+            Object o = clazz.newInstance();
 
-                if (!(o instanceof TransactionLog))
-                {
-                    throw new ClassCastException("TransactionLog class must implement " +
TransactionLog.class + ". Class " + clazz +
-                    " does not.");
-                }
-                _realTransactionLog = (TransactionLog) o;
+            if (!(o instanceof MessageStore))
+            {
+                throw new ClassCastException("Message store class must implement " + MessageStore.class
+ ". Class " + clazz +
+                                             " does not.");
             }
+            _realStore = (MessageStore) o;
+            _realStore.configure(virtualHost, base + ".store", config);
+        }
+        else
+        {
+            _realStore.configure(virtualHost, base + ".store", 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)
@@ -159,35 +151,42 @@
     public void close() throws Exception
     {
         doPreDelay("close");
-        _realTransactionLog.close();
+        _realStore.close();
         doPostDelay("close");
     }
 
+    public void removeMessage(StoreContext storeContext, Long messageId) throws AMQException
+    {
+        doPreDelay("removeMessage");
+        _realStore.removeMessage(storeContext, messageId);
+        doPostDelay("removeMessage");
+    }
+
     public void createExchange(Exchange exchange) throws AMQException
     {
         doPreDelay("createExchange");
-        _realRoutingTable.createExchange(exchange);
+        _realStore.createExchange(exchange);
         doPostDelay("createExchange");
     }
 
     public void removeExchange(Exchange exchange) throws AMQException
     {
         doPreDelay("removeExchange");
-        _realRoutingTable.removeExchange(exchange);
+        _realStore.removeExchange(exchange);
         doPostDelay("removeExchange");
     }
 
     public void bindQueue(Exchange exchange, AMQShortString routingKey, AMQQueue queue, FieldTable
args) throws AMQException
     {
         doPreDelay("bindQueue");
-        _realRoutingTable.bindQueue(exchange, routingKey, queue, args);
+        _realStore.bindQueue(exchange, routingKey, queue, args);
         doPostDelay("bindQueue");
     }
 
     public void unbindQueue(Exchange exchange, AMQShortString routingKey, AMQQueue queue,
FieldTable args) throws AMQException
     {
         doPreDelay("unbindQueue");
-        _realRoutingTable.unbindQueue(exchange, routingKey, queue, args);
+        _realStore.unbindQueue(exchange, routingKey, queue, args);
         doPostDelay("unbindQueue");
     }
 
@@ -199,100 +198,101 @@
     public void createQueue(AMQQueue queue, FieldTable arguments) throws AMQException
     {
         doPreDelay("createQueue");
-        _realRoutingTable.createQueue(queue, arguments);
+        _realStore.createQueue(queue, arguments);
         doPostDelay("createQueue");
     }
 
     public void removeQueue(AMQQueue queue) throws AMQException
     {
         doPreDelay("removeQueue");
-        _realRoutingTable.removeQueue(queue);
+        _realStore.removeQueue(queue);
         doPostDelay("removeQueue");
     }
 
-    public void enqueueMessage(StoreContext context, ArrayList<AMQQueue> queues, Long
messageId) throws AMQException
+    public void enqueueMessage(StoreContext context, AMQQueue queue, Long messageId) throws
AMQException
     {
         doPreDelay("enqueueMessage");
-        _realTransactionLog.enqueueMessage(context, queues, messageId);
+        _realStore.enqueueMessage(context, queue, messageId);
         doPostDelay("enqueueMessage");
     }
 
     public void dequeueMessage(StoreContext context, AMQQueue queue, Long messageId) throws
AMQException
     {
         doPreDelay("dequeueMessage");
-        _realTransactionLog.dequeueMessage(context, queue, messageId);
-        doPostDelay("dequeueMessage");
-    }
-
-    public void removeMessage(StoreContext context, Long messageId) throws AMQException
-    {
-        doPreDelay("dequeueMessage");
-        _realTransactionLog.removeMessage(context, messageId);
+        _realStore.dequeueMessage(context, queue, messageId);
         doPostDelay("dequeueMessage");
     }
 
     public void beginTran(StoreContext context) throws AMQException
     {
         doPreDelay("beginTran");
-        _realTransactionLog.beginTran(context);
+        _realStore.beginTran(context);
         doPostDelay("beginTran");
     }
 
     public void commitTran(StoreContext context) throws AMQException
     {
         doPreDelay("commitTran");
-        _realTransactionLog.commitTran(context);
+        _realStore.commitTran(context);
         doPostDelay("commitTran");
     }
 
     public void abortTran(StoreContext context) throws AMQException
     {
         doPreDelay("abortTran");
-        _realTransactionLog.abortTran(context);
+        _realStore.abortTran(context);
         doPostDelay("abortTran");
     }
 
     public boolean inTran(StoreContext context)
     {
         doPreDelay("inTran");
-        boolean b = _realTransactionLog.inTran(context);
+        boolean b = _realStore.inTran(context);
         doPostDelay("inTran");
         return b;
     }
 
+    public Long getNewMessageId()
+    {
+        doPreDelay("getNewMessageId");
+        Long l = _realStore.getNewMessageId();
+        doPostDelay("getNewMessageId");
+        return l;
+    }
+
     public void storeContentBodyChunk(StoreContext context, Long messageId, int index, ContentChunk
contentBody, boolean lastContentBody) throws AMQException
     {
         doPreDelay("storeContentBodyChunk");
-        _realTransactionLog.storeContentBodyChunk(context, messageId, index, contentBody,
lastContentBody);
+        _realStore.storeContentBodyChunk(context, messageId, index, contentBody, lastContentBody);
         doPostDelay("storeContentBodyChunk");
     }
 
     public void storeMessageMetaData(StoreContext context, Long messageId, MessageMetaData
messageMetaData) throws AMQException
     {
         doPreDelay("storeMessageMetaData");
-        _realTransactionLog.storeMessageMetaData(context, messageId, messageMetaData);
+        _realStore.storeMessageMetaData(context, messageId, messageMetaData);
         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 = _realStore.getMessageMetaData(context, messageId);
+        doPostDelay("getMessageMetaData");
+        return mmd;
+    }
+
+    public ContentChunk getContentBodyChunk(StoreContext context, Long messageId, int index)
throws AMQException
+    {
+        doPreDelay("getContentBodyChunk");
+        ContentChunk c = _realStore.getContentBodyChunk(context, messageId, index);
+        doPostDelay("getContentBodyChunk");
+        return c;
+    }
+
     public boolean isPersistent()
     {
-        return _realTransactionLog.isPersistent();
+        return _realStore.isPersistent();
     }
 
 }



---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project:      http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org


Mime
View raw message