Repository: activemq Updated Branches: refs/heads/master 9c8bd3360 -> 11622b3af
https://issues.apache.org/jira/browse/AMQ-6256 Moving beforeMarshall back to the store implementations because we don't want all store implementations to marshall (such as memory store) This reverts commit b9b98a45cee484b112dadeffa2d9a874c4ffe280. Project: http://git-wip-us.apache.org/repos/asf/activemq/repo Commit: http://git-wip-us.apache.org/repos/asf/activemq/commit/11622b3a Tree: http://git-wip-us.apache.org/repos/asf/activemq/tree/11622b3a Diff: http://git-wip-us.apache.org/repos/asf/activemq/diff/11622b3a Branch: refs/heads/master Commit: 11622b3af32fe9518df06f026e6c0fb65de4aa26 Parents: 9c8bd33 Author: Christopher L. Shannon (cshannon) <christopher.l.shan...@gmail.com> Authored: Mon Apr 18 12:34:25 2016 +0000 Committer: Christopher L. Shannon (cshannon) <christopher.l.shan...@gmail.com> Committed: Mon Apr 18 12:34:25 2016 +0000 ---------------------------------------------------------------------- .../src/main/java/org/apache/activemq/broker/region/Queue.java | 1 - .../src/main/java/org/apache/activemq/broker/region/Topic.java | 1 - .../main/java/org/apache/activemq/store/kahadb/KahaDBStore.java | 2 ++ .../src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala | 1 + 4 files changed, 3 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/activemq/blob/11622b3a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java ---------------------------------------------------------------------- diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java index a2f7184..097dfa5 100755 --- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java +++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Queue.java @@ -838,7 +838,6 @@ public class Queue extends BaseDestination implements Task, UsageListener, Index //condition if the original add is processed after the update, which can cause //a duplicate message to be stored if (messages.isCacheEnabled() && !isPersistJMSRedelivered()) { - message.beforeMarshall(null); result = store.asyncAddQueueMessage(context, message, isOptimizeStorage()); result.addListener(new PendingMarshalUsageTracker(message)); } else { http://git-wip-us.apache.org/repos/asf/activemq/blob/11622b3a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Topic.java ---------------------------------------------------------------------- diff --git a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Topic.java b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Topic.java index a13bcd5..1a9949e 100755 --- a/activemq-broker/src/main/java/org/apache/activemq/broker/region/Topic.java +++ b/activemq-broker/src/main/java/org/apache/activemq/broker/region/Topic.java @@ -511,7 +511,6 @@ public class Topic extends BaseDestination implements Task { waitForSpace(context,producerExchange, systemUsage.getStoreUsage(), getStoreUsageHighWaterMark(), logMessage); } - message.beforeMarshall(null); result = topicStore.asyncAddTopicMessage(context, message,isOptimizeStorage()); } http://git-wip-us.apache.org/repos/asf/activemq/blob/11622b3a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java ---------------------------------------------------------------------- diff --git a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java index 769e303..7f8283d 100644 --- a/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java +++ b/activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java @@ -383,6 +383,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter { public ListenableFuture<Object> asyncAddQueueMessage(final ConnectionContext context, final Message message) throws IOException { if (isConcurrentStoreAndDispatchQueues()) { + message.beforeMarshall(wireFormat); StoreQueueTask result = new StoreQueueTask(this, context, message); ListenableFuture<Object> future = result.getFuture(); message.getMessageId().setFutureOrSequenceLong(future); @@ -753,6 +754,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter { public ListenableFuture<Object> asyncAddTopicMessage(final ConnectionContext context, final Message message) throws IOException { if (isConcurrentStoreAndDispatchTopics()) { + message.beforeMarshall(wireFormat); StoreTopicTask result = new StoreTopicTask(this, context, message, subscriptionCount.get()); result.aquireLocks(); addTopicTask(this, result); http://git-wip-us.apache.org/repos/asf/activemq/blob/11622b3a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala ---------------------------------------------------------------------- diff --git a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala index f80e722..5865f35 100644 --- a/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala +++ b/activemq-leveldb-store/src/main/scala/org/apache/activemq/leveldb/LevelDBStore.scala @@ -754,6 +754,7 @@ class LevelDBStore extends LockableServiceSupport with BrokerServiceAware with P def doAdd(uow: DelayableUOW, context: ConnectionContext, message: Message, delay:Boolean): CountDownFuture[AnyRef] = { check_running + message.beforeMarshall(wireFormat); message.incrementReferenceCount() uow.addCompleteListener({ message.decrementReferenceCount()
