You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kw...@apache.org on 2017/12/10 16:54:58 UTC
[1/2] qpid-broker-j git commit: QPID-8038: [Broker-J] [AMQP
0-8..0-91] Add rollback/mandatory/close-when-no-route tests
Repository: qpid-broker-j
Updated Branches:
refs/heads/master 89b6ea16d -> f0c04045a
QPID-8038: [Broker-J] [AMQP 0-8..0-91] Add rollback/mandatory/close-when-no-route tests
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/f0c04045
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/f0c04045
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/f0c04045
Branch: refs/heads/master
Commit: f0c04045a65870ef13e5c3037619ff7452f550b4
Parents: 971fba3
Author: Keith Wall <ke...@gmail.com>
Authored: Sun Dec 10 16:48:39 2017 +0000
Committer: Keith Wall <ke...@gmail.com>
Committed: Sun Dec 10 16:49:45 2017 +0000
----------------------------------------------------------------------
.../tests/protocol/v0_8/BasicInteraction.java | 19 ++
.../protocol/v0_8/ConnectionInteraction.java | 8 +-
.../qpid/tests/protocol/v0_8/TxInteraction.java | 7 +
.../qpid/tests/protocol/v0_8/BasicTest.java | 140 +++++++++
.../tests/protocol/v0_8/TransactionTest.java | 298 ++++++++++++++++++-
5 files changed, 467 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/f0c04045/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/BasicInteraction.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/BasicInteraction.java b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/BasicInteraction.java
index d5b4c62..7f91d8d 100644
--- a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/BasicInteraction.java
+++ b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/BasicInteraction.java
@@ -174,6 +174,18 @@ public class BasicInteraction
return this;
}
+ public BasicInteraction publishMandatory(final boolean mandatory)
+ {
+ _publishMandatory = mandatory;
+ return this;
+ }
+
+ public BasicInteraction publishImmediate(final boolean immediate)
+ {
+ _publishImmediate = immediate;
+ return this;
+ }
+
public BasicInteraction qosPrefetchCount(final int prefetchCount)
{
_qosPrefetchCount = prefetchCount;
@@ -211,6 +223,12 @@ public class BasicInteraction
return this;
}
+ public BasicInteraction consumeNoAck(final boolean noAck)
+ {
+ _consumeNoAck = noAck;
+ return this;
+ }
+
public Interaction ack() throws Exception
{
return _interaction.sendPerformative(new BasicAckBody(_ackDeliveryTag, _ackMultiple));
@@ -227,6 +245,7 @@ public class BasicInteraction
return _interaction.sendPerformative(new BasicCancelBody(AMQShortString.valueOf(_consumeCancelTag),
_consumeCancelNoWait));
}
+
public BasicInteraction consumeCancelTag(final String consumeCancelTag)
{
_consumeCancelTag = consumeCancelTag;
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/f0c04045/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ConnectionInteraction.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ConnectionInteraction.java b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ConnectionInteraction.java
index 023e7fc..236c49a 100644
--- a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ConnectionInteraction.java
+++ b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/ConnectionInteraction.java
@@ -20,6 +20,7 @@
*/
package org.apache.qpid.tests.protocol.v0_8;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -33,7 +34,7 @@ public class ConnectionInteraction
{
private final Interaction _interaction;
- private final Map<String, Object> _startOkClientProperties = new HashMap<>();
+ private Map<String, Object> _startOkClientProperties = new HashMap<>();
private String _startOkMechanism;
private byte[] _startOkResponse;
private String _startOkLocale;
@@ -54,6 +55,11 @@ public class ConnectionInteraction
return this;
}
+ public ConnectionInteraction startOkClientProperties(final Map<String, Object> clientProperties)
+ {
+ _startOkClientProperties = clientProperties == null ? Collections.emptyMap() : new HashMap<>(clientProperties);
+ return this;
+ }
public Interaction startOk() throws Exception
{
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/f0c04045/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/TxInteraction.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/TxInteraction.java b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/TxInteraction.java
index 63a078a..ee89993 100644
--- a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/TxInteraction.java
+++ b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/TxInteraction.java
@@ -21,7 +21,9 @@
package org.apache.qpid.tests.protocol.v0_8;
import org.apache.qpid.server.protocol.v0_8.transport.TxCommitBody;
+import org.apache.qpid.server.protocol.v0_8.transport.TxRollbackBody;
import org.apache.qpid.server.protocol.v0_8.transport.TxSelectBody;
+import org.apache.qpid.tests.protocol.AbstractInteraction;
public class TxInteraction
{
@@ -41,4 +43,9 @@ public class TxInteraction
{
return _interaction.sendPerformative(TxCommitBody.INSTANCE);
}
+
+ public Interaction rollback() throws Exception
+ {
+ return _interaction.sendPerformative(TxRollbackBody.INSTANCE);
+ }
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/f0c04045/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/BasicTest.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/BasicTest.java b/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/BasicTest.java
index 6bcbb84..643e4e7 100644
--- a/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/BasicTest.java
+++ b/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/BasicTest.java
@@ -25,6 +25,7 @@ import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.CoreMatchers.notNullValue;
import static org.hamcrest.CoreMatchers.nullValue;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assume.assumeThat;
import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
@@ -32,10 +33,12 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
+import org.hamcrest.Matchers;
import org.junit.Before;
import org.junit.Test;
import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.protocol.ErrorCodes;
import org.apache.qpid.server.protocol.v0_8.AMQShortString;
import org.apache.qpid.server.protocol.v0_8.FieldTable;
import org.apache.qpid.server.protocol.v0_8.transport.BasicConsumeOkBody;
@@ -44,11 +47,13 @@ import org.apache.qpid.server.protocol.v0_8.transport.BasicDeliverBody;
import org.apache.qpid.server.protocol.v0_8.transport.BasicGetEmptyBody;
import org.apache.qpid.server.protocol.v0_8.transport.BasicGetOkBody;
import org.apache.qpid.server.protocol.v0_8.transport.BasicQosOkBody;
+import org.apache.qpid.server.protocol.v0_8.transport.BasicReturnBody;
import org.apache.qpid.server.protocol.v0_8.transport.ChannelCloseOkBody;
import org.apache.qpid.server.protocol.v0_8.transport.ChannelFlowOkBody;
import org.apache.qpid.server.protocol.v0_8.transport.ChannelOpenOkBody;
import org.apache.qpid.server.protocol.v0_8.transport.ContentBody;
import org.apache.qpid.server.protocol.v0_8.transport.ContentHeaderBody;
+import org.apache.qpid.server.protocol.v0_8.transport.QueueDeclareOkBody;
import org.apache.qpid.tests.protocol.SpecificationTest;
import org.apache.qpid.tests.utils.BrokerAdmin;
import org.apache.qpid.tests.utils.BrokerAdminUsingTestBase;
@@ -88,6 +93,141 @@ public class BasicTest extends BrokerAdminUsingTestBase
}
}
+ @Test
+ @SpecificationTest(section = "1.8.3.7", description = "indicate mandatory routing")
+ public void publishMandatoryMessage() throws Exception
+ {
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+ interaction.openAnonymousConnection()
+ .channel().open().consumeResponse(ChannelOpenOkBody.class)
+ .basic().publishExchange("")
+ .publishRoutingKey(BrokerAdmin.TEST_QUEUE_NAME)
+ .publishMandatory(true)
+ .content("Test")
+ .publishMessage()
+ .channel().close()
+ .consumeResponse(ChannelCloseOkBody.class);
+
+ assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(1)));
+ }
+ }
+
+ @Test
+ @SpecificationTest(section = "1.8.3.7",
+ description = "This flag [mandatory] tells the server how to react if the message cannot be routed to a "
+ + "queue. If this flag is set, the server will return an unroutable message with a "
+ + "Return method.")
+ public void publishUnrouteableMandatoryMessage() throws Exception
+ {
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+ String messageContent = "Test";
+ BasicReturnBody returned = interaction.openAnonymousConnection()
+ .channel().open().consumeResponse(ChannelOpenOkBody.class)
+ .basic().publishExchange("")
+ .publishRoutingKey("unrouteable")
+ .publishMandatory(true)
+ .content(messageContent)
+ .publishMessage()
+ .consumeResponse().getLatestResponse(BasicReturnBody.class);
+ assertThat(returned.getReplyCode(), is(equalTo(ErrorCodes.NO_ROUTE)));
+
+ ContentBody content = interaction.consumeResponse(ContentHeaderBody.class)
+ .consumeResponse().getLatestResponse(ContentBody.class);
+
+ assertThat(getContent(content), is(equalTo(messageContent)));
+
+ interaction.channel().close().consumeResponse(ChannelCloseOkBody.class);
+ }
+ }
+
+ @Test
+ @SpecificationTest(section = "1.8.3.7",
+ description = "This flag [mandatory] tells the server how to react if the message cannot be routed to a "
+ + "queue. If this flag is zero, the server silently drops the message.")
+ public void publishUnrouteableMessage() throws Exception
+ {
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+ interaction.openAnonymousConnection()
+ .channel().open().consumeResponse(ChannelOpenOkBody.class)
+ .basic().publishExchange("")
+ .publishRoutingKey("unrouteable")
+ .publishMandatory(false)
+ .content("Test")
+ .publishMessage()
+ .channel().close().consumeResponse(ChannelCloseOkBody.class);
+ }
+ }
+
+ @Test
+ @SpecificationTest(section = "1.8",
+ description = "The server SHOULD respect the persistent property of basic messages and SHOULD make a best "
+ + "effort to hold persistent basic messages on a reliable storage mechanism.")
+ public void messagePersistence() throws Exception
+ {
+ String queueName = "durableQueue";
+ String messageContent = "Test";
+ String messageContentType = "text/plain";
+ byte deliveryMode = (byte) 2;
+ Map<String, Object> messageHeaders = Collections.singletonMap("test", "testValue");
+
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+ interaction.openAnonymousConnection()
+ .channel().open().consumeResponse(ChannelOpenOkBody.class)
+ .queue().declareName(queueName).declareDurable(true).declare()
+ .consumeResponse(QueueDeclareOkBody.class)
+ .basic().contentHeaderPropertiesContentType(messageContentType)
+ .contentHeaderPropertiesHeaders(messageHeaders)
+ .contentHeaderPropertiesDeliveryMode(deliveryMode)
+ .publishExchange("")
+ .publishRoutingKey(queueName)
+ .content(messageContent)
+ .publishMessage()
+ .channel().close()
+ .consumeResponse(ChannelCloseOkBody.class);
+
+ assertThat(getBrokerAdmin().getQueueDepthMessages(queueName), is(equalTo(1)));
+ }
+
+ assumeThat(getBrokerAdmin().supportsRestart(), Matchers.is(true));
+ getBrokerAdmin().restart();
+ assertThat(getBrokerAdmin().getQueueDepthMessages(queueName), is(equalTo(1)));
+
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+
+ interaction.openAnonymousConnection()
+ .channel().open()
+ .consumeResponse(ChannelOpenOkBody.class)
+ .basic().getQueueName(queueName).getNoAck(true).get()
+ .consumeResponse(BasicGetOkBody.class);
+
+ ContentHeaderBody header = interaction.consumeResponse().getLatestResponse(ContentHeaderBody.class);
+ ContentBody content = interaction.consumeResponse().getLatestResponse(ContentBody.class);
+
+ String receivedContent = getContent(content);
+ BasicContentHeaderProperties properties = header.getProperties();
+
+ assertThat(receivedContent, is(equalTo(messageContent)));
+
+ Map<String, Object> receivedHeaders = new HashMap<>(FieldTable.convertToMap(properties.getHeaders()));
+ assertThat(receivedHeaders, is(equalTo(new HashMap<>(messageHeaders))));
+ assertThat(properties.getContentTypeAsString(), is(equalTo(messageContentType)));
+ assertThat(properties.getDeliveryMode(), is(equalTo(deliveryMode)));
+
+ interaction.channel().close().consumeResponse(ChannelCloseOkBody.class);
+
+ assertThat(getBrokerAdmin().getQueueDepthMessages(queueName), is(equalTo(0)));
+ }
+ }
@Test
@SpecificationTest(section = "1.8.3.3", description = "start a queue consumer")
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/f0c04045/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/TransactionTest.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/TransactionTest.java b/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/TransactionTest.java
index 6e1208a..93127ad 100644
--- a/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/TransactionTest.java
+++ b/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/TransactionTest.java
@@ -26,14 +26,14 @@ import static org.hamcrest.MatcherAssert.assertThat;
import java.net.InetSocketAddress;
import java.util.Collections;
+import java.util.Map;
import org.junit.Before;
import org.junit.Test;
-import org.apache.qpid.server.protocol.v0_8.transport.ChannelOpenOkBody;
-import org.apache.qpid.server.protocol.v0_8.transport.ExchangeDeclareOkBody;
-import org.apache.qpid.server.protocol.v0_8.transport.TxCommitOkBody;
-import org.apache.qpid.server.protocol.v0_8.transport.TxSelectOkBody;
+import org.apache.qpid.server.properties.ConnectionStartProperties;
+import org.apache.qpid.server.protocol.ErrorCodes;
+import org.apache.qpid.server.protocol.v0_8.transport.*;
import org.apache.qpid.tests.protocol.SpecificationTest;
import org.apache.qpid.tests.utils.BrokerAdmin;
import org.apache.qpid.tests.utils.BrokerAdminUsingTestBase;
@@ -50,6 +50,40 @@ public class TransactionTest extends BrokerAdminUsingTestBase
}
@Test
+ @SpecificationTest(section = "1.9.2.3", description = "The client MUST NOT use the Commit method on non-transacted "
+ + "channels.")
+ public void commitNonTransactedChannel() throws Exception
+ {
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+ ChannelCloseBody res = interaction.openAnonymousConnection()
+ .channel().open().consumeResponse(ChannelOpenOkBody.class)
+ .tx().commit()
+ .consumeResponse().getLatestResponse(ChannelCloseBody.class);
+ // TODO specification requires precondition-failed (406)
+ assertThat(res.getReplyCode(), is(equalTo(ErrorCodes.COMMAND_INVALID)));
+ }
+ }
+
+ @Test
+ @SpecificationTest(section = "1.9.2.3", description = "The client MUST NOT use the Rollback method on non-transacted "
+ + "channels.")
+ public void rollbackNonTransactedChannel() throws Exception
+ {
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+ ChannelCloseBody res = interaction.openAnonymousConnection()
+ .channel().open().consumeResponse(ChannelOpenOkBody.class)
+ .tx().rollback()
+ .consumeResponse().getLatestResponse(ChannelCloseBody.class);
+ // TODO specification requires precondition-failed (406)
+ assertThat(res.getReplyCode(), is(equalTo(ErrorCodes.COMMAND_INVALID)));
+ }
+ }
+
+ @Test
@SpecificationTest(section = "1.9.2.3", description = "commit the current transaction")
public void publishMessage() throws Exception
{
@@ -75,5 +109,261 @@ public class TransactionTest extends BrokerAdminUsingTestBase
assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(1)));
}
}
+ @Test
+ @SpecificationTest(section = "1.9.2.5", description = "abandon the current transaction")
+ public void publishMessageRollback() throws Exception
+ {
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+ interaction.openAnonymousConnection()
+ .channel().open().consumeResponse(ChannelOpenOkBody.class)
+ .tx().select().consumeResponse(TxSelectOkBody.class)
+ .basic()
+ .publishExchange("")
+ .publishRoutingKey(BrokerAdmin.TEST_QUEUE_NAME)
+ .content("Test")
+ .publishMessage()
+ .exchange().declarePassive(true).declare().consumeResponse(ExchangeDeclareOkBody.class);
+ assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(0)));
+
+ interaction.tx().rollback().consumeResponse(TxRollbackOkBody.class);
+
+ assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(0)));
+ }
+ }
+
+ @Test
+ @SpecificationTest(section = "1.9.2.3", description = "commit the current transaction")
+ public void consumeMessage() throws Exception
+ {
+ getBrokerAdmin().putMessageOnQueue(BrokerAdmin.TEST_QUEUE_NAME, "message");
+
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+
+ assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(1)));
+
+ interaction.openAnonymousConnection()
+ .channel().open().consumeResponse(ChannelOpenOkBody.class)
+ .tx().select().consumeResponse(TxSelectOkBody.class)
+ .basic().qosPrefetchCount(1)
+ .qos()
+ .consumeResponse(BasicQosOkBody.class)
+ .channel().flow(true)
+ .consumeResponse(ChannelFlowOkBody.class)
+ .basic()
+ .consumeConsumerTag("A")
+ .consumeQueue(BrokerAdmin.TEST_QUEUE_NAME)
+ .consume()
+ .consumeResponse(BasicConsumeOkBody.class)
+ .consumeResponse(BasicDeliverBody.class);
+
+ BasicDeliverBody delivery = interaction.getLatestResponse(BasicDeliverBody.class);
+ interaction.consumeResponse(ContentHeaderBody.class)
+ .consumeResponse(ContentBody.class);
+
+ interaction.basic().ackDeliveryTag(delivery.getDeliveryTag())
+ .ack()
+ .exchange().declarePassive(true).declare().consumeResponse(ExchangeDeclareOkBody.class);
+
+ assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(1)));
+
+ interaction.tx().commit().consumeResponse(TxCommitOkBody.class);
+
+ assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(0)));
+ }
+ }
+
+ @Test
+ @SpecificationTest(section = "1.9.2.5", description = "abandon the current transaction")
+ public void consumeMessageRollbackAckdMessage() throws Exception
+ {
+ getBrokerAdmin().putMessageOnQueue(BrokerAdmin.TEST_QUEUE_NAME, "message");
+
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+
+ assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(1)));
+
+ interaction.openAnonymousConnection()
+ .channel().open().consumeResponse(ChannelOpenOkBody.class)
+ .tx().select().consumeResponse(TxSelectOkBody.class)
+ .basic().qosPrefetchCount(1)
+ .qos()
+ .consumeResponse(BasicQosOkBody.class)
+ .channel().flow(true)
+ .consumeResponse(ChannelFlowOkBody.class)
+ .basic()
+ .consumeConsumerTag("A")
+ .consumeQueue(BrokerAdmin.TEST_QUEUE_NAME)
+ .consume()
+ .consumeResponse(BasicConsumeOkBody.class)
+ .consumeResponse(BasicDeliverBody.class);
+
+ BasicDeliverBody delivery = interaction.getLatestResponse(BasicDeliverBody.class);
+ interaction.consumeResponse(ContentHeaderBody.class)
+ .consumeResponse(ContentBody.class);
+
+ assertThat(delivery.getRedelivered(), is(equalTo(false)));
+
+ interaction.basic().ackDeliveryTag(delivery.getDeliveryTag())
+ .ack()
+ .exchange().declarePassive(true).declare().consumeResponse(ExchangeDeclareOkBody.class);
+
+ assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(1)));
+
+ interaction.tx().rollback().consumeResponse(TxRollbackOkBody.class);
+
+ assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(1)));
+
+ // TODO: the server currently resends the messages again, which is not specification compliant but the
+ // Qpid JMS AMQP 0-x relies on this behaviour.
+ BasicDeliverBody redelivery = interaction.consumeResponse().getLatestResponse(BasicDeliverBody.class);
+ interaction.consumeResponse(ContentHeaderBody.class)
+ .consumeResponse(ContentBody.class);
+
+ assertThat(redelivery.getRedelivered(), is(equalTo(false)));
+
+ // TODO: defect - the server actually sends the message again.
+ redelivery = interaction.consumeResponse().getLatestResponse(BasicDeliverBody.class);
+ interaction.consumeResponse(ContentHeaderBody.class)
+ .consumeResponse(ContentBody.class);
+
+ interaction.channel().close()
+ .consumeResponse(ChannelCloseOkBody.class);
+ }
+ }
+
+ @Test
+ @SpecificationTest(section = "1.9.2.5", description = "abandon the current transaction")
+ public void consumeMessageRollbackDeliveredMessage() throws Exception
+ {
+ getBrokerAdmin().putMessageOnQueue(BrokerAdmin.TEST_QUEUE_NAME, "message");
+
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+
+ assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(1)));
+
+ interaction.openAnonymousConnection()
+ .channel().open().consumeResponse(ChannelOpenOkBody.class)
+ .tx().select().consumeResponse(TxSelectOkBody.class)
+ .basic().qosPrefetchCount(1)
+ .qos()
+ .consumeResponse(BasicQosOkBody.class)
+ .channel().flow(true)
+ .consumeResponse(ChannelFlowOkBody.class)
+ .basic()
+ .consumeConsumerTag("A")
+ .consumeQueue(BrokerAdmin.TEST_QUEUE_NAME)
+ .consume()
+ .consumeResponse(BasicConsumeOkBody.class)
+ .consumeResponse(BasicDeliverBody.class);
+
+ BasicDeliverBody delivery = interaction.getLatestResponse(BasicDeliverBody.class);
+ interaction.consumeResponse(ContentHeaderBody.class)
+ .consumeResponse(ContentBody.class);
+
+ assertThat(delivery.getRedelivered(), is(equalTo(false)));
+
+ assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(1)));
+
+ interaction.tx().rollback().consumeResponse(TxRollbackOkBody.class);
+
+ assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(1)));
+
+ // TODO: the server currently resends the messages again, which is not specification compliant but the
+ // Qpid JMS AMQP 0-x relies on this behaviour.
+ BasicDeliverBody redelivery = interaction.consumeResponse().getLatestResponse(BasicDeliverBody.class);
+ interaction.consumeResponse(ContentHeaderBody.class)
+ .consumeResponse(ContentBody.class);
+
+ assertThat(redelivery.getRedelivered(), is(equalTo(true)));
+
+ interaction.channel().close()
+ .consumeResponse(ChannelCloseOkBody.class);
+ }
+ }
+ /** Qpid specific extension */
+ @Test
+ public void publishUnrouteableMandatoryMessageCloseWhenNoRoute() throws Exception
+ {
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+
+ Map<String, Object> clientProperties = Collections.singletonMap(ConnectionStartProperties.QPID_CLOSE_WHEN_NO_ROUTE, true);
+ ConnectionCloseBody response = interaction.negotiateProtocol()
+ .consumeResponse(ConnectionStartBody.class).connection()
+ .startOkClientProperties(clientProperties).startOkMechanism("ANONYMOUS").startOk()
+ .consumeResponse(ConnectionTuneBody.class)
+ .connection().tuneOk()
+ .connection().open()
+ .consumeResponse(ConnectionOpenOkBody.class)
+ .channel().open().consumeResponse(ChannelOpenOkBody.class)
+ .tx()
+ .select()
+ .consumeResponse(TxSelectOkBody.class)
+ .basic()
+ .publishExchange("")
+ .publishRoutingKey("unrouteable")
+ .publishMandatory(true)
+ .content("Test")
+ .publishMessage()
+ .consumeResponse().getLatestResponse(ConnectionCloseBody.class);
+
+ assertThat(response.getReplyCode(), is(equalTo(ErrorCodes.NO_ROUTE)));
+ }
+ }
+
+ /** Qpid specific extension */
+ @Test
+ public void publishUnrouteableMandatory() throws Exception
+ {
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+
+ Map<String, Object> clientProperties = Collections.singletonMap(ConnectionStartProperties.QPID_CLOSE_WHEN_NO_ROUTE, false);
+ final String messageContent = "Test";
+ BasicReturnBody returned = interaction.negotiateProtocol()
+ .consumeResponse(ConnectionStartBody.class)
+ .connection()
+ .startOkClientProperties(clientProperties)
+ .startOkMechanism("ANONYMOUS")
+ .startOk()
+ .consumeResponse(ConnectionTuneBody.class)
+ .connection()
+ .tuneOk()
+ .connection()
+ .open()
+ .consumeResponse(ConnectionOpenOkBody.class)
+ .channel()
+ .open()
+ .consumeResponse(ChannelOpenOkBody.class)
+ .tx()
+ .select()
+ .consumeResponse(TxSelectOkBody.class)
+ .basic()
+ .publishExchange("")
+ .publishRoutingKey("unrouteable")
+ .publishMandatory(true)
+ .content(messageContent)
+ .publishMessage()
+ .tx().commit()
+ .consumeResponse()
+ .getLatestResponse(BasicReturnBody.class);
+ assertThat(returned.getReplyCode(), is(equalTo(ErrorCodes.NO_ROUTE)));
+
+ interaction.consumeResponse(ContentHeaderBody.class)
+ .consumeResponse(ContentBody.class)
+ .consumeResponse(TxCommitOkBody.class)
+ .channel().close().consumeResponse(ChannelCloseOkBody.class);
+ }
+ }
}
---------------------------------------------------------------------
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-8038: [Broker-J] [AMQP
0-8..0-91] Add more queue/basic protocol tests
Posted by kw...@apache.org.
QPID-8038: [Broker-J] [AMQP 0-8..0-91] Add more queue/basic protocol tests
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/971fba30
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/971fba30
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/971fba30
Branch: refs/heads/master
Commit: 971fba301e50f95f827446ca921c4e37244dadbc
Parents: 89b6ea1
Author: Keith Wall <ke...@gmail.com>
Authored: Thu Dec 7 18:15:18 2017 +0000
Committer: Keith Wall <ke...@gmail.com>
Committed: Sun Dec 10 16:49:45 2017 +0000
----------------------------------------------------------------------
.../tests/protocol/v0_8/BasicInteraction.java | 24 +++++-
.../qpid/tests/protocol/v0_8/BasicTest.java | 91 ++++++++++++++++++--
.../qpid/tests/protocol/v0_8/QueueTest.java | 88 +++++++++++++++++++
3 files changed, 197 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/971fba30/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/BasicInteraction.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/BasicInteraction.java b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/BasicInteraction.java
index bcac578..d5b4c62 100644
--- a/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/BasicInteraction.java
+++ b/systests/protocol-tests-amqp-0-8/src/main/java/org/apache/qpid/tests/protocol/v0_8/BasicInteraction.java
@@ -34,6 +34,7 @@ import org.apache.qpid.server.protocol.v0_8.transport.BasicAckBody;
import org.apache.qpid.server.protocol.v0_8.transport.BasicCancelBody;
import org.apache.qpid.server.protocol.v0_8.transport.BasicConsumeBody;
import org.apache.qpid.server.protocol.v0_8.transport.BasicContentHeaderProperties;
+import org.apache.qpid.server.protocol.v0_8.transport.BasicGetBody;
import org.apache.qpid.server.protocol.v0_8.transport.BasicPublishBody;
import org.apache.qpid.server.protocol.v0_8.transport.BasicQosBody;
import org.apache.qpid.server.protocol.v0_8.transport.CompositeAMQDataBlock;
@@ -68,6 +69,9 @@ public class BasicInteraction
private String _consumeCancelTag;
private boolean _consumeCancelNoWait;
+ private String _getQueueName;
+ private boolean _getNoAck;
+
public BasicInteraction(final Interaction interaction)
{
_interaction = interaction;
@@ -223,10 +227,28 @@ public class BasicInteraction
return _interaction.sendPerformative(new BasicCancelBody(AMQShortString.valueOf(_consumeCancelTag),
_consumeCancelNoWait));
}
-
public BasicInteraction consumeCancelTag(final String consumeCancelTag)
{
_consumeCancelTag = consumeCancelTag;
return this;
}
+
+ public Interaction get() throws Exception
+ {
+ return _interaction.sendPerformative(new BasicGetBody(0,
+ AMQShortString.valueOf(_getQueueName),
+ _getNoAck));
+ }
+
+ public BasicInteraction getQueueName(final String queueName)
+ {
+ _getQueueName = queueName;
+ return this;
+ }
+
+ public BasicInteraction getNoAck(final boolean noAck)
+ {
+ _getNoAck = noAck;
+ return this;
+ }
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/971fba30/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/BasicTest.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/BasicTest.java b/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/BasicTest.java
index 1eb0edd..6bcbb84 100644
--- a/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/BasicTest.java
+++ b/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/BasicTest.java
@@ -41,6 +41,8 @@ import org.apache.qpid.server.protocol.v0_8.FieldTable;
import org.apache.qpid.server.protocol.v0_8.transport.BasicConsumeOkBody;
import org.apache.qpid.server.protocol.v0_8.transport.BasicContentHeaderProperties;
import org.apache.qpid.server.protocol.v0_8.transport.BasicDeliverBody;
+import org.apache.qpid.server.protocol.v0_8.transport.BasicGetEmptyBody;
+import org.apache.qpid.server.protocol.v0_8.transport.BasicGetOkBody;
import org.apache.qpid.server.protocol.v0_8.transport.BasicQosOkBody;
import org.apache.qpid.server.protocol.v0_8.transport.ChannelCloseOkBody;
import org.apache.qpid.server.protocol.v0_8.transport.ChannelFlowOkBody;
@@ -143,11 +145,7 @@ public class BasicTest extends BrokerAdminUsingTestBase
ContentBody content = interaction.consumeResponse(ContentBody.class).getLatestResponse(ContentBody.class);
- QpidByteBuffer payload = content.getPayload();
- byte[] contentData = new byte[payload.remaining()];
- payload.get(contentData);
- payload.dispose();
- String receivedContent = new String(contentData, StandardCharsets.UTF_8);
+ String receivedContent = getContent(content);
assertThat(receivedContent, is(equalTo(messageContent)));
assertThat(getBrokerAdmin().getQueueDepthMessages(queueName), is(equalTo(1)));
@@ -192,4 +190,87 @@ public class BasicTest extends BrokerAdminUsingTestBase
}
}
+ @Test
+ @SpecificationTest(section = "1.8.3.10", description = "direct access to a queue")
+ public void get() throws Exception
+ {
+ String messageContent = "message";
+ getBrokerAdmin().putMessageOnQueue(BrokerAdmin.TEST_QUEUE_NAME, messageContent);
+
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+ BasicGetOkBody response = interaction.openAnonymousConnection()
+ .channel().open()
+ .consumeResponse(ChannelOpenOkBody.class)
+ .basic().getQueueName(BrokerAdmin.TEST_QUEUE_NAME).get()
+ .consumeResponse().getLatestResponse(BasicGetOkBody.class);
+
+ long deliveryTag = response.getDeliveryTag();
+ ContentBody content = interaction.consumeResponse(ContentHeaderBody.class)
+ .consumeResponse().getLatestResponse(ContentBody.class);
+
+ String receivedContent = getContent(content);
+ assertThat(receivedContent, is(equalTo(messageContent)));
+
+ assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(1)));
+
+ interaction.basic().ackDeliveryTag(deliveryTag).ack()
+ .channel().close().consumeResponse(ChannelCloseOkBody.class);
+
+ assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(0)));
+ }
+ }
+
+ @Test
+ @SpecificationTest(section = "1.8.3.10", description = "direct access to a queue")
+ public void getNoAck() throws Exception
+ {
+ String messageContent = "message";
+ getBrokerAdmin().putMessageOnQueue(BrokerAdmin.TEST_QUEUE_NAME, messageContent);
+
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+ interaction.openAnonymousConnection()
+ .channel().open()
+ .consumeResponse(ChannelOpenOkBody.class)
+ .basic().getQueueName(BrokerAdmin.TEST_QUEUE_NAME).getNoAck(true).get()
+ .consumeResponse(BasicGetOkBody.class);
+
+ ContentBody content = interaction.consumeResponse(ContentHeaderBody.class)
+ .consumeResponse().getLatestResponse(ContentBody.class);
+
+ String receivedContent = getContent(content);
+ assertThat(receivedContent, is(equalTo(messageContent)));
+
+ interaction.channel().close().consumeResponse(ChannelCloseOkBody.class);
+
+ assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(0)));
+ }
+ }
+
+ @Test
+ @SpecificationTest(section = "1.8.3.10", description = "direct access to a queue")
+ public void getEmptyQueue() throws Exception
+ {
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+ interaction.openAnonymousConnection()
+ .channel().open()
+ .consumeResponse(ChannelOpenOkBody.class)
+ .basic().getQueueName(BrokerAdmin.TEST_QUEUE_NAME).get()
+ .consumeResponse().getLatestResponse(BasicGetEmptyBody.class);
+ }
+ }
+
+ private String getContent(final ContentBody content)
+ {
+ QpidByteBuffer payload = content.getPayload();
+ byte[] contentData = new byte[payload.remaining()];
+ payload.get(contentData);
+ payload.dispose();
+ return new String(contentData, StandardCharsets.UTF_8);
+ }
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/971fba30/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/QueueTest.java
----------------------------------------------------------------------
diff --git a/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/QueueTest.java b/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/QueueTest.java
index 5657c85..9a52a45 100644
--- a/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/QueueTest.java
+++ b/systests/protocol-tests-amqp-0-8/src/test/java/org/apache/qpid/tests/protocol/v0_8/QueueTest.java
@@ -389,6 +389,28 @@ public class QueueTest extends BrokerAdminUsingTestBase
}
@Test
+ @SpecificationTest(section = "1.7.2.9",
+ description = "The client MUST either specify a queue name or have previously declared a queue on the "
+ + "same channel")
+ public void queueDeleteDefaultQueue() throws Exception
+ {
+ getBrokerAdmin().createQueue(BrokerAdmin.TEST_QUEUE_NAME);
+ getBrokerAdmin().putMessageOnQueue(BrokerAdmin.TEST_QUEUE_NAME, "message");
+
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+ QueueDeleteOkBody deleteResponse = interaction.openAnonymousConnection()
+ .channel().open().consumeResponse(ChannelOpenOkBody.class)
+ .queue().declareName(BrokerAdmin.TEST_QUEUE_NAME).declare()
+ .consumeResponse(QueueDeclareOkBody.class)
+ .queue().delete()
+ .consumeResponse().getLatestResponse(QueueDeleteOkBody.class);
+ assertThat(deleteResponse.getMessageCount(), is(equalTo(1L)));
+ }
+ }
+
+ @Test
@SpecificationTest(section = "1.7.2.7", description = "purge a queue")
public void queuePurge() throws Exception
{
@@ -416,6 +438,21 @@ public class QueueTest extends BrokerAdminUsingTestBase
}
@Test
+ @SpecificationTest(section = "1.7.2.7", description = "The client MUST NOT attempt to purge a queue that does not exist.")
+ public void queuePurgeQueueNotFound() throws Exception
+ {
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+ ChannelCloseBody response = interaction.openAnonymousConnection()
+ .channel().open().consumeResponse(ChannelOpenOkBody.class)
+ .queue().purgeName(BrokerAdmin.TEST_QUEUE_NAME).purge()
+ .consumeResponse().getLatestResponse(ChannelCloseBody.class);
+ assertThat(response.getReplyCode(), is(equalTo(ErrorCodes.NOT_FOUND)));
+ }
+ }
+
+ @Test
@SpecificationTest(section = "1.7.2.3", description = "bind queue to an exchange")
public void queueBind() throws Exception
{
@@ -498,6 +535,34 @@ public class QueueTest extends BrokerAdminUsingTestBase
@Test
@SpecificationTest(section = "1.7.2.3",
+ description = "The client MUST either specify a queue name or have previously declared a queue on the same channel")
+ public void queueBindDefaultQueue() throws Exception
+ {
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ String testExchange = "testExchange";
+ final Interaction interaction = transport.newInteraction();
+ interaction.openAnonymousConnection()
+ .channel().open().consumeResponse(ChannelOpenOkBody.class)
+ .queue().declareName(BrokerAdmin.TEST_QUEUE_NAME).declare()
+ .consumeResponse(QueueDeclareOkBody.class)
+ .exchange().declareName(testExchange).declare()
+ .consumeResponse(ExchangeDeclareOkBody.class)
+ .queue().bindName(testExchange).bind()
+ .consumeResponse(QueueBindOkBody.class);
+
+ ExchangeBoundOkBody response = interaction.exchange()
+ .boundExchangeName(testExchange)
+ .boundQueue(BrokerAdmin.TEST_QUEUE_NAME)
+ .bound()
+ .consumeResponse()
+ .getLatestResponse(ExchangeBoundOkBody.class);
+ assertThat(response.getReplyCode(), is(equalTo(ExchangeBoundOkBody.OK)));
+ }
+ }
+
+ @Test
+ @SpecificationTest(section = "1.7.2.3",
description = "Bindings of durable queues to durable exchanges are automatically durable and the server "
+ "MUST restore such bindings after a server restart.")
public void queueDurableBind() throws Exception
@@ -599,6 +664,29 @@ public class QueueTest extends BrokerAdminUsingTestBase
}
@Test
+ @SpecificationTest(section = "1.7.2.5",
+ description = "The client MUST either specify a queue name or have previously declared a queue on the "
+ + "same channel")
+ public void queueUnbindDefaultQueue() throws Exception
+ {
+ try(FrameTransport transport = new FrameTransport(_brokerAddress).connect())
+ {
+ final Interaction interaction = transport.newInteraction();
+ String testExchange = "testExchange";
+ interaction.openAnonymousConnection()
+ .channel().open().consumeResponse(ChannelOpenOkBody.class)
+ .queue().declareName(BrokerAdmin.TEST_QUEUE_NAME).declare()
+ .consumeResponse(QueueDeclareOkBody.class)
+ .exchange().declareName(testExchange).declare()
+ .consumeResponse(ExchangeDeclareOkBody.class)
+ .queue().bindName(testExchange).bindQueueName(BrokerAdmin.TEST_QUEUE_NAME).bindRoutingKey("rk1").bind()
+ .consumeResponse(QueueBindOkBody.class)
+ .queue().unbindName(testExchange).unbindRoutingKey("rk1").unbind()
+ .consumeResponse(QueueUnbindOkBody.class);
+ }
+ }
+
+ @Test
@SpecificationTest(section = "1.7.2.5", description = "The client MUST NOT attempt to unbind a queue that does "
+ "not exist.")
public void queueUnbindUnknownQueue() throws Exception
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org