You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by or...@apache.org on 2019/07/29 11:17:14 UTC
[qpid-broker-j] 01/02: QPID-8345: [Broker-J][AMQP 1.0] Dequeue
messages sent non-transactionally as pre-settled
This is an automated email from the ASF dual-hosted git repository.
orudyy pushed a commit to branch 7.0.x
in repository https://gitbox.apache.org/repos/asf/qpid-broker-j.git
commit 67a3a525a717835003e506105fb5d4763cda1ac5
Author: Alex Rudyy <or...@apache.org>
AuthorDate: Wed Jul 24 11:31:42 2019 +0100
QPID-8345: [Broker-J][AMQP 1.0] Dequeue messages sent non-transactionally as pre-settled
(cherry picked from commit ca0d0d7ec22c31364c1adc32f5dffee0392ff230)
---
.../server/protocol/v1_0/ConsumerTarget_1_0.java | 41 +++++++++++++++++++++-
.../protocol/v1_0/messaging/TransferTest.java | 8 ++++-
2 files changed, 47 insertions(+), 2 deletions(-)
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
index 0d19b9d..c7a2cd6 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
@@ -238,9 +238,14 @@ class ConsumerTarget_1_0 extends AbstractConsumerTarget<ConsumerTarget_1_0>
if (_linkEndpoint.isAttached())
{
- if (SenderSettleMode.SETTLED.equals(getEndpoint().getSendingSettlementMode()))
+ boolean sendPreSettled = SenderSettleMode.SETTLED.equals(getEndpoint().getSendingSettlementMode());
+ if (sendPreSettled)
{
transfer.setSettled(true);
+ if (_acquires && _transactionId == null)
+ {
+ transfer.setState(new Accepted());
+ }
}
else
{
@@ -295,6 +300,11 @@ class ConsumerTarget_1_0 extends AbstractConsumerTarget<ConsumerTarget_1_0>
}
getSession().getAMQPConnection().registerMessageDelivered(message.getSize());
getEndpoint().transfer(transfer, false);
+
+ if (sendPreSettled && _acquires && _transactionId == null)
+ {
+ handleAcquiredEntrySentPareSettledNonTransactional(entry, consumer);
+ }
}
else
{
@@ -312,6 +322,35 @@ class ConsumerTarget_1_0 extends AbstractConsumerTarget<ConsumerTarget_1_0>
}
}
+ private void handleAcquiredEntrySentPareSettledNonTransactional(final MessageInstance entry,
+ final MessageInstanceConsumer consumer)
+ {
+ if (entry.makeAcquisitionUnstealable(consumer))
+ {
+ final ServerTransaction txn = _linkEndpoint.getAsyncAutoCommitTransaction();
+ txn.dequeue(entry.getEnqueueRecord(),
+ new ServerTransaction.Action()
+ {
+ @Override
+ public void postCommit()
+ {
+ entry.delete();
+ }
+
+ @Override
+ public void onRollback()
+ {
+ entry.release(consumer);
+ }
+ });
+ txn.commit();
+ }
+ else
+ {
+ entry.release(consumer);
+ }
+ }
+
@Override
public void flushBatched()
{
diff --git a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/TransferTest.java b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/TransferTest.java
index 70c479b..bd16988 100644
--- a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/TransferTest.java
+++ b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/TransferTest.java
@@ -737,7 +737,6 @@ public class TransferTest extends BrokerAdminUsingTestBase
.begin().consumeResponse()
.attachRole(Role.RECEIVER)
.attachSourceAddress(BrokerAdmin.TEST_QUEUE_NAME)
- .attachRcvSettleMode(ReceiverSettleMode.FIRST)
.attachSndSettleMode(SenderSettleMode.SETTLED)
.attach().consumeResponse(Attach.class);
Attach attach = interaction.getLatestResponse(Attach.class);
@@ -760,6 +759,13 @@ public class TransferTest extends BrokerAdminUsingTestBase
// verify no unexpected performative received by closing the connection
interaction.doCloseConnection();
}
+
+ if (getBrokerAdmin().isQueueDepthSupported())
+ {
+ assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(0)));
+ }
+ getBrokerAdmin().putMessageOnQueue(BrokerAdmin.TEST_QUEUE_NAME, "test");
+ assertThat(Utils.receiveMessage(_brokerAddress, BrokerAdmin.TEST_QUEUE_NAME), is(equalTo("test")));
}
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org