activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From chir...@apache.org
Subject svn commit: r1488375 - in /activemq/trunk: activemq-broker/src/main/java/org/apache/activemq/broker/ activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/ activemq-broker/src/main/java/org/apache/activemq/broker/scheduler/ activemq-j...
Date Fri, 31 May 2013 19:56:03 GMT
Author: chirino
Date: Fri May 31 19:56:03 2013
New Revision: 1488375

URL: http://svn.apache.org/r1488375
Log:
Additional fixes related to AMQ-4563: You can now configure the storeOpenWireVersion property
of a broker to control which version of openwire is used by the persistence stores.  This
needs to be set to version 10 to preserve the original AMQP message ids.

Modified:
    activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
    activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
    activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/scheduler/SchedulerBroker.java
    activemq/trunk/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCPersistenceAdapter.java
    activemq/trunk/activemq-jdbc-store/src/main/java/org/apache/activemq/store/journal/JournalPersistenceAdapter.java
    activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
    activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBTransactionStore.java
    activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/TempKahaDBStore.java
    activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala

Modified: activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java?rev=1488375&r1=1488374&r2=1488375&view=diff
==============================================================================
--- activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
(original)
+++ activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/BrokerService.java
Fri May 31 19:56:03 2013
@@ -92,6 +92,7 @@ import org.apache.activemq.network.Conne
 import org.apache.activemq.network.DiscoveryNetworkConnector;
 import org.apache.activemq.network.NetworkConnector;
 import org.apache.activemq.network.jms.JmsConnector;
+import org.apache.activemq.openwire.OpenWireFormat;
 import org.apache.activemq.proxy.ProxyConnector;
 import org.apache.activemq.security.MessageAuthorizationPolicy;
 import org.apache.activemq.selector.SelectorParser;
@@ -239,6 +240,8 @@ public class BrokerService implements Se
     private boolean restartAllowed = true;
     private boolean restartRequested = false;
 
