Return-Path: X-Original-To: apmail-activemq-commits-archive@www.apache.org Delivered-To: apmail-activemq-commits-archive@www.apache.org Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by minotaur.apache.org (Postfix) with SMTP id A26C118738 for ; Mon, 18 Apr 2016 12:40:44 +0000 (UTC) Received: (qmail 54853 invoked by uid 500); 18 Apr 2016 12:40:44 -0000 Delivered-To: apmail-activemq-commits-archive@activemq.apache.org Received: (qmail 54813 invoked by uid 500); 18 Apr 2016 12:40:44 -0000 Mailing-List: contact commits-help@activemq.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: dev@activemq.apache.org Delivered-To: mailing list commits@activemq.apache.org Received: (qmail 54804 invoked by uid 99); 18 Apr 2016 12:40:44 -0000 Received: from git1-us-west.apache.org (HELO git1-us-west.apache.org) (140.211.11.23) by apache.org (qpsmtpd/0.29) with ESMTP; Mon, 18 Apr 2016 12:40:44 +0000 Received: by git1-us-west.apache.org (ASF Mail Server at git1-us-west.apache.org, from userid 33) id 2E621DFB73; Mon, 18 Apr 2016 12:40:44 +0000 (UTC) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: cshannon@apache.org To: commits@activemq.apache.org Message-Id: X-Mailer: ASF-Git Admin Mailer Subject: activemq git commit: https://issues.apache.org/jira/browse/AMQ-6256 Date: Mon, 18 Apr 2016 12:40:44 +0000 (UTC) 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) Authored: Mon Apr 18 12:34:25 2016 +0000 Committer: Christopher L. Shannon (cshannon) 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 asyncAddQueueMessage(final ConnectionContext context, final Message message) throws IOException { if (isConcurrentStoreAndDispatchQueues()) { + message.beforeMarshall(wireFormat); StoreQueueTask result = new StoreQueueTask(this, context, message); ListenableFuture future = result.getFuture(); message.getMessageId().setFutureOrSequenceLong(future); @@ -753,6 +754,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter { public ListenableFuture 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()