activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gtu...@apache.org
Subject svn commit: r740765 - in /activemq/trunk/activemq-core/src: main/java/org/apache/activemq/broker/region/cursors/ main/java/org/apache/activemq/store/ main/java/org/apache/activemq/store/amq/ test/java/org/apache/activemq/broker/region/
Date Wed, 04 Feb 2009 15:20:02 GMT
Author: gtully
Date: Wed Feb  4 15:19:56 2009
New Revision: 740765

URL: http://svn.apache.org/viewvc?rev=740765&view=rev
Log:
move setBatch to MessageStore interface to keep cursors store agnostic -  http://issues.apache.org/activemq/browse/AMQ-2020
- some store specific tests to follow

Modified:
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/QueueStorePrefetch.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/AbstractMessageStore.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/MessageStore.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyMessageStore.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyTopicMessageStore.java
    activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQMessageStore.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/region/QueueDuplicatesFromStoreTest.java

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/QueueStorePrefetch.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/QueueStorePrefetch.java?rev=740765&r1=740764&r2=740765&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/QueueStorePrefetch.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/QueueStorePrefetch.java
Wed Feb  4 15:19:56 2009
@@ -75,15 +75,7 @@
     }
     
     protected void setBatch(MessageId messageId) {
-        AMQMessageStore amqStore = (AMQMessageStore) store;
-        try {
-            amqStore.flush();
-        } catch (InterruptedIOException e) {
-            LOG.debug("flush on setBatch resulted in exception", e);        
-        }
-        KahaReferenceStore kahaStore = 
-            (KahaReferenceStore) amqStore.getReferenceStore();
-        kahaStore.setBatch(messageId);
+        store.setBatch(messageId);
         batchResetNeeded = false;
     }
 

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/AbstractMessageStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/AbstractMessageStore.java?rev=740765&r1=740764&r2=740765&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/AbstractMessageStore.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/AbstractMessageStore.java
Wed Feb  4 15:19:56 2009
@@ -17,6 +17,7 @@
 package org.apache.activemq.store;
 
 import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.MessageId;
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.usage.MemoryUsage;
 
@@ -42,4 +43,7 @@
 
     public void setMemoryUsage(MemoryUsage memoryUsage) {
     }
+    
+    public void setBatch(MessageId messageId) {
+    }
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/MessageStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/MessageStore.java?rev=740765&r1=740764&r2=740765&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/MessageStore.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/MessageStore.java
Wed Feb  4 15:19:56 2009
@@ -110,4 +110,11 @@
     void recoverNextMessages(int maxReturned, MessageRecoveryListener listener) throws Exception;
 
     void dispose(ConnectionContext context);
+
+    /**
+     * allow caching cursors to set the current batch offset when cache is exhausted
+     * @param messageId
+     */
+    void setBatch(MessageId messageId);
+    
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyMessageStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyMessageStore.java?rev=740765&r1=740764&r2=740765&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyMessageStore.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyMessageStore.java
Wed Feb  4 15:19:56 2009
@@ -92,4 +92,8 @@
         delegate.resetBatching();
 
     }
+
+    public void setBatch(MessageId messageId) {
+        delegate.setBatch(messageId);
+    }
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyTopicMessageStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyTopicMessageStore.java?rev=740765&r1=740764&r2=740765&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyTopicMessageStore.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyTopicMessageStore.java
Wed Feb  4 15:19:56 2009
@@ -134,4 +134,8 @@
         delegate.resetBatching();
 
     }
+
+    public void setBatch(MessageId messageId) {
+        delegate.setBatch(messageId);
+    }
 }

Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQMessageStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQMessageStore.java?rev=740765&r1=740764&r2=740765&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQMessageStore.java
(original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQMessageStore.java
Wed Feb  4 15:19:56 2009
@@ -558,4 +558,14 @@
         referenceStore.dispose(context);
         super.dispose(context);
     }
+
+    public void setBatch(MessageId messageId) {
+        try {
+            flush();
+        } catch (InterruptedIOException e) {
+            LOG.debug("flush on setBatch resulted in exception", e);
+        }
+        getReferenceStore().setBatch(messageId);
+    }
+    
 }

Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/region/QueueDuplicatesFromStoreTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/region/QueueDuplicatesFromStoreTest.java?rev=740765&r1=740764&r2=740765&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/region/QueueDuplicatesFromStoreTest.java
(original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/region/QueueDuplicatesFromStoreTest.java
Wed Feb  4 15:19:56 2009
@@ -46,6 +46,7 @@
 import org.apache.activemq.filter.MessageEvaluationContext;
 import org.apache.activemq.state.ProducerState;
 import org.apache.activemq.store.MessageStore;
+import org.apache.activemq.store.PersistenceAdapter;
 import org.apache.activemq.store.amq.AMQPersistenceAdapter;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -70,16 +71,20 @@
     final int ackWindow = 50;
     final int ackBatchSize = 50;
     final int fullWindow = 200;
-    final int count = 20000;
+    protected int count = 20000;
 
     public void setUp() throws Exception {
-        brokerService = new BrokerService();
+        brokerService = createBroker();
         brokerService.setUseJmx(false);
         brokerService.deleteAllMessages();
         brokerService.start();        
     }
 
-    public void tearDown() throws Exception {
+    protected BrokerService createBroker() throws Exception {
+        return new BrokerService();
+    }
+
+	public void tearDown() throws Exception {
         brokerService.stop();
     }
 
@@ -92,8 +97,7 @@
     }
 
     public void doTestNoDuplicateAfterCacheFullAndAcked(final int auditDepth) throws Exception
{
-        final AMQPersistenceAdapter persistenceAdapter = 
-            (AMQPersistenceAdapter) brokerService.getPersistenceAdapter();
+        final PersistenceAdapter persistenceAdapter =  brokerService.getPersistenceAdapter();
         final MessageStore queueMessageStore = 
             persistenceAdapter.createQueueMessageStore(destination);
         final ConnectionContext contextNotInTx = new ConnectionContext();
@@ -127,10 +131,9 @@
             Message message = getMessage(i);
             queue.send(producerExchange, message);
         }
-        
-        assertEquals("store count is correct", count, queueMessageStore
-                .getMessageCount());
 
+        assertEquals("store count is correct", count, queueMessageStore.getMessageCount());
+        
         // pull from store in small windows
         Subscription subscription = new Subscription() {
 
@@ -305,7 +308,6 @@
                     if (removeIndex % 1000 == 0) {
                         LOG.info("acked: " + removeIndex);
                         persistenceAdapter.checkpoint(true);
-                        persistenceAdapter.cleanup();
                     }
                 }
             }



Mime
View raw message