+    private int storeOpenWireVersion = OpenWireFormat.DEFAULT_VERSION;
+
     static {
 
         try {
@@ -2880,4 +2883,12 @@ public class BrokerService implements Se
     public void requestRestart() {
         this.restartRequested = true;
     }
+
+    public int getStoreOpenWireVersion() {
+        return storeOpenWireVersion;
+    }
+
+    public void setStoreOpenWireVersion(int storeOpenWireVersion) {
+        this.storeOpenWireVersion = storeOpenWireVersion;
+    }
 }

Modified: activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java?rev=1488375&r1=1488374&r2=1488375&view=diff
==============================================================================
--- activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
(original)
+++ activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/region/cursors/FilePendingMessageCursor.java
Fri May 31 19:56:03 2013
@@ -83,6 +83,9 @@ public class FilePendingMessageCursor ex
     @Override
     public void start() throws Exception {
         if (started.compareAndSet(false, true)) {
+            if( this.broker != null) {
+                wireFormat.setVersion(this.broker.getBrokerService().getStoreOpenWireVersion());
+            }
             super.start();
             if (systemUsage != null) {
                 systemUsage.getMemoryUsage().addUsageListener(this);

Modified: activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/scheduler/SchedulerBroker.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/scheduler/SchedulerBroker.java?rev=1488375&r1=1488374&r2=1488375&view=diff
==============================================================================
--- activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/scheduler/SchedulerBroker.java
(original)
+++ activemq/trunk/activemq-broker/src/main/java/org/apache/activemq/broker/scheduler/SchedulerBroker.java
Fri May 31 19:56:03 2013
@@ -65,6 +65,8 @@ public class SchedulerBroker extends Bro
         this.context.setSecurityContext(SecurityContext.BROKER_SECURITY_CONTEXT);
         this.context.setBroker(next);
         this.systemUsage = brokerService.getSystemUsage();
+
+        wireFormat.setVersion(brokerService.getStoreOpenWireVersion());
     }
 
     public synchronized JobScheduler getJobScheduler() throws Exception {

Modified: activemq/trunk/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCPersistenceAdapter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCPersistenceAdapter.java?rev=1488375&r1=1488374&r2=1488375&view=diff
==============================================================================
--- activemq/trunk/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCPersistenceAdapter.java
(original)
+++ activemq/trunk/activemq-jdbc-store/src/main/java/org/apache/activemq/store/jdbc/JDBCPersistenceAdapter.java
Fri May 31 19:56:03 2013
@@ -304,6 +304,11 @@ public class JDBCPersistenceAdapter exte
     }
 
     public void doStart() throws Exception {
+
+        if( brokerService!=null ) {
+          wireFormat.setVersion(brokerService.getStoreOpenWireVersion());
+        }
+
         // Cleanup the db periodically.
         if (cleanupPeriod > 0) {
             cleanupTicket = getScheduledThreadPoolExecutor().scheduleWithFixedDelay(new Runnable()
{

Modified: activemq/trunk/activemq-jdbc-store/src/main/java/org/apache/activemq/store/journal/JournalPersistenceAdapter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-jdbc-store/src/main/java/org/apache/activemq/store/journal/JournalPersistenceAdapter.java?rev=1488375&r1=1488374&r2=1488375&view=diff
==============================================================================
--- activemq/trunk/activemq-jdbc-store/src/main/java/org/apache/activemq/store/journal/JournalPersistenceAdapter.java
(original)
+++ activemq/trunk/activemq-jdbc-store/src/main/java/org/apache/activemq/store/journal/JournalPersistenceAdapter.java
Fri May 31 19:56:03 2013
@@ -241,6 +241,10 @@ public class JournalPersistenceAdapter i
             return;
         }
 
+        if( brokerService!=null ) {
+          wireFormat.setVersion(brokerService.getStoreOpenWireVersion());
+        }
+
         checkpointTask = taskRunnerFactory.createTaskRunner(new Task() {
             public boolean iterate() {
                 return doCheckpoint();

Modified: activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java?rev=1488375&r1=1488374&r2=1488375&view=diff
==============================================================================
--- activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
(original)
+++ activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
Fri May 31 19:56:03 2013
@@ -177,6 +177,9 @@ public class KahaDBStore extends Message
 
     @Override
     public void doStart() throws Exception {
+        if( brokerService!=null ) {
+            wireFormat.setVersion(brokerService.getStoreOpenWireVersion());
+        }
         super.doStart();
         this.globalQueueSemaphore = new Semaphore(getMaxAsyncJobs());
         this.globalTopicSemaphore = new Semaphore(getMaxAsyncJobs());

Modified: activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBTransactionStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBTransactionStore.java?rev=1488375&r1=1488374&r2=1488375&view=diff
==============================================================================
--- activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBTransactionStore.java
(original)
+++ activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBTransactionStore.java
Fri May 31 19:56:03 2013
@@ -62,13 +62,16 @@ import org.slf4j.LoggerFactory;
 public class KahaDBTransactionStore implements TransactionStore {
     static final Logger LOG = LoggerFactory.getLogger(KahaDBTransactionStore.class);
     ConcurrentHashMap<Object, Tx> inflightTransactions = new ConcurrentHashMap<Object,
Tx>();
-    private final WireFormat wireFormat = new OpenWireFormat();
     private final KahaDBStore theStore;
 
     public KahaDBTransactionStore(KahaDBStore theStore) {
         this.theStore = theStore;
     }
 
+    private WireFormat wireFormat(){
+      return this.theStore.wireFormat;
+    }
+
     public class Tx {
         private final ArrayList<AddMessageCommand> messages = new ArrayList<AddMessageCommand>();
 
@@ -335,13 +338,13 @@ public class KahaDBTransactionStore impl
             for (Operation op : entry.getValue()) {
                 if (op.getClass() == AddOpperation.class) {
                     AddOpperation addOp = (AddOpperation) op;
-                    Message msg = (Message) wireFormat.unmarshal(new DataInputStream(addOp.getCommand().getMessage()
+                    Message msg = (Message) wireFormat().unmarshal(new DataInputStream(addOp.getCommand().getMessage()
                             .newInput()));
                     messageList.add(msg);
                 } else {
                     RemoveOpperation rmOp = (RemoveOpperation) op;
                     Buffer ackb = rmOp.getCommand().getAck();
-                    MessageAck ack = (MessageAck) wireFormat.unmarshal(new DataInputStream(ackb.newInput()));
+                    MessageAck ack = (MessageAck) wireFormat().unmarshal(new DataInputStream(ackb.newInput()));
                     ackList.add(ack);
                 }
             }

Modified: activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/TempKahaDBStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/TempKahaDBStore.java?rev=1488375&r1=1488374&r2=1488375&view=diff
==============================================================================
--- activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/TempKahaDBStore.java
(original)
+++ activemq/trunk/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/TempKahaDBStore.java
Fri May 31 19:56:03 2013
@@ -24,6 +24,9 @@ import java.util.Iterator;
 import java.util.Map;
 import java.util.Set;
 import java.util.Map.Entry;
+
+import org.apache.activemq.broker.BrokerService;
+import org.apache.activemq.broker.BrokerServiceAware;
 import org.apache.activemq.broker.ConnectionContext;
 import org.apache.activemq.command.ActiveMQDestination;
 import org.apache.activemq.command.ActiveMQQueue;
@@ -60,9 +63,10 @@ import org.apache.activemq.wireformat.Wi
 import org.apache.activemq.store.kahadb.disk.journal.Location;
 import org.apache.activemq.store.kahadb.disk.page.Transaction;
 
-public class TempKahaDBStore extends TempMessageDatabase implements PersistenceAdapter {
+public class TempKahaDBStore extends TempMessageDatabase implements PersistenceAdapter, BrokerServiceAware
{
 
     private final WireFormat wireFormat = new OpenWireFormat();
+    private BrokerService brokerService;
 
     public void setBrokerName(String brokerName) {
     }
@@ -575,5 +579,17 @@ public class TempKahaDBStore extends Tem
     public long getLastProducerSequenceId(ProducerId id) {
         return -1;
     }
-        
+
+    @Override
+    public void setBrokerService(BrokerService brokerService) {
+        this.brokerService = brokerService;
+    }
+
+    @Override
+    public void load() throws IOException {
+        if( brokerService!=null ) {
+            wireFormat.setVersion(brokerService.getStoreOpenWireVersion());
+        }
+        super.load();
+    }
 }

Modified: activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala?rev=1488375&r1=1488374&r2=1488375&view=diff
==============================================================================
--- activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
(original)
+++ activemq/trunk/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala
Fri May 31 19:56:03 2013
@@ -206,7 +206,9 @@ class LevelDBStore extends LockableServi
   var snappyCompressLogs = false
 
   def doStart: Unit = {
-
+    if( brokerService!=null ) {
+      wireFormat.setVersion(brokerService.getStoreOpenWireVersion)
+    }
     snappyCompressLogs = logCompression.toLowerCase == "snappy" && Snappy != null
     debug("starting")
 



Mime
View raw message