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 2017/10/06 09:59:07 UTC
[1/2] qpid-broker-j git commit: QPID-7960: [Java Broker,
AMQP 1.0] Add support for undeliverable-here on Modified outcome
Repository: qpid-broker-j
Updated Branches:
refs/heads/master fcad8ea35 -> 37b1a71ae
QPID-7960: [Java Broker, AMQP 1.0] Add support for undeliverable-here on Modified outcome
Project: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/commit/ac6f6c30
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/ac6f6c30
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/ac6f6c30
Branch: refs/heads/master
Commit: ac6f6c30a7dd859ec7e38708e118f1d8dff71c7e
Parents: fcad8ea
Author: Alex Rudyy <or...@apache.org>
Authored: Fri Oct 6 10:55:39 2017 +0100
Committer: Alex Rudyy <or...@apache.org>
Committed: Fri Oct 6 10:55:58 2017 +0100
----------------------------------------------------------------------
.../protocol/v1_0/ConsumerTarget_1_0.java | 8 +-
.../qpid/tests/protocol/v1_0/Interaction.java | 14 +++
.../protocol/v1_0/messaging/OutcomeTest.java | 106 +++++++++++++++++++
3 files changed, 126 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/ac6f6c30/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
----------------------------------------------------------------------
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 eb6d512..094164c 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
@@ -521,9 +521,13 @@ class ConsumerTarget_1_0 extends AbstractConsumerTarget<ConsumerTarget_1_0>
@Override
public void postCommit()
{
- // TODO: add handling of undeliverable-here
+ Modified modifiedOutcome = (Modified) outcome;
+ if (Boolean.TRUE.equals(modifiedOutcome.getUndeliverableHere()))
+ {
+ _queueEntry.reject(getConsumer());
+ }
- if(Boolean.TRUE.equals(((Modified)outcome).getDeliveryFailed()))
+ if(Boolean.TRUE.equals(modifiedOutcome.getDeliveryFailed()))
{
incrementDeliveryCountOrRouteToAlternateOrDiscard();
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/ac6f6c30/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Interaction.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Interaction.java b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Interaction.java
index 6b59960..0b91273 100644
--- a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Interaction.java
+++ b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/Interaction.java
@@ -105,6 +105,7 @@ public class Interaction
private int _deliveryIdCounter;
private List<Transfer> _latestDelivery;
private Object _decodedLatestDelivery;
+ private UnsignedInteger _latestDeliveryId;
Interaction(final FrameTransport frameTransport)
{
@@ -604,6 +605,11 @@ public class Interaction
return this;
}
+ public Interaction flowNextIncomingIdFromLatestDelivery()
+ {
+ return flowNextIncomingId(_latestDeliveryId.add(UnsignedInteger.ONE));
+ }
+
public Interaction flowOutgoingWindow(final UnsignedInteger outgoingWindow)
{
_flow.setOutgoingWindow(outgoingWindow);
@@ -850,6 +856,12 @@ public class Interaction
return this;
}
+ public Interaction dispositionFirstFromLatestDelivery()
+ {
+ _disposition.setFirst(_latestDeliveryId);
+ return this;
+ }
+
public Interaction disposition() throws Exception
{
sendPerformativeAndChainFuture(copyDisposition(_disposition), _sessionChannel);
@@ -1064,6 +1076,7 @@ public class Interaction
{
sync();
_latestDelivery = receiveAllTransfers();
+ _latestDeliveryId = _latestDelivery.size() > 0 ? _latestDelivery.get(0).getDeliveryId() : null;
return this;
}
@@ -1109,4 +1122,5 @@ public class Interaction
{
return new InteractionTransactionalState(handle);
}
+
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/ac6f6c30/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/OutcomeTest.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/OutcomeTest.java b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/OutcomeTest.java
new file mode 100644
index 0000000..f081006
--- /dev/null
+++ b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/messaging/OutcomeTest.java
@@ -0,0 +1,106 @@
+/*
+ *
+ * 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.qpid.tests.protocol.v1_0.messaging;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+import java.net.InetSocketAddress;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Modified;
+import org.apache.qpid.server.protocol.v1_0.type.transport.Attach;
+import org.apache.qpid.server.protocol.v1_0.type.transport.Begin;
+import org.apache.qpid.server.protocol.v1_0.type.transport.Open;
+import org.apache.qpid.server.protocol.v1_0.type.transport.Role;
+import org.apache.qpid.tests.protocol.v1_0.FrameTransport;
+import org.apache.qpid.tests.protocol.v1_0.Interaction;
+import org.apache.qpid.tests.protocol.v1_0.SpecificationTest;
+import org.apache.qpid.tests.utils.BrokerAdmin;
+import org.apache.qpid.tests.utils.BrokerAdminUsingTestBase;
+
+public class OutcomeTest extends BrokerAdminUsingTestBase
+{
+ private InetSocketAddress _brokerAddress;
+
+ @Before
+ public void setUp()
+ {
+ getBrokerAdmin().createQueue(BrokerAdmin.TEST_QUEUE_NAME);
+ _brokerAddress = getBrokerAdmin().getBrokerAddress(BrokerAdmin.PortType.ANONYMOUS_AMQP);
+ }
+
+
+ @Test
+ @SpecificationTest(section = "3.4.5", description = "If the undeliverable-here is set, then any messages released"
+ + " MUST NOT be redelivered to the modifying link endpoint.")
+ public void modifiedOutcomeWithUndeliverableHere() throws Exception
+ {
+ getBrokerAdmin().putMessageOnQueue(BrokerAdmin.TEST_QUEUE_NAME, "message1");
+ getBrokerAdmin().putMessageOnQueue(BrokerAdmin.TEST_QUEUE_NAME, "message2");
+
+ try (FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction()
+ .negotiateProtocol().consumeResponse()
+ .open().consumeResponse(Open.class)
+ .begin().consumeResponse(Begin.class)
+ .attachRole(Role.RECEIVER)
+ .attachSourceAddress(BrokerAdmin.TEST_QUEUE_NAME)
+ .attach().consumeResponse(Attach.class)
+ .flowIncomingWindow(UnsignedInteger.ONE)
+ .flowLinkCredit(UnsignedInteger.ONE)
+ .flowHandleFromLinkHandle()
+ .flow()
+ .receiveDelivery()
+
+ .decodeLatestDelivery();
+
+ Object firstDeliveryPayload = interaction.getDecodedLatestDelivery();
+ assertThat(firstDeliveryPayload, is(equalTo("message1")));
+
+ Modified modifiedOutcome = new Modified();
+ modifiedOutcome.setUndeliverableHere(Boolean.TRUE);
+ interaction.dispositionSettled(true)
+ .dispositionRole(Role.RECEIVER)
+ .dispositionFirstFromLatestDelivery()
+ .dispositionState(modifiedOutcome)
+ .disposition()
+ .flowIncomingWindow(UnsignedInteger.valueOf(2))
+ .flowLinkCredit(UnsignedInteger.valueOf(2))
+ .flowNextIncomingIdFromLatestDelivery()
+ .flow()
+ .receiveDelivery()
+ .decodeLatestDelivery();
+ ;
+
+ Object secondDeliveryPayload = interaction.getDecodedLatestDelivery();
+ assertThat(secondDeliveryPayload, is(equalTo("message2")));
+
+ // verify that no unexpected performative is received by closing
+ transport.doCloseConnection();
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[2/2] qpid-broker-j git commit: QPID-7531: [Java Broker,
AMQP 1.0] Improve error message
Posted by or...@apache.org.
QPID-7531: [Java Broker, AMQP 1.0] Improve error message
Project: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/commit/37b1a71a
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/37b1a71a
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/37b1a71a
Branch: refs/heads/master
Commit: 37b1a71aef67ef9043e64ea6766020fd47943112
Parents: ac6f6c3
Author: Alex Rudyy <or...@apache.org>
Authored: Fri Oct 6 10:57:03 2017 +0100
Committer: Alex Rudyy <or...@apache.org>
Committed: Fri Oct 6 10:57:03 2017 +0100
----------------------------------------------------------------------
.../java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/37b1a71a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
----------------------------------------------------------------------
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
index f54a19f..0f4b7ba 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
@@ -1361,8 +1361,8 @@ public class Session_1_0 extends AbstractAMQPSession<Session_1_0, ConsumerTarget
{
endpoint.close(new Error(AmqpError.RESOURCE_LIMIT_EXCEEDED,
String.format(
- "Cannot find free handle for endpoint '%d' on session '%s'",
- _attach.getHandle(),
+ "Cannot find free handle for endpoint '%s' on session '%s'",
+ _attach.getName(),
endpoint.getSession().toLogString())));
}
else
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org