You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ac...@apache.org on 2006/07/28 11:54:33 UTC
svn commit: r426461 - in /incubator/activemq/trunk:
activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/
activemq-core/src/test/java/org/apache/activemq/broker/ft/
activemq-core/src/test/java/org/apache/activemq/broker/store/
activemq-core...
Author: aco
Date: Fri Jul 28 02:54:32 2006
New Revision: 426461
URL: http://svn.apache.org/viewvc?rev=426461&view=rev
Log:
Renamed the kahaPersistentAdaptor to kahaPersistenceAdapter to be consistent with other persistence adapters. Makes configuration slightly more intuitive. :)
Added:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaPersistenceAdapter.java
Removed:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaPersistentAdaptor.java
Modified:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaTransactionStore.java
incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/ft/TransactedTopicMasterSlaveTest.java
incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/store/KahaRecoveryBrokerTest.java
incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/store/KahaXARecoveryBrokerTest.java
incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/perf/KahaDurableTopicTest.java
incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/perf/KahaQueueTest.java
incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/ft/master.xml
incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/ft/slave.xml
incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/store/kahabroker.xml
incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/perf/kahaBroker.xml
incubator/activemq/trunk/activemq-perftest/src/main/resources/broker-conf/kaha.xml
Added: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaPersistenceAdapter.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaPersistenceAdapter.java?rev=426461&view=auto
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaPersistenceAdapter.java (added)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaPersistenceAdapter.java Fri Jul 28 02:54:32 2006
@@ -0,0 +1,179 @@
+/**
+ *
+ * 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.activemq.store.kahadaptor;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
+import org.apache.activemq.broker.ConnectionContext;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQQueue;
+import org.apache.activemq.command.ActiveMQTopic;
+import org.apache.activemq.kaha.MapContainer;
+import org.apache.activemq.kaha.Store;
+import org.apache.activemq.kaha.StoreFactory;
+import org.apache.activemq.kaha.StringMarshaller;
+import org.apache.activemq.memory.UsageManager;
+import org.apache.activemq.openwire.OpenWireFormat;
+import org.apache.activemq.store.MessageStore;
+import org.apache.activemq.store.PersistenceAdapter;
+import org.apache.activemq.store.TopicMessageStore;
+import org.apache.activemq.store.TransactionStore;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
+/**
+ * @org.apache.xbean.XBean
+ *
+ * @version $Revision: 1.4 $
+ */
+public class KahaPersistenceAdapter implements PersistenceAdapter{
+ private static final Log log=LogFactory.getLog(KahaPersistenceAdapter.class);
+ static final String PREPARED_TRANSACTIONS_NAME="PreparedTransactions";
+ KahaTransactionStore transactionStore;
+ ConcurrentHashMap topics=new ConcurrentHashMap();
+ ConcurrentHashMap queues=new ConcurrentHashMap();
+ ConcurrentHashMap messageStores=new ConcurrentHashMap();
+ private boolean useExternalMessageReferences;
+ private OpenWireFormat wireFormat=new OpenWireFormat();
+ Store store;
+
+ public KahaPersistenceAdapter(File dir) throws IOException{
+ if(!dir.exists()){
+ dir.mkdirs();
+ }
+ String name=dir.getAbsolutePath()+File.separator+"kaha.db";
+ store=StoreFactory.open(name,"rw");
+ }
+
+ public Set getDestinations(){
+
+ Set rc=new HashSet();
+ try {
+ for(Iterator i=store.getMapContainerIds().iterator();i.hasNext();){
+ Object obj=i.next();
+ if(obj instanceof ActiveMQDestination){
+ rc.add(obj);
+ }
+ }
+ }catch(IOException e){
+ log.error("Failed to get destinations " ,e);
+ }
+ return rc;
+ }
+
+ public synchronized MessageStore createQueueMessageStore(ActiveMQQueue destination) throws IOException{
+ MessageStore rc=(MessageStore) queues.get(destination);
+ if(rc==null){
+ rc=new KahaMessageStore(getMapContainer(destination,"queue-data"),destination);
+ messageStores.put(destination, rc);
+ if(transactionStore!=null){
+ rc=transactionStore.proxy(rc);
+ }
+ queues.put(destination,rc);
+ }
+ return rc;
+ }
+
+ public synchronized TopicMessageStore createTopicMessageStore(ActiveMQTopic destination) throws IOException{
+ TopicMessageStore rc=(TopicMessageStore) topics.get(destination);
+ if(rc==null){
+ MapContainer messageContainer=getMapContainer(destination,"topic-data");
+ MapContainer subsContainer=getMapContainer(destination.toString()+"-Subscriptions","topic-subs");
+ MapContainer ackContainer=store.getMapContainer(destination.toString(),"topic-acks");
+ ackContainer.setKeyMarshaller(new StringMarshaller());
+ ackContainer.setValueMarshaller(new AtomicIntegerMarshaller());
+ rc=new KahaTopicMessageStore(store,messageContainer,ackContainer,subsContainer,destination);
+ messageStores.put(destination, rc);
+ if(transactionStore!=null){
+ rc=transactionStore.proxy(rc);
+ }
+ topics.put(destination,rc);
+
+ }
+ return rc;
+ }
+
+ protected MessageStore retrieveMessageStore(Object id){
+ MessageStore result = (MessageStore) messageStores.get(id);
+ return result;
+ }
+
+ public TransactionStore createTransactionStore() throws IOException{
+ if(transactionStore==null){
+ MapContainer container=store.getMapContainer(PREPARED_TRANSACTIONS_NAME,"transactions");
+ container.setKeyMarshaller(new CommandMarshaller(wireFormat));
+ container.setValueMarshaller(new TransactionMarshaller(wireFormat));
+ container.load();
+ transactionStore=new KahaTransactionStore(this,container);
+ }
+ return transactionStore;
+ }
+
+ public void beginTransaction(ConnectionContext context){}
+
+ public void commitTransaction(ConnectionContext context) throws IOException{
+ store.force();
+ }
+
+ public void rollbackTransaction(ConnectionContext context){}
+
+ public void start() throws Exception{}
+
+ public void stop() throws Exception{
+ store.close();
+ }
+
+ public long getLastMessageBrokerSequenceId() throws IOException{
+ return 0;
+ }
+
+ public void deleteAllMessages() throws IOException{
+ if(store!=null){
+ store.delete();
+ }
+ }
+
+ public boolean isUseExternalMessageReferences(){
+ return useExternalMessageReferences;
+ }
+
+ public void setUseExternalMessageReferences(boolean useExternalMessageReferences){
+ this.useExternalMessageReferences=useExternalMessageReferences;
+ }
+
+ protected MapContainer getMapContainer(Object id,String containerName) throws IOException{
+ MapContainer container=store.getMapContainer(id,containerName);
+ container.setKeyMarshaller(new StringMarshaller());
+ if(useExternalMessageReferences){
+ container.setValueMarshaller(new StringMarshaller());
+ }else{
+ container.setValueMarshaller(new CommandMarshaller(wireFormat));
+ }
+ container.load();
+ return container;
+ }
+
+ /**
+ * @param usageManager
+ * The UsageManager that is controlling the broker's memory usage.
+ */
+ public void setUsageManager(UsageManager usageManager){}
+}
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaTransactionStore.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaTransactionStore.java?rev=426461&r1=426460&r2=426461&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaTransactionStore.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaTransactionStore.java Fri Jul 28 02:54:32 2006
@@ -43,9 +43,9 @@
public class KahaTransactionStore implements TransactionStore{
private Map transactions=new ConcurrentHashMap();
private Map prepared;
- private KahaPersistentAdaptor adaptor;
+ private KahaPersistenceAdapter adaptor;
- KahaTransactionStore(KahaPersistentAdaptor adaptor,Map preparedMap){
+ KahaTransactionStore(KahaPersistenceAdapter adaptor,Map preparedMap){
this.adaptor=adaptor;
this.prepared=preparedMap;
}
Modified: incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/ft/TransactedTopicMasterSlaveTest.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/ft/TransactedTopicMasterSlaveTest.java?rev=426461&r1=426460&r2=426461&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/ft/TransactedTopicMasterSlaveTest.java (original)
+++ incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/ft/TransactedTopicMasterSlaveTest.java Fri Jul 28 02:54:32 2006
@@ -22,7 +22,7 @@
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.JmsTopicTransactionTest;
import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.store.kahadaptor.KahaPersistentAdaptor;
+import org.apache.activemq.store.kahadaptor.KahaPersistenceAdapter;
import org.apache.activemq.test.JmsResourceProvider;
/**
* Test failover for Queues
@@ -39,7 +39,7 @@
// this will create the main (or master broker)
broker=createBroker();
broker.start();
- KahaPersistentAdaptor adaptor=new KahaPersistentAdaptor(new File("activemq-data/slave"));
+ KahaPersistenceAdapter adaptor=new KahaPersistenceAdapter(new File("activemq-data/slave"));
slave = new BrokerService();
slave.setBrokerName("slave");
slave.setPersistenceAdapter(adaptor);
@@ -66,7 +66,7 @@
protected BrokerService createBroker() throws Exception,URISyntaxException{
BrokerService broker=new BrokerService();
broker.setBrokerName("master");
- KahaPersistentAdaptor adaptor=new KahaPersistentAdaptor(new File("activemq-data/master"));
+ KahaPersistenceAdapter adaptor=new KahaPersistenceAdapter(new File("activemq-data/master"));
broker.setPersistenceAdapter(adaptor);
broker.addConnector("tcp://localhost:62001");
broker.setDeleteAllMessagesOnStartup(true);
Modified: incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/store/KahaRecoveryBrokerTest.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/store/KahaRecoveryBrokerTest.java?rev=426461&r1=426460&r2=426461&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/store/KahaRecoveryBrokerTest.java (original)
+++ incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/store/KahaRecoveryBrokerTest.java Fri Jul 28 02:54:32 2006
@@ -22,7 +22,7 @@
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.RecoveryBrokerTest;
-import org.apache.activemq.store.kahadaptor.KahaPersistentAdaptor;
+import org.apache.activemq.store.kahadaptor.KahaPersistenceAdapter;
import org.apache.activemq.xbean.BrokerFactoryBean;
import org.springframework.core.io.ClassPathResource;
@@ -41,7 +41,7 @@
protected BrokerService createRestartedBroker() throws Exception {
BrokerService broker = new BrokerService();
- KahaPersistentAdaptor adaptor = new KahaPersistentAdaptor(new File( System.getProperty("basedir", ".")+"/target/activemq-data/kaha-store.db"));
+ KahaPersistenceAdapter adaptor = new KahaPersistenceAdapter(new File( System.getProperty("basedir", ".")+"/target/activemq-data/kaha-store.db"));
broker.setPersistenceAdapter(adaptor);
broker.addConnector("tcp://localhost:0");
return broker;
Modified: incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/store/KahaXARecoveryBrokerTest.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/store/KahaXARecoveryBrokerTest.java?rev=426461&r1=426460&r2=426461&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/store/KahaXARecoveryBrokerTest.java (original)
+++ incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/store/KahaXARecoveryBrokerTest.java Fri Jul 28 02:54:32 2006
@@ -22,7 +22,7 @@
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.XARecoveryBrokerTest;
-import org.apache.activemq.store.kahadaptor.KahaPersistentAdaptor;
+import org.apache.activemq.store.kahadaptor.KahaPersistenceAdapter;
import org.apache.activemq.xbean.BrokerFactoryBean;
import org.springframework.core.io.ClassPathResource;
@@ -50,7 +50,7 @@
protected BrokerService createRestartedBroker() throws Exception {
BrokerService broker = new BrokerService();
- KahaPersistentAdaptor adaptor = new KahaPersistentAdaptor(new File( System.getProperty("basedir", ".")+"/target/activemq-data/kaha-store.db"));
+ KahaPersistenceAdapter adaptor = new KahaPersistenceAdapter(new File( System.getProperty("basedir", ".")+"/target/activemq-data/kaha-store.db"));
broker.setPersistenceAdapter(adaptor);
broker.addConnector("tcp://localhost:0");
return broker;
Modified: incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/perf/KahaDurableTopicTest.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/perf/KahaDurableTopicTest.java?rev=426461&r1=426460&r2=426461&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/perf/KahaDurableTopicTest.java (original)
+++ incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/perf/KahaDurableTopicTest.java Fri Jul 28 02:54:32 2006
@@ -19,7 +19,7 @@
import java.io.File;
import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.store.kahadaptor.KahaPersistentAdaptor;
+import org.apache.activemq.store.kahadaptor.KahaPersistenceAdapter;
/**
* @version $Revision: 1.3 $
*/
@@ -37,7 +37,7 @@
*/
protected void configureBroker(BrokerService answer) throws Exception{
- KahaPersistentAdaptor adaptor = new KahaPersistentAdaptor(new File("activemq-data/perfTest"));
+ KahaPersistenceAdapter adaptor = new KahaPersistenceAdapter(new File("activemq-data/perfTest"));
answer.setPersistenceAdapter(adaptor);
answer.addConnector(bindAddress);
answer.setDeleteAllMessagesOnStartup(true);
@@ -47,4 +47,4 @@
-}
\ No newline at end of file
+}
Modified: incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/perf/KahaQueueTest.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/perf/KahaQueueTest.java?rev=426461&r1=426460&r2=426461&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/perf/KahaQueueTest.java (original)
+++ incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/perf/KahaQueueTest.java Fri Jul 28 02:54:32 2006
@@ -22,7 +22,7 @@
import javax.jms.JMSException;
import javax.jms.Session;
import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.store.kahadaptor.KahaPersistentAdaptor;
+import org.apache.activemq.store.kahadaptor.KahaPersistenceAdapter;
/**
* @version $Revision: 1.3 $
*/
@@ -30,10 +30,10 @@
protected void configureBroker(BrokerService answer) throws Exception{
- KahaPersistentAdaptor adaptor = new KahaPersistentAdaptor(new File("activemq-data/perfTest"));
+ KahaPersistenceAdapter adaptor = new KahaPersistenceAdapter(new File("activemq-data/perfTest"));
answer.setPersistenceAdapter(adaptor);
answer.addConnector(bindAddress);
answer.setDeleteAllMessagesOnStartup(true);
}
-}
\ No newline at end of file
+}
Modified: incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/ft/master.xml
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/ft/master.xml?rev=426461&r1=426460&r2=426461&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/ft/master.xml (original)
+++ incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/ft/master.xml Fri Jul 28 02:54:32 2006
@@ -24,7 +24,7 @@
</transportConnectors>
<persistenceAdapter>
- <kahaPersistentAdaptor dir = "${basedir}/target/activemq-data/master"/>
+ <kahaPersistenceAdapter dir = "${basedir}/target/activemq-data/master"/>
</persistenceAdapter>
</broker>
Modified: incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/ft/slave.xml
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/ft/slave.xml?rev=426461&r1=426460&r2=426461&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/ft/slave.xml (original)
+++ incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/ft/slave.xml Fri Jul 28 02:54:32 2006
@@ -26,7 +26,7 @@
<persistenceAdapter>
- <kahaPersistentAdaptor dir = "${basedir}/target/activemq-data/slave"/>
+ <kahaPersistenceAdapter dir = "${basedir}/target/activemq-data/slave"/>
</persistenceAdapter>
</broker>
Modified: incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/store/kahabroker.xml
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/store/kahabroker.xml?rev=426461&r1=426460&r2=426461&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/store/kahabroker.xml (original)
+++ incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/broker/store/kahabroker.xml Fri Jul 28 02:54:32 2006
@@ -24,7 +24,7 @@
</transportConnectors>
<persistenceAdapter>
- <kahaPersistentAdaptor dir = "${basedir}/target/activemq-data/kaha-broker.db"/>
+ <kahaPersistenceAdapter dir = "${basedir}/target/activemq-data/kaha-broker.db"/>
</persistenceAdapter>
</broker>
Modified: incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/perf/kahaBroker.xml
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/perf/kahaBroker.xml?rev=426461&r1=426460&r2=426461&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/perf/kahaBroker.xml (original)
+++ incubator/activemq/trunk/activemq-core/src/test/resources/org/apache/activemq/perf/kahaBroker.xml Fri Jul 28 02:54:32 2006
@@ -23,7 +23,7 @@
<transportConnector uri="tcp://localhost:61616"/>
</transportConnectors>
<persistenceAdapter>
- <kahaPersistentAdaptor dir = "activemq-data"/>
+ <kahaPersistenceAdapter dir = "activemq-data"/>
</persistenceAdapter>
</broker>
Modified: incubator/activemq/trunk/activemq-perftest/src/main/resources/broker-conf/kaha.xml
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-perftest/src/main/resources/broker-conf/kaha.xml?rev=426461&r1=426460&r2=426461&view=diff
==============================================================================
--- incubator/activemq/trunk/activemq-perftest/src/main/resources/broker-conf/kaha.xml (original)
+++ incubator/activemq/trunk/activemq-perftest/src/main/resources/broker-conf/kaha.xml Fri Jul 28 02:54:32 2006
@@ -3,7 +3,7 @@
<broker useJmx="false" brokerName="kahaBroker" start="false" persistent="true" useShutdownHook="false" deleteAllMessagesOnStartup="true" advisorySupport="false">
<persistenceAdapter>
- <kahaPersistentAdaptor dir="target/kaha-data"/>
+ <kahaPersistenceAdapter dir="target/kaha-data"/>
</persistenceAdapter>
<transportConnectors>