You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by jb...@apache.org on 2017/03/29 01:32:50 UTC
[2/4] activemq-artemis git commit: ARTEMIS-1080 Implementing
AMQP::reject
ARTEMIS-1080 Implementing AMQP::reject
Project: http://git-wip-us.apache.org/repos/asf/activemq-artemis/repo
Commit: http://git-wip-us.apache.org/repos/asf/activemq-artemis/commit/746220e1
Tree: http://git-wip-us.apache.org/repos/asf/activemq-artemis/tree/746220e1
Diff: http://git-wip-us.apache.org/repos/asf/activemq-artemis/diff/746220e1
Branch: refs/heads/master
Commit: 746220e11eaf19939b058ca19aeb6df80530c734
Parents: 13a272b
Author: Clebert Suconic <cl...@apache.org>
Authored: Tue Mar 28 17:12:02 2017 -0400
Committer: Justin Bertram <jb...@apache.org>
Committed: Tue Mar 28 20:32:28 2017 -0500
----------------------------------------------------------------------
.../protocol/amqp/broker/AMQPSessionCallback.java | 9 +++++++++
.../amqp/proton/ProtonServerSenderContext.java | 2 +-
.../artemis/core/server/ServerConsumer.java | 2 ++
.../core/server/impl/ServerConsumerImpl.java | 16 ++++++++++++++++
.../amqp/AmqpReceiverDispositionTest.java | 4 ++--
.../tests/integration/cli/DummyServerConsumer.java | 5 +++++
6 files changed, 35 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/746220e1/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 58134f5..58d51db 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
@@ -336,6 +336,15 @@ public class AMQPSessionCallback implements SessionCallback {
}
}
+ public void reject(Object brokerConsumer, Message message) throws Exception {
+ recoverContext();
+ try {
+ ((ServerConsumer) brokerConsumer).reject(message.getMessageID());
+ } finally {
+ resetContext();
+ }
+ }
+
public void resumeDelivery(Object consumer) {
((ServerConsumer) consumer).receiveCredits(-1);
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/746220e1/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 780ca4d..fb540a8 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
@@ -546,7 +546,7 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
}
} else if (remoteState instanceof Rejected) {
try {
- sessionSPI.cancel(brokerConsumer, message, true);
+ sessionSPI.reject(brokerConsumer, message);
} catch (Exception e) {
throw ActiveMQAMQPProtocolMessageBundle.BUNDLE.errorCancellingMessage(message.toString(), e.getMessage());
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/746220e1/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 ce9c489..37232bc 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
@@ -92,6 +92,8 @@ public interface ServerConsumer extends Consumer {
void individualAcknowledge(Transaction tx, long messageID) throws Exception;
+ void reject(final long messageID) throws Exception;
+
void individualCancel(final long messageID, boolean failed) throws Exception;
void forceDelivery(long sequence);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/746220e1/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
----------------------------------------------------------------------
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
index e812f2e..3552b93 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ServerConsumerImpl.java
@@ -910,6 +910,22 @@ public class ServerConsumerImpl implements ServerConsumer, ReadyListener {
ref.getQueue().cancel(ref, System.currentTimeMillis());
}
+
+ @Override
+ public synchronized void reject(final long messageID) throws Exception {
+ if (browseOnly) {
+ return;
+ }
+
+ MessageReference ref = removeReferenceByID(messageID);
+
+ if (ref == null) {
+ return; // nothing to be done
+ }
+
+ ref.getQueue().sendToDeadLetterAddress(null, ref);
+ }
+
@Override
public synchronized void backToDelivering(MessageReference reference) {
deliveringRefs.addFirst(reference);
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/746220e1/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 f206654..b347d37 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
@@ -97,13 +97,13 @@ public class AmqpReceiverDispositionTest extends AmqpClientTestSupport {
// Reject is a terminal outcome and should not be redelivered to the rejecting receiver
// or any other as it should move to the archived state.
receiver1.flow(1);
- message = receiver1.receive(1, TimeUnit.SECONDS);
+ message = receiver1.receiveNoWait();
assertNull("Should not receive message again", message);
// Attempt to Read the message again with another receiver to validate it is archived.
AmqpReceiver receiver2 = session.createReceiver(getTestName());
receiver2.flow(1);
- assertNull(receiver2.receive(3, TimeUnit.SECONDS));
+ assertNull(receiver2.receiveNoWait());
connection.close();
}
http://git-wip-us.apache.org/repos/asf/activemq-artemis/blob/746220e1/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cli/DummyServerConsumer.java
----------------------------------------------------------------------
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cli/DummyServerConsumer.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cli/DummyServerConsumer.java
index d79b444..78b3e09 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cli/DummyServerConsumer.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cli/DummyServerConsumer.java
@@ -123,6 +123,11 @@ public class DummyServerConsumer implements ServerConsumer {
}
@Override
+ public void reject(long messageID) throws Exception {
+
+ }
+
+ @Override
public void acknowledge(Transaction tx, long messageID) throws Exception {
}