You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2017/03/04 16:47:13 UTC
[1/3] activemq-artemis git commit: ARTEMIS-1013 Queue deliver after
AMQP msg release [Forced Update!]
Repository: activemq-artemis
Updated Branches:
refs/heads/artemis-1009 aea122292 -> 80f6ae6ba (forced update)
ARTEMIS-1013 Queue deliver after AMQP msg release
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/15127aa5
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/15127aa5
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/15127aa5
Branch: refs/heads/artemis-1009
Commit: 15127aa5b119c3c4394951ba9ad9eb7f6add4aae
Parents: 8e9a83d
Author: Martyn Taylor <mt...@redhat.com>
Authored: Thu Mar 2 14:50:56 2017 +0000
Committer: Clebert Suconic <cl...@apache.org>
Committed: Sat Mar 4 11:45:57 2017 -0500
----------------------------------------------------------------------
.../amqp/broker/AMQPSessionCallback.java | 2 +-
.../amqp/AmqpReceiverDispositionTest.java | 9 ++++--
.../tests/integration/amqp/ProtonTest.java | 33 ++++++++++++++++++--
3 files changed, 37 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/15127aa5/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
index 5931afe..15816bb 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
@@ -327,6 +327,7 @@ public class AMQPSessionCallback implements SessionCallback {
recoverContext();
try {
((ServerConsumer) brokerConsumer).individualCancel(message.getMessageID(), updateCounts);
+ ((ServerConsumer) brokerConsumer).getQueue().forceDelivery();
} finally {
resetContext();
}
@@ -560,7 +561,6 @@ public class AMQPSessionCallback implements SessionCallback {
Transaction tx = protonSPI.getTransaction(txid);
tx.rollback();
protonSPI.removeTransaction(txid);
-
}
public SimpleString getMatchingQueue(SimpleString address, RoutingType routingType) throws Exception {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/15127aa5/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverDispositionTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverDispositionTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverDispositionTest.java
index d92fa0f..f206654 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverDispositionTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpReceiverDispositionTest.java
@@ -44,6 +44,10 @@ public class AmqpReceiverDispositionTest extends AmqpClientTestSupport {
receiver1.flow(1);
AmqpMessage message = receiver1.receive(5, TimeUnit.SECONDS);
+
+ AmqpReceiver receiver2 = session.createReceiver(getTestName());
+
+
assertNotNull("did not receive message first time", message);
assertEquals("MessageID:0", message.getMessageId());
@@ -51,12 +55,11 @@ public class AmqpReceiverDispositionTest extends AmqpClientTestSupport {
assertNotNull(protonMessage);
assertEquals("Unexpected initial value for AMQP delivery-count", 0, protonMessage.getDeliveryCount());
+ receiver2.flow(1);
message.release();
- // Read the message again and validate its state
- AmqpReceiver receiver2 = session.createReceiver(getTestName());
- receiver2.flow(1);
+ // Read the message again and validate its state
message = receiver2.receive(10, TimeUnit.SECONDS);
assertNotNull("did not receive message again", message);
assertEquals("MessageID:0", message.getMessageId());
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/15127aa5/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/ProtonTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/ProtonTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/ProtonTest.java
index 4640c33..16f2e70 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/ProtonTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/ProtonTest.java
@@ -110,7 +110,6 @@ public class ProtonTest extends ProtonTestBase {
private static final String amqpConnectionUri = "amqp://localhost:5672";
private static final String tcpAmqpConnectionUri = "tcp://localhost:5672";
-
private static final String brokerName = "my-broker";
private static final long maxSizeBytes = 1 * 1024 * 1024;
@@ -472,7 +471,7 @@ public class ProtonTest extends ProtonTestBase {
session.close();
Queue q = (Queue) server.getPostOffice().getBinding(new SimpleString(coreAddress)).getBindable();
//because tx commit is executed async on broker, we use a timed wait.
- assertTrue(TimeUtils.waitOnBoolean(true, 10000, ()-> q.getMessageCount() == 10));
+ assertTrue(TimeUtils.waitOnBoolean(true, 10000, () -> q.getMessageCount() == 10));
}
@Test
@@ -548,7 +547,7 @@ public class ProtonTest extends ProtonTestBase {
session.rollback();
Queue q = (Queue) server.getPostOffice().getBinding(new SimpleString(coreAddress)).getBindable();
//because tx rollback is executed async on broker, we use a timed wait.
- assertTrue(TimeUtils.waitOnBoolean(true, 10000, ()-> q.getMessageCount() == 10));
+ assertTrue(TimeUtils.waitOnBoolean(true, 10000, () -> q.getMessageCount() == 10));
}
@@ -1855,4 +1854,32 @@ public class ProtonTest extends ProtonTestBase {
return count;
}
}
+
+ @Test
+ public void testReleaseDisposition() throws Exception {
+ AmqpClient client = new AmqpClient(new URI(tcpAmqpConnectionUri), userName, password);
+ AmqpConnection connection = client.connect();
+ try {
+ AmqpSession session = connection.createSession();
+
+ AmqpSender sender = session.createSender(address);
+ AmqpMessage message = new AmqpMessage();
+ message.setText("Test-Message");
+ sender.send(message);
+
+ AmqpReceiver receiver = session.createReceiver(address);
+ receiver.flow(10);
+
+ AmqpMessage m1 = receiver.receive(5, TimeUnit.SECONDS);
+ assertNotNull(m1);
+ m1.release();
+
+ //receiver.flow(10);
+ AmqpMessage m2 = receiver.receive(5, TimeUnit.SECONDS);
+ assertNotNull(m2);
+ m2.accept();
+ } finally {
+ connection.close();
+ }
+ }
}
[2/3] activemq-artemis git commit: ARTEMIS-1012 Fix AMQP Transaction
Retirement
Posted by cl...@apache.org.
ARTEMIS-1012 Fix AMQP Transaction Retirement
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/5a9cdadb
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/5a9cdadb
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/5a9cdadb
Branch: refs/heads/artemis-1009
Commit: 5a9cdadb1d667fbd706e68c77445043ef187de11
Parents: 15127aa
Author: Martyn Taylor <mt...@redhat.com>
Authored: Thu Mar 2 21:06:41 2017 +0000
Committer: Clebert Suconic <cl...@apache.org>
Committed: Sat Mar 4 11:45:58 2017 -0500
----------------------------------------------------------------------
.../amqp/broker/AMQPConnectionCallback.java | 4 +-
.../amqp/broker/AMQPSessionCallback.java | 7 +-
.../amqp/proton/AMQPSessionContext.java | 1 +
.../proton/ProtonServerReceiverContext.java | 6 +-
.../amqp/proton/ProtonServerSenderContext.java | 33 +++--
.../amqp/proton/ProtonTransactionHandler.java | 134 ------------------
.../transaction/ProtonTransactionHandler.java | 135 +++++++++++++++++++
.../transaction/ProtonTransactionImpl.java | 91 +++++++++++++
.../ProtonTransactionRefsOperation.java | 62 +++++++++
.../artemis/core/server/impl/RefsOperation.java | 28 ++--
.../integration/amqp/AmqpClientTestSupport.java | 15 +++
.../integration/amqp/AmqpTransactionTest.java | 14 +-
12 files changed, 354 insertions(+), 176 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a9cdadb/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallback.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallback.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallback.java
index 9021272..850671a 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallback.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPConnectionCallback.java
@@ -35,12 +35,12 @@ import org.apache.activemq.artemis.core.remoting.FailureListener;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.transaction.Transaction;
-import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
import org.apache.activemq.artemis.core.transaction.impl.XidImpl;
import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPException;
import org.apache.activemq.artemis.protocol.amqp.logger.ActiveMQAMQPProtocolMessageBundle;
import org.apache.activemq.artemis.protocol.amqp.proton.AMQPConnectionContext;
import org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport;
+import org.apache.activemq.artemis.protocol.amqp.proton.transaction.ProtonTransactionImpl;
import org.apache.activemq.artemis.protocol.amqp.proton.handler.ExtCapability;
import org.apache.activemq.artemis.protocol.amqp.sasl.AnonymousServerSASL;
import org.apache.activemq.artemis.protocol.amqp.sasl.PlainSASL;
@@ -220,7 +220,7 @@ public class AMQPConnectionCallback implements FailureListener, CloseListener {
public Binary newTransaction() {
XidImpl xid = newXID();
- Transaction transaction = new TransactionImpl(xid, server.getStorageManager(), -1);
+ Transaction transaction = new ProtonTransactionImpl(xid, server.getStorageManager(), -1);
transactions.put(xid, transaction);
return new Binary(xid.getGlobalTransactionId());
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a9cdadb/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
index 15816bb..2ffcebf 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
@@ -49,6 +49,7 @@ import org.apache.activemq.artemis.protocol.amqp.proton.AMQPConnectionContext;
import org.apache.activemq.artemis.protocol.amqp.proton.AMQPSessionContext;
import org.apache.activemq.artemis.protocol.amqp.proton.AmqpSupport;
import org.apache.activemq.artemis.protocol.amqp.proton.ProtonServerSenderContext;
+import org.apache.activemq.artemis.protocol.amqp.proton.transaction.ProtonTransactionImpl;
import org.apache.activemq.artemis.protocol.amqp.sasl.PlainSASLResult;
import org.apache.activemq.artemis.protocol.amqp.sasl.SASLResult;
import org.apache.activemq.artemis.spi.core.protocol.SessionCallback;
@@ -487,7 +488,7 @@ public class AMQPSessionCallback implements SessionCallback {
ProtonServerSenderContext plugSender = (ProtonServerSenderContext) consumer.getProtocolContext();
try {
- return plugSender.deliverMessage(CoreAmqpConverter.checkAMQP(message), deliveryCount);
+ return plugSender.deliverMessage(ref, deliveryCount);
} catch (Exception e) {
synchronized (connection.getLock()) {
plugSender.getSender().setCondition(new ErrorCondition(AmqpError.INTERNAL_ERROR, e.getMessage()));
@@ -563,6 +564,10 @@ public class AMQPSessionCallback implements SessionCallback {
protonSPI.removeTransaction(txid);
}
+ public void dischargeTx(Binary txid) throws ActiveMQAMQPException {
+ ((ProtonTransactionImpl) protonSPI.getTransaction(txid)).discharge();
+ }
+
public SimpleString getMatchingQueue(SimpleString address, RoutingType routingType) throws Exception {
return serverSession.getMatchingQueue(address, routingType);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a9cdadb/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPSessionContext.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPSessionContext.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPSessionContext.java
index bd7b979..89b6ed3 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPSessionContext.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPSessionContext.java
@@ -25,6 +25,7 @@ import org.apache.activemq.artemis.protocol.amqp.broker.AMQPSessionCallback;
import org.apache.activemq.artemis.protocol.amqp.client.ProtonClientSenderContext;
import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPException;
import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPInternalErrorException;
+import org.apache.activemq.artemis.protocol.amqp.proton.transaction.ProtonTransactionHandler;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.transaction.Coordinator;
import org.apache.qpid.proton.amqp.transport.ErrorCondition;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a9cdadb/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerReceiverContext.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerReceiverContext.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerReceiverContext.java
index ea2635e..d5fc196 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerReceiverContext.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerReceiverContext.java
@@ -16,6 +16,9 @@
*/
package org.apache.activemq.artemis.protocol.amqp.proton;
+import java.util.Arrays;
+import java.util.List;
+
import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator;
import org.apache.activemq.artemis.api.core.RoutingType;
@@ -35,9 +38,6 @@ import org.apache.qpid.proton.engine.Delivery;
import org.apache.qpid.proton.engine.Receiver;
import org.jboss.logging.Logger;
-import java.util.Arrays;
-import java.util.List;
-
public class ProtonServerReceiverContext extends ProtonInitializable implements ProtonDeliveryHandler {
private static final Logger log = Logger.getLogger(ProtonServerReceiverContext.class);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a9cdadb/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java
index 4e33c9b..fcbd47e 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java
@@ -30,17 +30,19 @@ import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.postoffice.impl.CompositeAddress;
import org.apache.activemq.artemis.core.server.AddressQueryResult;
+import org.apache.activemq.artemis.core.server.MessageReference;
import org.apache.activemq.artemis.core.server.QueueQueryResult;
-import org.apache.activemq.artemis.core.transaction.Transaction;
import org.apache.activemq.artemis.jms.client.ActiveMQConnection;
import org.apache.activemq.artemis.protocol.amqp.broker.AMQPMessage;
import org.apache.activemq.artemis.protocol.amqp.broker.AMQPSessionCallback;
+import org.apache.activemq.artemis.protocol.amqp.converter.CoreAmqpConverter;
import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPException;
import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPIllegalStateException;
import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPInternalErrorException;
import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPNotFoundException;
import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPResourceLimitExceededException;
import org.apache.activemq.artemis.protocol.amqp.logger.ActiveMQAMQPProtocolMessageBundle;
+import org.apache.activemq.artemis.protocol.amqp.proton.transaction.ProtonTransactionImpl;
import org.apache.activemq.artemis.protocol.amqp.util.CreditsSemaphore;
import org.apache.activemq.artemis.selector.filter.FilterException;
import org.apache.activemq.artemis.selector.impl.SelectorParser;
@@ -474,26 +476,29 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
if (closed) {
return;
}
- Message message = (Message)delivery.getContext();
+
+ Message message = ((MessageReference) delivery.getContext()).getMessage();
boolean preSettle = sender.getRemoteSenderSettleMode() == SenderSettleMode.SETTLED;
DeliveryState remoteState = delivery.getRemoteState();
+ boolean settleImmediate = true;
if (remoteState != null) {
// If we are transactional then we need ack if the msg has been accepted
if (remoteState instanceof TransactionalState) {
TransactionalState txState = (TransactionalState) remoteState;
- Transaction tx = this.sessionSPI.getTransaction(txState.getTxnId());
+ ProtonTransactionImpl tx = (ProtonTransactionImpl) this.sessionSPI.getTransaction(txState.getTxnId());
+
if (txState.getOutcome() != null) {
+ settleImmediate = false;
Outcome outcome = txState.getOutcome();
if (outcome instanceof Accepted) {
if (!delivery.remotelySettled()) {
TransactionalState txAccepted = new TransactionalState();
txAccepted.setOutcome(Accepted.getInstance());
txAccepted.setTxnId(txState.getTxnId());
-
delivery.disposition(txAccepted);
}
// we have to individual ack as we can't guarantee we will get the delivery
@@ -501,6 +506,7 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
// from dealer, a perf hit but a must
try {
sessionSPI.ack(tx, brokerConsumer, message);
+ tx.addDelivery(delivery, this);
} catch (Exception e) {
throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorAcknowledgingMessage(message.toString(), e.getMessage());
}
@@ -550,16 +556,19 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
protonSession.replaceTag(delivery.getTag());
}
- synchronized (connection.getLock()) {
- delivery.settle();
- sender.offer(1);
- }
+ if (settleImmediate) settle(delivery);
} else {
// todo not sure if we need to do anything here
}
}
+ public void settle(Delivery delivery) {
+ synchronized (connection.getLock()) {
+ delivery.settle();
+ }
+ }
+
public synchronized void checkState() {
sessionSPI.resumeDelivery(brokerConsumer);
}
@@ -567,7 +576,9 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
/**
* handle an out going message from ActiveMQ Artemis, send via the Proton Sender
*/
- public int deliverMessage(AMQPMessage message, int deliveryCount) throws Exception {
+ public int deliverMessage(MessageReference messageReference, int deliveryCount) throws Exception {
+ AMQPMessage message = CoreAmqpConverter.checkAMQP(messageReference.getMessage());
+
if (closed) {
return 0;
}
@@ -602,14 +613,14 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
final Delivery delivery;
delivery = sender.delivery(tag, 0, tag.length);
delivery.setMessageFormat((int) message.getMessageFormat());
- delivery.setContext(message);
+ delivery.setContext(messageReference);
// this will avoid a copy.. patch provided by Norman using buffer.array()
sender.send(nettyBuffer.array(), nettyBuffer.arrayOffset() + nettyBuffer.readerIndex(), nettyBuffer.readableBytes());
if (preSettle) {
// Presettled means the client implicitly accepts any delivery we send it.
- sessionSPI.ack(null, brokerConsumer, message);
+ sessionSPI.ack(null, brokerConsumer, messageReference.getMessage());
delivery.settle();
} else {
sender.advance();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a9cdadb/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonTransactionHandler.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonTransactionHandler.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonTransactionHandler.java
deleted file mode 100644
index 1afeba8..0000000
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonTransactionHandler.java
+++ /dev/null
@@ -1,134 +0,0 @@
-/*
- * 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.artemis.protocol.amqp.proton;
-
-import io.netty.buffer.ByteBuf;
-import io.netty.buffer.PooledByteBufAllocator;
-import org.apache.activemq.artemis.protocol.amqp.broker.AMQPSessionCallback;
-import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPException;
-import org.apache.activemq.artemis.protocol.amqp.logger.ActiveMQAMQPProtocolMessageBundle;
-import org.apache.activemq.artemis.protocol.amqp.util.DeliveryUtil;
-import org.apache.activemq.artemis.protocol.amqp.util.NettyWritable;
-import org.apache.qpid.proton.amqp.Binary;
-import org.apache.qpid.proton.amqp.Symbol;
-import org.apache.qpid.proton.amqp.messaging.Accepted;
-import org.apache.qpid.proton.amqp.messaging.AmqpValue;
-import org.apache.qpid.proton.amqp.messaging.Rejected;
-import org.apache.qpid.proton.amqp.transaction.Declare;
-import org.apache.qpid.proton.amqp.transaction.Declared;
-import org.apache.qpid.proton.amqp.transaction.Discharge;
-import org.apache.qpid.proton.amqp.transport.ErrorCondition;
-import org.apache.qpid.proton.engine.Delivery;
-import org.apache.qpid.proton.engine.Receiver;
-import org.apache.qpid.proton.message.impl.MessageImpl;
-import org.jboss.logging.Logger;
-
-/**
- * handles an amqp Coordinator to deal with transaction boundaries etc
- */
-public class ProtonTransactionHandler implements ProtonDeliveryHandler {
-
- private static final Logger log = Logger.getLogger(ProtonTransactionHandler.class);
-
- final AMQPSessionCallback sessionSPI;
-
- public ProtonTransactionHandler(AMQPSessionCallback sessionSPI) {
- this.sessionSPI = sessionSPI;
- }
-
- @Override
- public void onMessage(Delivery delivery) throws ActiveMQAMQPException {
- ByteBuf buffer = PooledByteBufAllocator.DEFAULT.heapBuffer(1024);
-
- final Receiver receiver;
- try {
- receiver = ((Receiver) delivery.getLink());
-
- if (!delivery.isReadable()) {
- return;
- }
-
- receiver.recv(new NettyWritable(buffer));
-
- receiver.advance();
-
- MessageImpl msg = DeliveryUtil.decodeMessageImpl(buffer);
-
- Object action = ((AmqpValue) msg.getBody()).getValue();
-
- if (action instanceof Declare) {
- Binary txID = sessionSPI.newTransaction();
- Declared declared = new Declared();
- declared.setTxnId(txID);
- delivery.disposition(declared);
- delivery.settle();
- } else if (action instanceof Discharge) {
- Discharge discharge = (Discharge) action;
-
- Binary txID = discharge.getTxnId();
- if (discharge.getFail()) {
- try {
- sessionSPI.rollbackTX(txID, true);
- delivery.disposition(new Accepted());
- } catch (Exception e) {
- throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorRollingbackCoordinator(e.getMessage());
- }
- } else {
- try {
- sessionSPI.commitTX(txID);
- delivery.disposition(new Accepted());
- } catch (ActiveMQAMQPException amqpE) {
- throw amqpE;
- } catch (Exception e) {
- throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorCommittingCoordinator(e.getMessage());
- }
- }
- }
- } catch (ActiveMQAMQPException amqpE) {
- delivery.disposition(createRejected(amqpE.getAmqpError(), amqpE.getMessage()));
- } catch (Exception e) {
- log.warn(e.getMessage(), e);
- delivery.disposition(createRejected(Symbol.getSymbol("failed"), e.getMessage()));
- } finally {
- delivery.settle();
- buffer.release();
- }
- }
-
- private Rejected createRejected(Symbol amqpError, String message) {
- Rejected rejected = new Rejected();
- ErrorCondition condition = new ErrorCondition();
- condition.setCondition(amqpError);
- condition.setDescription(message);
- rejected.setError(condition);
- return rejected;
- }
-
- @Override
- public void onFlow(int credits, boolean drain) {
- }
-
- @Override
- public void close(boolean linkRemoteClose) throws ActiveMQAMQPException {
- // no op
- }
-
- @Override
- public void close(ErrorCondition condition) throws ActiveMQAMQPException {
- // no op
- }
-}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a9cdadb/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionHandler.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionHandler.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionHandler.java
new file mode 100644
index 0000000..2cdb072
--- /dev/null
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionHandler.java
@@ -0,0 +1,135 @@
+/*
+ * 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.artemis.protocol.amqp.proton.transaction;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import org.apache.activemq.artemis.protocol.amqp.broker.AMQPSessionCallback;
+import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPException;
+import org.apache.activemq.artemis.protocol.amqp.logger.ActiveMQAMQPProtocolMessageBundle;
+import org.apache.activemq.artemis.protocol.amqp.proton.ProtonDeliveryHandler;
+import org.apache.activemq.artemis.protocol.amqp.util.DeliveryUtil;
+import org.apache.activemq.artemis.protocol.amqp.util.NettyWritable;
+import org.apache.qpid.proton.amqp.Binary;
+import org.apache.qpid.proton.amqp.Symbol;
+import org.apache.qpid.proton.amqp.messaging.Accepted;
+import org.apache.qpid.proton.amqp.messaging.AmqpValue;
+import org.apache.qpid.proton.amqp.messaging.Rejected;
+import org.apache.qpid.proton.amqp.transaction.Declare;
+import org.apache.qpid.proton.amqp.transaction.Declared;
+import org.apache.qpid.proton.amqp.transaction.Discharge;
+import org.apache.qpid.proton.amqp.transport.ErrorCondition;
+import org.apache.qpid.proton.engine.Delivery;
+import org.apache.qpid.proton.engine.Receiver;
+import org.apache.qpid.proton.message.impl.MessageImpl;
+import org.jboss.logging.Logger;
+
+/**
+ * handles an amqp Coordinator to deal with transaction boundaries etc
+ */
+public class ProtonTransactionHandler implements ProtonDeliveryHandler {
+
+ private static final Logger log = Logger.getLogger(ProtonTransactionHandler.class);
+
+ final AMQPSessionCallback sessionSPI;
+
+ public ProtonTransactionHandler(AMQPSessionCallback sessionSPI) {
+ this.sessionSPI = sessionSPI;
+ }
+
+ @Override
+ public void onMessage(Delivery delivery) throws ActiveMQAMQPException {
+ ByteBuf buffer = PooledByteBufAllocator.DEFAULT.heapBuffer(1024);
+
+ final Receiver receiver;
+ try {
+ receiver = ((Receiver) delivery.getLink());
+
+ if (!delivery.isReadable()) {
+ return;
+ }
+
+ receiver.recv(new NettyWritable(buffer));
+
+ receiver.advance();
+
+ MessageImpl msg = DeliveryUtil.decodeMessageImpl(buffer);
+
+ Object action = ((AmqpValue) msg.getBody()).getValue();
+
+ if (action instanceof Declare) {
+ Binary txID = sessionSPI.newTransaction();
+ Declared declared = new Declared();
+ declared.setTxnId(txID);
+ delivery.disposition(declared);
+ } else if (action instanceof Discharge) {
+ Discharge discharge = (Discharge) action;
+
+ Binary txID = discharge.getTxnId();
+ sessionSPI.dischargeTx(txID);
+ if (discharge.getFail()) {
+ try {
+ sessionSPI.rollbackTX(txID, true);
+ delivery.disposition(new Accepted());
+ } catch (Exception e) {
+ throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorRollingbackCoordinator(e.getMessage());
+ }
+ } else {
+ try {
+ sessionSPI.commitTX(txID);
+ delivery.disposition(new Accepted());
+ } catch (ActiveMQAMQPException amqpE) {
+ throw amqpE;
+ } catch (Exception e) {
+ throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorCommittingCoordinator(e.getMessage());
+ }
+ }
+ }
+ } catch (ActiveMQAMQPException amqpE) {
+ delivery.disposition(createRejected(amqpE.getAmqpError(), amqpE.getMessage()));
+ } catch (Exception e) {
+ log.warn(e.getMessage(), e);
+ delivery.disposition(createRejected(Symbol.getSymbol("failed"), e.getMessage()));
+ } finally {
+ delivery.settle();
+ buffer.release();
+ }
+ }
+
+ private Rejected createRejected(Symbol amqpError, String message) {
+ Rejected rejected = new Rejected();
+ ErrorCondition condition = new ErrorCondition();
+ condition.setCondition(amqpError);
+ condition.setDescription(message);
+ rejected.setError(condition);
+ return rejected;
+ }
+
+ @Override
+ public void onFlow(int credits, boolean drain) {
+ }
+
+ @Override
+ public void close(boolean linkRemoteClose) throws ActiveMQAMQPException {
+ // no op
+ }
+
+ @Override
+ public void close(ErrorCondition condition) throws ActiveMQAMQPException {
+ // no op
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a9cdadb/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionImpl.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionImpl.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionImpl.java
new file mode 100644
index 0000000..dbe2bec
--- /dev/null
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionImpl.java
@@ -0,0 +1,91 @@
+/*
+ * 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.artemis.protocol.amqp.proton.transaction;
+
+import javax.transaction.xa.Xid;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.activemq.artemis.api.core.Pair;
+import org.apache.activemq.artemis.core.persistence.StorageManager;
+import org.apache.activemq.artemis.core.server.MessageReference;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.server.impl.RefsOperation;
+import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
+import org.apache.activemq.artemis.protocol.amqp.proton.ProtonServerSenderContext;
+import org.apache.qpid.proton.engine.Delivery;
+
+
+/**
+ * AMQP Protocol has different TX Rollback behaviour for Acks depending on whether an AMQP delivery has been settled
+ * or not. This class extends the Core TransactionImpl used for normal TX behaviour. In the case where deliveries
+ * have been settled, normal Ack rollback is applied. For cases where deliveries are unsettled and rolled back,
+ * we increment the delivery count and return to the consumer.
+ */
+public class ProtonTransactionImpl extends TransactionImpl {
+
+ /* We need to track the Message reference against the AMQP objects, so we can check whether the corresponding
+ deliveries have been settled. We also need to ensure we are settling on the correct link. Hence why we keep a ref
+ to the ProtonServerSenderContext here.
+ */
+ private final Map<MessageReference, Pair<Delivery, ProtonServerSenderContext>> deliveries = new HashMap<>();
+
+ private boolean discharged;
+
+ public ProtonTransactionImpl(final Xid xid, final StorageManager storageManager, final int timeoutSeconds) {
+ super(xid, storageManager, timeoutSeconds);
+ }
+
+ @Override
+ public RefsOperation createRefsOperation(Queue queue) {
+ return new ProtonTransactionRefsOperation(queue, storageManager);
+ }
+
+ @Override
+ public void rollback() throws Exception {
+ super.rollback();
+ }
+
+ public void addDelivery(Delivery delivery, ProtonServerSenderContext context) {
+ deliveries.put(((MessageReference) delivery.getContext()), new Pair<>(delivery, context));
+ }
+
+ public Map<MessageReference, Pair<Delivery, ProtonServerSenderContext>> getDeliveries() {
+ return deliveries;
+ }
+
+ @Override
+ public void commit() throws Exception {
+ super.commit();
+
+ // Settle all unsettled deliveries if commit is successful
+ for (Pair<Delivery, ProtonServerSenderContext> p : deliveries.values()) {
+ if (!p.getA().isSettled()) p.getB().settle(p.getA());
+ }
+ }
+
+ public boolean isDischarged() {
+ return discharged;
+ }
+
+ public void discharge() {
+ discharged = true;
+ }
+}
+
+
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a9cdadb/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionRefsOperation.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionRefsOperation.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionRefsOperation.java
new file mode 100644
index 0000000..7b48ac0
--- /dev/null
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionRefsOperation.java
@@ -0,0 +1,62 @@
+/*
+ * 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.artemis.protocol.amqp.proton.transaction;
+
+import java.util.LinkedList;
+import java.util.Map;
+
+import org.apache.activemq.artemis.core.persistence.StorageManager;
+import org.apache.activemq.artemis.core.server.MessageReference;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.server.ServerConsumer;
+import org.apache.activemq.artemis.core.server.impl.QueueImpl;
+import org.apache.activemq.artemis.core.server.impl.RefsOperation;
+import org.apache.activemq.artemis.core.transaction.Transaction;
+import org.apache.qpid.proton.engine.Delivery;
+
+/**
+ * AMQP Protocol has different TX Rollback behaviour for Acks depending on whether an AMQP delivery has been settled
+ * or not. This class extends the Core RefsOperation used for normal acks. In the case where deliveries have been
+ * settled, normal Ack rollback is applied. For cases where deliveries are unsettled and rolled back, we increment
+ * the delivery count and return to the consumer.
+ */
+public class ProtonTransactionRefsOperation extends RefsOperation {
+
+ public ProtonTransactionRefsOperation(final Queue queue, StorageManager storageManager) {
+ super(queue, storageManager);
+ }
+
+ @Override
+ public void rollbackRedelivery(Transaction txn, MessageReference ref, long timeBase, Map<QueueImpl, LinkedList<MessageReference>> queueMap) throws Exception {
+ ProtonTransactionImpl tx = (ProtonTransactionImpl) txn;
+
+ if (tx.getDeliveries().containsKey(ref)) {
+ Delivery del = tx.getDeliveries().get(ref).getA();
+ ServerConsumer consumer = (ServerConsumer) tx.getDeliveries().get(ref).getB().getBrokerConsumer();
+ // Rollback normally if the delivery is not settled or a forced TX rollback is done (e.g. connection drop).
+ if (del.remotelySettled() || !tx.isDischarged()) {
+ super.rollbackRedelivery(tx, ref, timeBase, queueMap);
+ } else {
+ ref.incrementDeliveryCount();
+ consumer.backToDelivering(ref);
+ del.disposition(del.getLocalState() == null ? del.getDefaultDeliveryState() : del.getLocalState());
+ }
+ } else {
+ super.rollbackRedelivery(tx, ref, timeBase, queueMap);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a9cdadb/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java
index 0f3da07..c5935d7 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RefsOperation.java
@@ -89,18 +89,7 @@ public class RefsOperation extends TransactionOperationAbstract {
if (ref.isAlreadyAcked()) {
ackedRefs.add(ref);
}
- // if ignore redelivery check, we just perform redelivery straight
- if (ref.getQueue().checkRedelivery(ref, timeBase, ignoreRedeliveryCheck)) {
- LinkedList<MessageReference> toCancel = queueMap.get(ref.getQueue());
-
- if (toCancel == null) {
- toCancel = new LinkedList<>();
-
- queueMap.put((QueueImpl) ref.getQueue(), toCancel);
- }
-
- toCancel.addFirst(ref);
- }
+ rollbackRedelivery(tx, ref, timeBase, queueMap);
} catch (Exception e) {
ActiveMQServerLogger.LOGGER.errorCheckingDLQ(e);
}
@@ -145,6 +134,21 @@ public class RefsOperation extends TransactionOperationAbstract {
}
}
+ protected void rollbackRedelivery(Transaction tx, MessageReference ref, long timeBase, Map<QueueImpl, LinkedList<MessageReference>> queueMap) throws Exception {
+ // if ignore redelivery check, we just perform redelivery straight
+ if (ref.getQueue().checkRedelivery(ref, timeBase, ignoreRedeliveryCheck)) {
+ LinkedList<MessageReference> toCancel = queueMap.get(ref.getQueue());
+
+ if (toCancel == null) {
+ toCancel = new LinkedList<>();
+
+ queueMap.put((QueueImpl) ref.getQueue(), toCancel);
+ }
+
+ toCancel.addFirst(ref);
+ }
+ }
+
@Override
public void afterCommit(final Transaction tx) {
for (MessageReference ref : refsToAck) {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a9cdadb/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpClientTestSupport.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpClientTestSupport.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpClientTestSupport.java
index db2e831..d9b45d3 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpClientTestSupport.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpClientTestSupport.java
@@ -99,12 +99,23 @@ public class AmqpClientTestSupport extends ActiveMQTestBase {
ActiveMQServer server = createServer(true, true);
serverManager = new JMSServerManagerImpl(server);
Configuration serverConfig = server.getConfiguration();
+
+ // Address 1
CoreAddressConfiguration address = new CoreAddressConfiguration();
address.setName(getTestName()).getRoutingTypes().add(RoutingType.ANYCAST);
CoreQueueConfiguration queueConfig = new CoreQueueConfiguration();
queueConfig.setName(getTestName()).setAddress(getTestName()).setRoutingType(RoutingType.ANYCAST);
address.getQueueConfigurations().add(queueConfig);
serverConfig.addAddressConfiguration(address);
+
+ // Address 2
+ CoreAddressConfiguration address2 = new CoreAddressConfiguration();
+ address2.setName(getTestName2()).getRoutingTypes().add(RoutingType.ANYCAST);
+ CoreQueueConfiguration queueConfig2 = new CoreQueueConfiguration();
+ queueConfig2.setName(getTestName2()).setAddress(getTestName2()).setRoutingType(RoutingType.ANYCAST);
+ address2.getQueueConfigurations().add(queueConfig2);
+ serverConfig.addAddressConfiguration(address2);
+
serverConfig.getAddressesSettings().put("#", new AddressSettings().setAutoCreateQueues(true).setAutoCreateAddresses(true).setDeadLetterAddress(new SimpleString("ActiveMQ.DLQ")));
serverConfig.setSecurityEnabled(false);
Set<TransportConfiguration> acceptors = serverConfig.getAcceptorConfigurations();
@@ -127,6 +138,10 @@ public class AmqpClientTestSupport extends ActiveMQTestBase {
return getName();
}
+ public String getTestName2() {
+ return getName() + "2";
+ }
+
public AmqpClientTestSupport() {
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/5a9cdadb/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpTransactionTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpTransactionTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpTransactionTest.java
index 1708720..1b2a1b0 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpTransactionTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpTransactionTest.java
@@ -514,7 +514,7 @@ public class AmqpTransactionTest extends AmqpClientTestSupport {
// Rollback the other half the consumed messages
txnSession.begin();
for (int i = NUM_MESSAGES / 2; i < NUM_MESSAGES; ++i) {
- messages.get(i).accept(txnSession);
+ messages.get(i).accept(txnSession, false);
}
txnSession.rollback();
@@ -525,18 +525,6 @@ public class AmqpTransactionTest extends AmqpClientTestSupport {
message.release();
}
- // Commit the other half the consumed messages
- // This is a variation from the .NET client tests which doesn't settle the
- // messages in the TX until commit is called but on ActiveMQ they will be
- // redispatched regardless and not stay in the acquired state.
- txnSession.begin();
- for (int i = NUM_MESSAGES / 2; i < NUM_MESSAGES; ++i) {
- AmqpMessage message = receiver.receive(5, TimeUnit.SECONDS);
- assertNotNull(message);
- message.accept();
- }
- txnSession.commit();
-
// The final message should still be pending.
{
AmqpMessage message = receiver.receive(5, TimeUnit.SECONDS);
[3/3] activemq-artemis git commit: fixing tests
Posted by cl...@apache.org.
fixing tests
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/80f6ae6b
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/80f6ae6b
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/80f6ae6b
Branch: refs/heads/artemis-1009
Commit: 80f6ae6bacbf0bea94dadfe211bf3c7d6794fcde
Parents: 5a9cdad
Author: Clebert Suconic <cl...@apache.org>
Authored: Sat Mar 4 11:38:14 2017 -0500
Committer: Clebert Suconic <cl...@apache.org>
Committed: Sat Mar 4 11:45:58 2017 -0500
----------------------------------------------------------------------
.../activemq/artemis/utils/TypedProperties.java | 12 ++++
.../activemq/artemis/api/core/Message.java | 4 ++
.../artemis/core/message/impl/CoreMessage.java | 46 ++++++++++-----
.../protocol/amqp/broker/AMQPMessage.java | 34 ++++++++---
.../amqp/broker/AMQPSessionCallback.java | 1 -
.../amqp/converter/AmqpCoreConverter.java | 1 +
.../amqp/converter/CoreAmqpConverter.java | 1 +
.../core/protocol/openwire/OpenwireMessage.java | 10 ++++
.../core/postoffice/impl/PostOfficeImpl.java | 1 -
.../artemis/core/server/ServerConsumer.java | 1 -
.../core/server/impl/ServerSessionImpl.java | 3 +-
.../management/impl/ManagementServiceImpl.java | 1 +
.../impl/ScheduledDeliveryHandlerTest.java | 10 ++++
.../amqp/AmqpDescribedTypePayloadTest.java | 6 +-
.../tests/integration/amqp/ProtonTest.java | 2 +-
.../integration/client/AcknowledgeTest.java | 10 ++++
.../cluster/bridge/SimpleTransformer.java | 61 ++++++++++----------
17 files changed, 145 insertions(+), 59 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/TypedProperties.java
----------------------------------------------------------------------
diff --git a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/TypedProperties.java b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/TypedProperties.java
index a421484..fda135b 100644
--- a/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/TypedProperties.java
+++ b/artemis-commons/src/main/java/org/apache/activemq/artemis/utils/TypedProperties.java
@@ -50,10 +50,14 @@ import static org.apache.activemq.artemis.utils.DataConstants.STRING;
*/
public final class TypedProperties {
+ private static final SimpleString AMQ_PROPNAME = new SimpleString("_AMQ_");
+
private Map<SimpleString, PropertyValue> properties;
private volatile int size;
+ private boolean internalProperties;
+
public TypedProperties() {
}
@@ -77,6 +81,10 @@ public final class TypedProperties {
size = other.size;
}
+ public boolean hasInternalProperties() {
+ return internalProperties;
+ }
+
public void putBooleanProperty(final SimpleString key, final boolean value) {
checkCreateProperties();
doPutValue(key, new BooleanValue(value));
@@ -497,6 +505,10 @@ public final class TypedProperties {
}
private synchronized void doPutValue(final SimpleString key, final PropertyValue value) {
+ if (key.startsWith(AMQ_PROPNAME)) {
+ internalProperties = true;
+ }
+
PropertyValue oldValue = properties.put(key, value);
if (oldValue != null) {
size += value.encodeSize() - oldValue.encodeSize();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java
index b39b719..f4f0e84 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/api/core/Message.java
@@ -226,6 +226,10 @@ public interface Message {
* */
RefCountMessageListener getContext();
+ SimpleString getReplyTo();
+
+ Message setReplyTo(SimpleString address);
+
Message setContext(RefCountMessageListener context);
/** The buffer will belong to this message, until release is called. */
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/CoreMessage.java
----------------------------------------------------------------------
diff --git a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/CoreMessage.java b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/CoreMessage.java
index 51c3701..22f9aa3 100644
--- a/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/CoreMessage.java
+++ b/artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/CoreMessage.java
@@ -37,6 +37,7 @@ import org.apache.activemq.artemis.core.buffers.impl.ResetLimitWrappedActiveMQBu
import org.apache.activemq.artemis.core.message.LargeBodyEncoder;
import org.apache.activemq.artemis.core.persistence.Persister;
import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
+import org.apache.activemq.artemis.reader.MessageUtil;
import org.apache.activemq.artemis.utils.DataConstants;
import org.apache.activemq.artemis.utils.TypedProperties;
import org.apache.activemq.artemis.utils.UUID;
@@ -109,22 +110,24 @@ public class CoreMessage extends RefCountMessage implements ICoreMessage {
@Override
public void cleanupInternalProperties() {
- LinkedList<SimpleString> valuesToRemove = null;
-
- for (SimpleString name : getPropertyNames()) {
- // We use properties to establish routing context on clustering.
- // However if the client resends the message after receiving, it needs to be removed
- if ((name.startsWith(Message.HDR_ROUTE_TO_IDS) && !name.equals(Message.HDR_ROUTE_TO_IDS)) || (name.startsWith(Message.HDR_ROUTE_TO_ACK_IDS) && !name.equals(Message.HDR_ROUTE_TO_ACK_IDS))) {
- if (valuesToRemove == null) {
- valuesToRemove = new LinkedList<>();
+ if (properties.hasInternalProperties()) {
+ LinkedList<SimpleString> valuesToRemove = null;
+
+ for (SimpleString name : getPropertyNames()) {
+ // We use properties to establish routing context on clustering.
+ // However if the client resends the message after receiving, it needs to be removed
+ if ((name.startsWith(Message.HDR_ROUTE_TO_IDS) && !name.equals(Message.HDR_ROUTE_TO_IDS)) || (name.startsWith(Message.HDR_ROUTE_TO_ACK_IDS) && !name.equals(Message.HDR_ROUTE_TO_ACK_IDS))) {
+ if (valuesToRemove == null) {
+ valuesToRemove = new LinkedList<>();
+ }
+ valuesToRemove.add(name);
}
- valuesToRemove.add(name);
}
- }
- if (valuesToRemove != null) {
- for (SimpleString removal : valuesToRemove) {
- this.removeProperty(removal);
+ if (valuesToRemove != null) {
+ for (SimpleString removal : valuesToRemove) {
+ this.removeProperty(removal);
+ }
}
}
}
@@ -152,6 +155,23 @@ public class CoreMessage extends RefCountMessage implements ICoreMessage {
}
@Override
+ public SimpleString getReplyTo() {
+ return getSimpleStringProperty(MessageUtil.REPLYTO_HEADER_NAME);
+ }
+
+ @Override
+ public CoreMessage setReplyTo(SimpleString address) {
+
+ if (address == null) {
+ checkProperties();
+ properties.removeProperty(MessageUtil.REPLYTO_HEADER_NAME);
+ } else {
+ putStringProperty(MessageUtil.REPLYTO_HEADER_NAME, address);
+ }
+ return this;
+ }
+
+ @Override
public void receiveBuffer(ByteBuf buffer) {
this.buffer = buffer;
this.buffer.retain();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java
index 83970e5..a533760 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPMessage.java
@@ -60,8 +60,8 @@ public class AMQPMessage extends RefCountMessage {
MessageImpl protonMessage;
private volatile int memoryEstimate = -1;
private long expiration = 0;
- // this can be used to encode the header again and the rest of the message buffer
- private int headerEnd = -1;
+ // this is to store where to start sending bytes, ignoring header and delivery annotations.
+ private int sendFrom = -1;
private boolean parsedHeaders = false;
private Header _header;
private DeliveryAnnotations _deliveryAnnotations;
@@ -111,7 +111,7 @@ public class AMQPMessage extends RefCountMessage {
private void initalizeObjects() {
if (protonMessage == null) {
if (data == null) {
- this.headerEnd = -1;
+ this.sendFrom = -1;
_header = new Header();
_deliveryAnnotations = new DeliveryAnnotations(new HashMap<>());
_properties = new Properties();
@@ -214,7 +214,7 @@ public class AMQPMessage extends RefCountMessage {
}
if (section instanceof Header) {
- headerEnd = buffer.position();
+ sendFrom = buffer.position();
_header = (Header) section;
if (_header.getTtl() != null) {
@@ -228,10 +228,11 @@ public class AMQPMessage extends RefCountMessage {
}
} else {
// meaning there is no header
- headerEnd = 0;
+ sendFrom = 0;
}
if (section instanceof DeliveryAnnotations) {
_deliveryAnnotations = (DeliveryAnnotations) section;
+ sendFrom = buffer.position();
if (buffer.hasRemaining()) {
section = (Section) decoder.readObject();
@@ -258,7 +259,6 @@ public class AMQPMessage extends RefCountMessage {
} else {
section = null;
}
-
}
if (section instanceof ApplicationProperties) {
@@ -450,7 +450,7 @@ public class AMQPMessage extends RefCountMessage {
TLSEncode.getEncoder().writeObject(header);
}
}
- buffer.writeBytes(data, headerEnd, data.writerIndex() - headerEnd);
+ buffer.writeBytes(data, sendFrom, data.writerIndex() - sendFrom);
}
@Override
@@ -770,6 +770,26 @@ public class AMQPMessage extends RefCountMessage {
}
}
+
+ @Override
+ public SimpleString getReplyTo() {
+ if (getProperties() != null) {
+ return SimpleString.toSimpleString(getProperties().getReplyTo());
+ } else {
+ return null;
+ }
+
+ }
+
+ @Override
+ public AMQPMessage setReplyTo(SimpleString address) {
+ if (getProperties() != null) {
+ getProperties().setReplyTo(address.toString());
+ }
+ return this;
+ }
+
+
@Override
public int getPersistSize() {
checkBuffer();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
index 2ffcebf..a079190 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
@@ -40,7 +40,6 @@ import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.core.server.impl.ServerConsumerImpl;
import org.apache.activemq.artemis.core.transaction.Transaction;
import org.apache.activemq.artemis.jms.client.ActiveMQConnection;
-import org.apache.activemq.artemis.protocol.amqp.converter.CoreAmqpConverter;
import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPException;
import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPInternalErrorException;
import org.apache.activemq.artemis.protocol.amqp.exceptions.ActiveMQAMQPResourceLimitExceededException;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/AmqpCoreConverter.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/AmqpCoreConverter.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/AmqpCoreConverter.java
index 656f2f2..030a7a0 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/AmqpCoreConverter.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/AmqpCoreConverter.java
@@ -171,6 +171,7 @@ public class AmqpCoreConverter {
}
populateMessage(result, message.getProtonMessage());
+ result.getInnerMessage().setReplyTo(message.getReplyTo());
result.encode();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/CoreAmqpConverter.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/CoreAmqpConverter.java b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/CoreAmqpConverter.java
index 5ba3371..111de8c 100644
--- a/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/CoreAmqpConverter.java
+++ b/artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/converter/CoreAmqpConverter.java
@@ -316,6 +316,7 @@ public class CoreAmqpConverter {
AMQPMessage amqpMessage = new AMQPMessage(messageFormat, data);
amqpMessage.setMessageID(message.getInnerMessage().getMessageID());
+ amqpMessage.setReplyTo(coreMessage.getReplyTo());
return amqpMessage;
} finally {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenwireMessage.java
----------------------------------------------------------------------
diff --git a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenwireMessage.java b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenwireMessage.java
index 005186d..6c86751 100644
--- a/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenwireMessage.java
+++ b/artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenwireMessage.java
@@ -42,6 +42,16 @@ public class OpenwireMessage implements Message {
}
@Override
+ public SimpleString getReplyTo() {
+ return null;
+ }
+
+ @Override
+ public Message setReplyTo(SimpleString address) {
+ return null;
+ }
+
+ @Override
public boolean containsDeliveryAnnotationProperty(SimpleString property) {
return false;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
index 9737dc0..464859f 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/postoffice/impl/PostOfficeImpl.java
@@ -22,7 +22,6 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
-import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerConsumer.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerConsumer.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerConsumer.java
index 0d57a78..ce9c489 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerConsumer.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ServerConsumer.java
@@ -18,7 +18,6 @@ package org.apache.activemq.artemis.core.server;
import java.util.List;
-import org.apache.activemq.artemis.api.core.ActiveMQException;
import org.apache.activemq.artemis.core.transaction.Transaction;
/**
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
index 6d95341..5361983 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerSessionImpl.java
@@ -42,7 +42,6 @@ import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.api.core.client.ClientSession;
import org.apache.activemq.artemis.api.core.management.CoreNotificationType;
import org.apache.activemq.artemis.api.core.management.ManagementHelper;
-import org.apache.activemq.artemis.core.client.impl.ClientMessageImpl;
import org.apache.activemq.artemis.core.exception.ActiveMQXAException;
import org.apache.activemq.artemis.core.filter.Filter;
import org.apache.activemq.artemis.core.filter.impl.FilterImpl;
@@ -1550,7 +1549,7 @@ public class ServerSessionImpl implements ServerSession, FailureListener {
Message reply = managementService.handleMessage(message);
- SimpleString replyTo = message.getSimpleStringProperty(ClientMessageImpl.REPLYTO_HEADER_NAME);
+ SimpleString replyTo = message.getReplyTo();
if (replyTo != null) {
// TODO: move this check somewhere else? this is a JMS-specific bit of logic in the core impl
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/impl/ManagementServiceImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/impl/ManagementServiceImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/impl/ManagementServiceImpl.java
index cda0a8a..f45aea7 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/impl/ManagementServiceImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/management/impl/ManagementServiceImpl.java
@@ -370,6 +370,7 @@ public class ManagementServiceImpl implements ManagementService {
message = message.toCore();
// a reply message is sent with the result stored in the message body.
CoreMessage reply = new CoreMessage(storageManager.generateID(), 512);
+ reply.setReplyTo(message.getReplyTo());
String resourceName = message.getStringProperty(ManagementHelper.HDR_RESOURCE_NAME);
if (logger.isDebugEnabled()) {
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
index 447fb08..faf8b12 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/impl/ScheduledDeliveryHandlerTest.java
@@ -284,6 +284,16 @@ public class ScheduledDeliveryHandlerTest extends Assert {
class FakeMessage extends RefCountMessage {
@Override
+ public SimpleString getReplyTo() {
+ return null;
+ }
+
+ @Override
+ public Message setReplyTo(SimpleString address) {
+ return null;
+ }
+
+ @Override
public boolean containsDeliveryAnnotationProperty(SimpleString property) {
return false;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDescribedTypePayloadTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDescribedTypePayloadTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDescribedTypePayloadTest.java
index bbb9c26..138f3cc 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDescribedTypePayloadTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/AmqpDescribedTypePayloadTest.java
@@ -24,6 +24,7 @@ import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
+import javax.jms.ObjectMessage;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
@@ -35,6 +36,7 @@ import org.apache.activemq.transport.amqp.client.AmqpNoLocalFilter;
import org.apache.activemq.transport.amqp.client.AmqpReceiver;
import org.apache.activemq.transport.amqp.client.AmqpSender;
import org.apache.activemq.transport.amqp.client.AmqpSession;
+import org.apache.qpid.jms.JmsConnectionFactory;
import org.junit.Test;
/**
@@ -119,7 +121,7 @@ public class AmqpDescribedTypePayloadTest extends AmqpClientTestSupport {
assertEquals(1, queue.getMessageCount());
// Receive and resend with OpenWire JMS client
- ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
+ JmsConnectionFactory factory = new JmsConnectionFactory("amqp://localhost:61616");
Connection jmsConnection = factory.createConnection();
try {
Session jmsSession = jmsConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
@@ -129,7 +131,7 @@ public class AmqpDescribedTypePayloadTest extends AmqpClientTestSupport {
Message received = jmsConsumer.receive(5000);
assertNotNull(received);
- assertTrue(received instanceof BytesMessage);
+ assertTrue(received instanceof ObjectMessage);
MessageProducer jmsProducer = jmsSession.createProducer(destination);
jmsProducer.send(received);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/ProtonTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/ProtonTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/ProtonTest.java
index 16f2e70..1308c37 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/ProtonTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/ProtonTest.java
@@ -926,7 +926,7 @@ public class ProtonTest extends ProtonTestBase {
request.setText("[]");
sender.send(request);
- AmqpMessage response = receiver.receive(50, TimeUnit.SECONDS);
+ AmqpMessage response = receiver.receive(5, TimeUnit.SECONDS);
Assert.assertNotNull(response);
assertNotNull(response);
Object section = response.getWrappedMessage().getBody();
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/AcknowledgeTest.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/AcknowledgeTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/AcknowledgeTest.java
index 887524a..08d9787 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/AcknowledgeTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/AcknowledgeTest.java
@@ -341,6 +341,16 @@ public class AcknowledgeTest extends ActiveMQTestBase {
final long id;
@Override
+ public SimpleString getReplyTo() {
+ return null;
+ }
+
+ @Override
+ public Message setReplyTo(SimpleString address) {
+ return null;
+ }
+
+ @Override
public boolean containsDeliveryAnnotationProperty(SimpleString property) {
return false;
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/80f6ae6b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/bridge/SimpleTransformer.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/bridge/SimpleTransformer.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/bridge/SimpleTransformer.java
index c0487d0..e462240 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/bridge/SimpleTransformer.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/bridge/SimpleTransformer.java
@@ -16,43 +16,42 @@
*/
package org.apache.activemq.artemis.tests.integration.cluster.bridge;
-
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.server.cluster.Transformer;
public class SimpleTransformer implements Transformer {
@Override
- public Message transform(final Message message) {
-
- // TODO-now: fix this test!!!
-
- throw new RuntimeException(("Fix me"));
-// SimpleString oldProp = (SimpleString) message.getObjectProperty(new SimpleString("wibble"));
-//
-// if (!oldProp.equals(new SimpleString("bing"))) {
-// throw new IllegalStateException("Wrong property value!!");
-// }
-//
-// // Change a property
-// message.putStringProperty(new SimpleString("wibble"), new SimpleString("bong"));
-//
-// // Change the body
-// ActiveMQBuffer buffer = message.getBodyBuffer();
-//
-// buffer.readerIndex(0);
-//
-// String str = buffer.readString();
-//
-// if (!str.equals("doo be doo be doo be doo")) {
-// throw new IllegalStateException("Wrong body!!");
-// }
-//
-// buffer.clear();
-//
-// buffer.writeString("dee be dee be dee be dee");
-//
-// return message;
+ public Message transform(final Message messageParameter) {
+ ICoreMessage message = messageParameter.toCore();
+ SimpleString oldProp = (SimpleString) message.getObjectProperty(new SimpleString("wibble"));
+
+ if (!oldProp.equals(new SimpleString("bing"))) {
+ throw new IllegalStateException("Wrong property value!!");
+ }
+
+ // Change a property
+ message.putStringProperty(new SimpleString("wibble"), new SimpleString("bong"));
+
+ // Change the body
+ ActiveMQBuffer buffer = message.getBodyBuffer();
+
+ buffer.readerIndex(0);
+
+ String str = buffer.readString();
+
+ if (!str.equals("doo be doo be doo be doo")) {
+ throw new IllegalStateException("Wrong body!!");
+ }
+
+ buffer.clear();
+
+ buffer.writeString("dee be dee be dee be dee");
+
+ return message;
}
}