| /* |
| * |
| * 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.v0_8; |
| |
| import static org.hamcrest.CoreMatchers.anyOf; |
| import static org.hamcrest.CoreMatchers.equalTo; |
| import static org.hamcrest.CoreMatchers.is; |
| import static org.hamcrest.CoreMatchers.not; |
| import static org.hamcrest.MatcherAssert.assertThat; |
| import static org.hamcrest.Matchers.isEmptyString; |
| import static org.junit.Assume.assumeThat; |
| |
| import java.net.InetSocketAddress; |
| import java.util.Collections; |
| import java.util.Map; |
| |
| import org.hamcrest.Matchers; |
| import org.junit.Before; |
| import org.junit.Ignore; |
| import org.junit.Test; |
| |
| import org.apache.qpid.server.exchange.ExchangeDefaults; |
| import org.apache.qpid.server.filter.AMQPFilterTypes; |
| import org.apache.qpid.server.protocol.ErrorCodes; |
| import org.apache.qpid.server.protocol.v0_8.AMQShortString; |
| import org.apache.qpid.server.protocol.v0_8.transport.BasicCancelOkBody; |
| import org.apache.qpid.server.protocol.v0_8.transport.BasicConsumeOkBody; |
| import org.apache.qpid.server.protocol.v0_8.transport.ChannelCloseBody; |
| import org.apache.qpid.server.protocol.v0_8.transport.ChannelOpenOkBody; |
| import org.apache.qpid.server.protocol.v0_8.transport.ConnectionCloseBody; |
| import org.apache.qpid.server.protocol.v0_8.transport.ExchangeBoundOkBody; |
| import org.apache.qpid.server.protocol.v0_8.transport.ExchangeDeclareOkBody; |
| import org.apache.qpid.server.protocol.v0_8.transport.ExchangeDeleteOkBody; |
| import org.apache.qpid.server.protocol.v0_8.transport.QueueBindOkBody; |
| import org.apache.qpid.server.protocol.v0_8.transport.QueueDeclareOkBody; |
| import org.apache.qpid.server.protocol.v0_8.transport.QueueDeleteOkBody; |
| import org.apache.qpid.server.protocol.v0_8.transport.QueuePurgeOkBody; |
| import org.apache.qpid.server.protocol.v0_8.transport.QueueUnbindOkBody; |
| import org.apache.qpid.server.protocol.v0_8.transport.TxCommitOkBody; |
| import org.apache.qpid.server.protocol.v0_8.transport.TxSelectOkBody; |
| import org.apache.qpid.tests.protocol.SpecificationTest; |
| import org.apache.qpid.tests.utils.BrokerAdmin; |
| import org.apache.qpid.tests.utils.BrokerAdminUsingTestBase; |
| |
| public class QueueTest extends BrokerAdminUsingTestBase |
| { |
| private InetSocketAddress _brokerAddress; |
| |
| @Before |
| public void setUp() |
| { |
| _brokerAddress = getBrokerAdmin().getBrokerAddress(BrokerAdmin.PortType.ANONYMOUS_AMQP); |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.1", description = "declare queue, create if needed") |
| public void queueDeclare() throws Exception |
| { |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| QueueDeclareOkBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().declareName(BrokerAdmin.TEST_QUEUE_NAME).declare() |
| .consumeResponse().getLatestResponse(QueueDeclareOkBody.class); |
| |
| assertThat(response.getQueue(), is(equalTo(AMQShortString.valueOf(BrokerAdmin.TEST_QUEUE_NAME)))); |
| assertThat(response.getMessageCount(), is(equalTo(0L))); |
| assertThat(response.getConsumerCount(), is(equalTo(0L))); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.1", description = "If not set and the queue exists, the server MUST check " |
| + "that the existing queue has the same values for durable, " |
| + "exclusive, auto-delete, and arguments fields.") |
| public void queueDeclareEquivalent() throws Exception |
| { |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| QueueInteraction queueInteraction = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue(); |
| QueueDeclareOkBody response = queueInteraction.declareName(BrokerAdmin.TEST_QUEUE_NAME) |
| .declareExclusive(false).declare() |
| .consumeResponse().getLatestResponse(QueueDeclareOkBody.class); |
| |
| assertThat(response.getQueue(), is(equalTo(AMQShortString.valueOf(BrokerAdmin.TEST_QUEUE_NAME)))); |
| |
| QueueDeclareOkBody equalDeclareResponse = queueInteraction.declareName(BrokerAdmin.TEST_QUEUE_NAME) |
| .declareExclusive(false).declare() |
| .consumeResponse().getLatestResponse(QueueDeclareOkBody.class); |
| assertThat(equalDeclareResponse.getQueue(), is(equalTo(AMQShortString.valueOf(BrokerAdmin.TEST_QUEUE_NAME)))); |
| |
| ChannelCloseBody unequalDeclareResponse = queueInteraction.declareName(BrokerAdmin.TEST_QUEUE_NAME) |
| .declareExclusive(true).declare() |
| .consumeResponse().getLatestResponse(ChannelCloseBody.class); |
| assertThat(unequalDeclareResponse.getReplyCode(), is(equalTo(ErrorCodes.ALREADY_EXISTS))); |
| |
| interaction.channel().closeOk(); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.1", |
| description = "If [declarePassive is] set, the server will reply with Declare-Ok if the queue already exists" |
| + "with the same name, and raise an error if not.") |
| public void queueDeclarePassive() throws Exception |
| { |
| getBrokerAdmin().createQueue(BrokerAdmin.TEST_QUEUE_NAME); |
| |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| QueueDeclareOkBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().declarePassive(true).declareName(BrokerAdmin.TEST_QUEUE_NAME).declare() |
| .consumeResponse().getLatestResponse(QueueDeclareOkBody.class); |
| |
| assertThat(response.getQueue(), is(equalTo(AMQShortString.valueOf(BrokerAdmin.TEST_QUEUE_NAME)))); |
| assertThat(response.getMessageCount(), is(equalTo(0L))); |
| assertThat(response.getConsumerCount(), is(equalTo(0L))); |
| |
| getBrokerAdmin().deleteQueue(BrokerAdmin.TEST_QUEUE_NAME); |
| |
| ChannelCloseBody closeResponse = interaction.queue() |
| .deleteName(BrokerAdmin.TEST_QUEUE_NAME).delete() |
| .consumeResponse().getLatestResponse(ChannelCloseBody.class); |
| assertThat(closeResponse.getReplyCode(), is(equalTo(ErrorCodes.NOT_FOUND))); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.1", |
| description = "If [durable is] set when creating a new queue, the queue will be marked as durable. " |
| + "Durable queues remain active when a server restarts.") |
| public void queueDeclareDurable() throws Exception |
| { |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| QueueDeclareOkBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().declareDurable(true).declareName(BrokerAdmin.TEST_QUEUE_NAME).declare() |
| .consumeResponse().getLatestResponse(QueueDeclareOkBody.class); |
| |
| assertThat(response.getQueue(), is(equalTo(AMQShortString.valueOf(BrokerAdmin.TEST_QUEUE_NAME)))); |
| assertThat(response.getMessageCount(), is(equalTo(0L))); |
| assertThat(response.getConsumerCount(), is(equalTo(0L))); |
| } |
| |
| assumeThat(getBrokerAdmin().supportsRestart(), Matchers.is(true)); |
| getBrokerAdmin().restart(); |
| |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| QueueDeclareOkBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().declarePassive(true).declareName(BrokerAdmin.TEST_QUEUE_NAME).declare() |
| .consumeResponse().getLatestResponse(QueueDeclareOkBody.class); |
| |
| assertThat(response.getQueue(), is(equalTo(AMQShortString.valueOf(BrokerAdmin.TEST_QUEUE_NAME)))); |
| assertThat(response.getMessageCount(), is(equalTo(0L))); |
| assertThat(response.getConsumerCount(), is(equalTo(0L))); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.1", |
| description = "If [auto-delete] set, the queue is deleted when all consumers have finished using it. The " |
| + "last consumer can be cancelled either explicitly or because its channel is closed. " |
| + "If there was no consumer ever on the queue, it won't be deleted.") |
| public void queueDeclareAutoDelete() throws Exception |
| { |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| QueueDeclareOkBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().declareAutoDelete(true).declareName(BrokerAdmin.TEST_QUEUE_NAME).declare() |
| .consumeResponse().getLatestResponse(QueueDeclareOkBody.class); |
| |
| assertThat(response.getQueue(), is(equalTo(AMQShortString.valueOf(BrokerAdmin.TEST_QUEUE_NAME)))); |
| } |
| |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| |
| QueueDeclareOkBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().declarePassive(true).declareName(BrokerAdmin.TEST_QUEUE_NAME).declare() |
| .consumeResponse().getLatestResponse(QueueDeclareOkBody.class); |
| assertThat(response.getQueue(), is(equalTo(AMQShortString.valueOf(BrokerAdmin.TEST_QUEUE_NAME)))); |
| |
| final String consumerTag = "lastConsumer"; |
| interaction.basic() |
| .consumeConsumerTag(consumerTag).consumeQueue(BrokerAdmin.TEST_QUEUE_NAME).consume() |
| .consumeResponse(BasicConsumeOkBody.class) |
| .basic().consumeCancelTag(consumerTag).cancel() |
| .consumeResponse().getLatestResponse(BasicCancelOkBody.class); |
| |
| ChannelCloseBody closeResponse = interaction.queue() |
| .declarePassive(true) |
| .declareName(BrokerAdmin.TEST_QUEUE_NAME) |
| .declare() |
| .consumeResponse() |
| .getLatestResponse(ChannelCloseBody.class); |
| assertThat(closeResponse.getReplyCode(), is(equalTo(ErrorCodes.NOT_FOUND))); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.1", |
| description = "The server MUST ignore the auto-delete field if the queue already exists.") |
| @Ignore("The server does not ignore the auto-delete field if the queue already exists.") |
| public void queueDeclareAutoDeletePreexistingQueue() throws Exception |
| { |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| QueueDeclareOkBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().declareName(BrokerAdmin.TEST_QUEUE_NAME).declare() |
| .consumeResponse().getLatestResponse(QueueDeclareOkBody.class); |
| |
| assertThat(response.getQueue(), is(equalTo(AMQShortString.valueOf(BrokerAdmin.TEST_QUEUE_NAME)))); |
| |
| QueueDeclareOkBody passiveResponse = |
| interaction.queue().declareAutoDelete(true).declareName(BrokerAdmin.TEST_QUEUE_NAME).declare() |
| .consumeResponse().getLatestResponse(QueueDeclareOkBody.class); |
| assertThat(passiveResponse.getQueue(), is(equalTo(AMQShortString.valueOf(BrokerAdmin.TEST_QUEUE_NAME)))); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.1", |
| description = "The client MAY NOT attempt to use a queue that was declared as exclusive by another " |
| + "still-open connection.") |
| public void queueDeclareExclusive() throws Exception |
| { |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| QueueDeclareOkBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().declareName(BrokerAdmin.TEST_QUEUE_NAME).declareExclusive(true).declare() |
| .consumeResponse().getLatestResponse(QueueDeclareOkBody.class); |
| |
| assertThat(response.getQueue(), is(equalTo(AMQShortString.valueOf(BrokerAdmin.TEST_QUEUE_NAME)))); |
| |
| try(FrameTransport transport2 = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction2 = transport2.newInteraction(); |
| ConnectionCloseBody closeResponse = interaction2.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().declareName(BrokerAdmin.TEST_QUEUE_NAME).declare() |
| .consumeResponse().getLatestResponse(ConnectionCloseBody.class); |
| /* TODO: 0-91 specification requires 'resource-locked' (405) but server uses (530) */ |
| assertThat(closeResponse.getReplyCode(), anyOf(equalTo(ErrorCodes.NOT_ALLOWED), equalTo(405))); |
| } |
| } |
| |
| try(FrameTransport transport2 = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction2 = transport2.newInteraction(); |
| QueueDeclareOkBody response = interaction2.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().declareName(BrokerAdmin.TEST_QUEUE_NAME).declare() |
| .consumeResponse().getLatestResponse(QueueDeclareOkBody.class); |
| assertThat(response.getQueue(), is(equalTo(AMQShortString.valueOf(BrokerAdmin.TEST_QUEUE_NAME)))); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.1", |
| description = "The queue name MAY be empty, in which case the server MUST create a new queue with a unique " |
| + "generated name and return this to the client in the Declare-Ok method.") |
| public void queueDeclareServerAssignedName() throws Exception |
| { |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| QueueDeclareOkBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().declare() |
| .consumeResponse().getLatestResponse(QueueDeclareOkBody.class); |
| |
| String serverAssignedQueueName = response.getQueue().toString(); |
| assertThat(serverAssignedQueueName, is(not(isEmptyString()))); |
| |
| QueueDeclareOkBody passive = interaction.queue() |
| .declareName(serverAssignedQueueName) |
| .declarePassive(true).declare() |
| .consumeResponse().getLatestResponse(QueueDeclareOkBody.class); |
| assertThat(passive.getQueue(), is(equalTo(AMQShortString.valueOf(serverAssignedQueueName)))); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.9", description = "delete a queue") |
| public void queueDelete() 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 response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().deleteName(BrokerAdmin.TEST_QUEUE_NAME).delete() |
| .consumeResponse().getLatestResponse(QueueDeleteOkBody.class); |
| |
| assertThat(response.getMessageCount(), is(equalTo(1L))); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.9", |
| description = "The client MUST NOT attempt to delete a queue that does not exist.") |
| public void queueDeleteQueueNotFound() throws Exception |
| { |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| ChannelCloseBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().deleteName(BrokerAdmin.TEST_QUEUE_NAME).delete() |
| .consumeResponse().getLatestResponse(ChannelCloseBody.class); |
| |
| assertThat(response.getReplyCode(), is(equalTo(ErrorCodes.NOT_FOUND))); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.9", |
| description = "If [if-unused is] set, the server will only delete the queue if it has no consumers. " |
| + "If the queue has consumers the server does does not delete it but raises a channel " |
| + "exception instead..") |
| public void queueDeleteWithConsumer() throws Exception |
| { |
| getBrokerAdmin().createQueue(BrokerAdmin.TEST_QUEUE_NAME); |
| |
| try(FrameTransport consumerTransport = new FrameTransport(_brokerAddress).connect(); |
| FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final String consumerTag = "A"; |
| final Interaction consumerInteraction = consumerTransport.newInteraction(); |
| final BasicInteraction basicInteraction = consumerInteraction.openAnonymousConnection() |
| .channel() |
| .open() |
| .consumeResponse(ChannelOpenOkBody.class) |
| .basic(); |
| basicInteraction.consumeConsumerTag(consumerTag).consumeQueue(BrokerAdmin.TEST_QUEUE_NAME).consume() |
| .consumeResponse(BasicConsumeOkBody.class); |
| |
| final Interaction deleterInteraction = transport.newInteraction(); |
| ChannelCloseBody response = deleterInteraction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().deleteName(BrokerAdmin.TEST_QUEUE_NAME).deleteIfUnused(true).delete() |
| .consumeResponse().getLatestResponse(ChannelCloseBody.class); |
| |
| assertThat(response.getReplyCode(), is(equalTo(ErrorCodes.IN_USE))); |
| deleterInteraction.channel().closeOk(); |
| |
| basicInteraction.consumeCancelTag(consumerTag).cancel() |
| .consumeResponse().getLatestResponse(BasicCancelOkBody.class); |
| |
| deleterInteraction.channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().deleteName(BrokerAdmin.TEST_QUEUE_NAME).delete() |
| .consumeResponse().getLatestResponse(QueueDeleteOkBody.class); |
| } |
| } |
| |
| @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 |
| { |
| 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(); |
| QueuePurgeOkBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().purgeName(BrokerAdmin.TEST_QUEUE_NAME).purge() |
| .consumeResponse().getLatestResponse(QueuePurgeOkBody.class); |
| |
| assertThat(response.getMessageCount(), is(equalTo(1L))); |
| |
| QueueDeclareOkBody passive = interaction.queue() |
| .declareName(BrokerAdmin.TEST_QUEUE_NAME) |
| .declarePassive(true).declare() |
| .consumeResponse().getLatestResponse(QueueDeclareOkBody.class); |
| assertThat(passive.getQueue(), is(equalTo(AMQShortString.valueOf(BrokerAdmin.TEST_QUEUE_NAME)))); |
| assertThat(passive.getMessageCount(), is(equalTo(0L))); |
| } |
| } |
| |
| @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 |
| { |
| getBrokerAdmin().createQueue(BrokerAdmin.TEST_QUEUE_NAME); |
| |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| String testExchange = "testExchange"; |
| interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .exchange().declareName(testExchange).declare() |
| .consumeResponse(ExchangeDeclareOkBody.class) |
| .queue().bindName(testExchange).bindQueueName(BrokerAdmin.TEST_QUEUE_NAME).bindRoutingKey("rk1").bind() |
| .consumeResponse(QueueBindOkBody.class); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.3", description = "A server MUST allow ignore duplicate bindings") |
| public void queueBindIgnoreDuplicates() throws Exception |
| { |
| getBrokerAdmin().createQueue(BrokerAdmin.TEST_QUEUE_NAME); |
| |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| String testExchange = "testExchange"; |
| interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .exchange().declareName(testExchange).declare() |
| .consumeResponse(ExchangeDeclareOkBody.class) |
| .queue().bindName(testExchange).bindQueueName(BrokerAdmin.TEST_QUEUE_NAME).bindRoutingKey("rk1").bind() |
| .consumeResponse(QueueBindOkBody.class) |
| .queue().bindName(testExchange).bindQueueName(BrokerAdmin.TEST_QUEUE_NAME).bindRoutingKey("rk1").bind() |
| .consumeResponse(QueueBindOkBody.class); |
| |
| ExchangeBoundOkBody response = interaction.exchange() |
| .boundExchangeName(testExchange) |
| .bound() |
| .consumeResponse() |
| .getLatestResponse(ExchangeBoundOkBody.class); |
| assertThat(response.getReplyCode(), is(equalTo(ExchangeBoundOkBody.OK))); |
| |
| interaction.queue() |
| .unbindName(testExchange) |
| .unbindQueueName(BrokerAdmin.TEST_QUEUE_NAME) |
| .unbindRoutingKey("rk1") |
| .unbind() |
| .consumeResponse(QueueUnbindOkBody.class); |
| |
| response = interaction.exchange() |
| .boundExchangeName(testExchange) |
| .bound() |
| .consumeResponse() |
| .getLatestResponse(ExchangeBoundOkBody.class); |
| assertThat(response.getReplyCode(), is(equalTo(ExchangeBoundOkBody.NO_BINDINGS))); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.3", |
| description = "The client MUST NOT attempt to bind a queue that does not exist.") |
| public void queueBindUnknownQueue() throws Exception |
| { |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| String testExchange = "testExchange"; |
| final Interaction interaction = transport.newInteraction(); |
| ChannelCloseBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .exchange().declareName(testExchange).declare() |
| .consumeResponse(ExchangeDeclareOkBody.class) |
| .queue().bindName(testExchange).bindQueueName(BrokerAdmin.TEST_QUEUE_NAME) |
| .bind() |
| .consumeResponse().getLatestResponse(ChannelCloseBody.class); |
| assertThat(response.getReplyCode(), is(equalTo(ErrorCodes.NOT_FOUND))); |
| } |
| } |
| |
| @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 |
| { |
| String testExchange = "testExchange"; |
| String testRoutingKey = "rk1"; |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().declareName(BrokerAdmin.TEST_QUEUE_NAME).declareDurable(true).declare() |
| .consumeResponse(QueueDeclareOkBody.class) |
| .exchange().declareName(testExchange).declareDurable(true).declare() |
| .consumeResponse(ExchangeDeclareOkBody.class) |
| .queue().bindName(testExchange).bindQueueName(BrokerAdmin.TEST_QUEUE_NAME).bindRoutingKey(testRoutingKey) |
| .bind() |
| .consumeResponse(QueueBindOkBody.class); |
| |
| ExchangeBoundOkBody response = interaction.exchange() |
| .boundExchangeName(testExchange) |
| .boundQueue(BrokerAdmin.TEST_QUEUE_NAME) |
| .boundRoutingKey(testRoutingKey) |
| .bound() |
| .consumeResponse() |
| .getLatestResponse(ExchangeBoundOkBody.class); |
| assertThat(response.getReplyCode(), is(equalTo(ExchangeBoundOkBody.OK))); |
| } |
| |
| assumeThat(getBrokerAdmin().supportsRestart(), Matchers.is(true)); |
| getBrokerAdmin().restart(); |
| |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| ExchangeBoundOkBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .exchange() |
| .boundExchangeName(testExchange) |
| .boundQueue(BrokerAdmin.TEST_QUEUE_NAME) |
| .boundRoutingKey(testRoutingKey) |
| .bound() |
| .consumeResponse() |
| .getLatestResponse(ExchangeBoundOkBody.class); |
| assertThat(response.getReplyCode(), is(equalTo(ExchangeBoundOkBody.OK))); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.3", |
| description = "The server MUST allow a durable queue to bind to a transient exchange.") |
| public void queueBindDurableQueueToTransientExchange() throws Exception |
| { |
| String testExchange = "testExchange"; |
| String testRoutingKey = "rk1"; |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue().declareName(BrokerAdmin.TEST_QUEUE_NAME).declareDurable(true).declare() |
| .consumeResponse(QueueDeclareOkBody.class) |
| .exchange().declareName(testExchange).declareDurable(false).declare() |
| .consumeResponse(ExchangeDeclareOkBody.class) |
| .queue().bindName(testExchange).bindQueueName(BrokerAdmin.TEST_QUEUE_NAME).bindRoutingKey(testRoutingKey) |
| .bind() |
| .consumeResponse(QueueBindOkBody.class); |
| |
| ExchangeBoundOkBody response = interaction.exchange() |
| .boundExchangeName(testExchange) |
| .boundQueue(BrokerAdmin.TEST_QUEUE_NAME) |
| .boundRoutingKey(testRoutingKey) |
| .bound() |
| .consumeResponse() |
| .getLatestResponse(ExchangeBoundOkBody.class); |
| assertThat(response.getReplyCode(), is(equalTo(ExchangeBoundOkBody.OK))); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.3", |
| description = "If the queue name is provided but the routing key is empty," |
| + " the server does the binding with that empty routing key.") |
| public void queueBindWithoutRoutingKey() throws Exception |
| { |
| getBrokerAdmin().createQueue(BrokerAdmin.TEST_QUEUE_NAME); |
| |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| String testExchange = "testExchange"; |
| interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .exchange().declareName(testExchange).declare() |
| .consumeResponse(ExchangeDeclareOkBody.class) |
| .queue().bindName(testExchange).bindQueueName(BrokerAdmin.TEST_QUEUE_NAME).bind() |
| .consumeResponse(QueueBindOkBody.class); |
| |
| ExchangeBoundOkBody response = interaction.exchange() |
| .boundExchangeName(testExchange) |
| .boundQueue(BrokerAdmin.TEST_QUEUE_NAME) |
| .boundRoutingKey("") |
| .bound() |
| .consumeResponse() |
| .getLatestResponse(ExchangeBoundOkBody.class); |
| assertThat(response.getReplyCode(), is(equalTo(ExchangeBoundOkBody.OK))); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.5", description = "unbind a queue from an exchange") |
| public void queueUnbind() throws Exception |
| { |
| getBrokerAdmin().createQueue(BrokerAdmin.TEST_QUEUE_NAME); |
| |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| String testExchange = "testExchange"; |
| interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.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).unbindQueueName(BrokerAdmin.TEST_QUEUE_NAME).unbindRoutingKey("rk1").unbind() |
| .consumeResponse(QueueUnbindOkBody.class); |
| } |
| } |
| |
| @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 |
| { |
| getBrokerAdmin().createQueue(BrokerAdmin.TEST_QUEUE_NAME); |
| |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| String testExchange = "testExchange"; |
| ChannelCloseBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.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) |
| .unbindQueueName("unknownQueue") |
| .unbindRoutingKey("rk1") |
| .unbind() |
| .consumeResponse() |
| .getLatestResponse(ChannelCloseBody.class); |
| assertThat(response.getReplyCode(), is(equalTo(ErrorCodes.NOT_FOUND))); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.5", description = "The client MUST NOT attempt to unbind a queue from an " |
| + "exchange that does not exist.") |
| public void queueUnbindUnknownExchange() throws Exception |
| { |
| getBrokerAdmin().createQueue(BrokerAdmin.TEST_QUEUE_NAME); |
| |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| String testExchange = "testExchange"; |
| ChannelCloseBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .exchange().declareName(testExchange).declare() |
| .consumeResponse(ExchangeDeclareOkBody.class) |
| .queue() |
| .bindName(testExchange) |
| .bindQueueName(BrokerAdmin.TEST_QUEUE_NAME) |
| .bindRoutingKey("rk1") |
| .bind() |
| .consumeResponse(QueueBindOkBody.class) |
| .queue() |
| .unbindName("unknownExchange") |
| .unbindQueueName(BrokerAdmin.TEST_QUEUE_NAME) |
| .unbindRoutingKey("rk1") |
| .unbind() |
| .consumeResponse() |
| .getLatestResponse(ChannelCloseBody.class); |
| assertThat(response.getReplyCode(), is(equalTo(ErrorCodes.NOT_FOUND))); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.5", description = "If a unbind fails, the server MUST raise a connection " |
| + "exception") |
| public void queueUnbindUnknownRoutingKey() throws Exception |
| { |
| getBrokerAdmin().createQueue(BrokerAdmin.TEST_QUEUE_NAME); |
| |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| String testExchange = "testExchange"; |
| ChannelCloseBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.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) |
| .unbindQueueName(BrokerAdmin.TEST_QUEUE_NAME) |
| .unbindRoutingKey("rk2") |
| .unbind() |
| .consumeResponse() |
| .getLatestResponse(ChannelCloseBody.class); |
| assertThat(response.getReplyCode(), is(equalTo(ErrorCodes.NOT_FOUND))); |
| } |
| } |
| |
| @Test |
| @SpecificationTest(section = "1.7.2.5", description = "unbind a queue from an exchange") |
| public void queueUnbindWithoutRoutingKey() throws Exception |
| { |
| getBrokerAdmin().createQueue(BrokerAdmin.TEST_QUEUE_NAME); |
| |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| String testExchange = "testExchange"; |
| interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .exchange().declareName(testExchange).declare() |
| .consumeResponse(ExchangeDeclareOkBody.class) |
| .queue().bindName(testExchange).bindQueueName(BrokerAdmin.TEST_QUEUE_NAME).bind() |
| .consumeResponse(QueueBindOkBody.class) |
| .queue().unbindName(testExchange).unbindQueueName(BrokerAdmin.TEST_QUEUE_NAME).unbind() |
| .consumeResponse(QueueUnbindOkBody.class); |
| } |
| } |
| |
| /** Qpid specific extension */ |
| @Test |
| public void queueDeclareWithAlternateExchange() throws Exception |
| { |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final String altExchName = "altExchange"; |
| final Interaction interaction = transport.newInteraction(); |
| interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .exchange() |
| .declareName(altExchName) |
| .declare() |
| .consumeResponse(ExchangeDeclareOkBody.class) |
| .queue() |
| .declareName(BrokerAdmin.TEST_QUEUE_NAME) |
| .declareArguments(Collections.singletonMap("alternateExchange", altExchName)).declare() |
| .consumeResponse(QueueDeclareOkBody.class); |
| |
| ChannelCloseBody inUseResponse = interaction.exchange() |
| .deleteExchangeName(altExchName) |
| .delete() |
| .consumeResponse().getLatestResponse(ChannelCloseBody.class); |
| assertThat(inUseResponse.getReplyCode(), is(equalTo(ErrorCodes.NOT_ALLOWED))); |
| interaction.channel().closeOk(); |
| |
| interaction.channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue() |
| .deleteName(BrokerAdmin.TEST_QUEUE_NAME) |
| .delete() |
| .consumeResponse(QueueDeleteOkBody.class) |
| .exchange() |
| .deleteExchangeName(altExchName) |
| .delete() |
| .consumeResponse(ExchangeDeleteOkBody.class); |
| } |
| } |
| |
| /** Qpid specific extension */ |
| @Test |
| public void queueDeclareWithUnknownAlternateExchange() throws Exception |
| { |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| ConnectionCloseBody response = interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .queue() |
| .declareName(BrokerAdmin.TEST_QUEUE_NAME) |
| .declareArguments(Collections.singletonMap("alternateExchange", "notKnown")).declare() |
| .consumeResponse().getLatestResponse(ConnectionCloseBody.class); |
| assertThat(response.getReplyCode(), is(equalTo(ErrorCodes.NOT_FOUND))); |
| } |
| } |
| |
| /** Qpid specific extension */ |
| @Test |
| public void topicExchangeInstancesAllowRebindWithDifferentArguments() throws Exception |
| { |
| getBrokerAdmin().createQueue(BrokerAdmin.TEST_QUEUE_NAME); |
| |
| final String content = "content"; |
| final String routingKey = "rk1"; |
| try(FrameTransport transport = new FrameTransport(_brokerAddress).connect()) |
| { |
| final Interaction interaction = transport.newInteraction(); |
| final Map<String, Object> messageProps = Collections.singletonMap("prop", 0); |
| interaction.openAnonymousConnection() |
| .channel().open().consumeResponse(ChannelOpenOkBody.class) |
| .tx().select() |
| .consumeResponse(TxSelectOkBody.class) |
| .queue() |
| .bindName(ExchangeDefaults.TOPIC_EXCHANGE_NAME) |
| .bindRoutingKey(routingKey) |
| .bindQueueName(BrokerAdmin.TEST_QUEUE_NAME) |
| .bindArguments(Collections.singletonMap(AMQPFilterTypes.JMS_SELECTOR.getValue(), "prop = 1")) |
| .bind() |
| .consumeResponse(QueueBindOkBody.class) |
| .basic() |
| .publishExchange(ExchangeDefaults.TOPIC_EXCHANGE_NAME) |
| .publishRoutingKey(routingKey) |
| .content(content) |
| .contentHeaderPropertiesHeaders(messageProps) |
| .publishMessage() |
| .tx().commit() |
| .consumeResponse(TxCommitOkBody.class); |
| |
| assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(0))); |
| |
| interaction.queue() |
| .bindName(ExchangeDefaults.TOPIC_EXCHANGE_NAME) |
| .bindRoutingKey(routingKey) |
| .bindQueueName(BrokerAdmin.TEST_QUEUE_NAME) |
| .bindArguments(Collections.singletonMap(AMQPFilterTypes.JMS_SELECTOR.getValue(), "prop = 0")) |
| .bind() |
| .consumeResponse(QueueBindOkBody.class) |
| .basic().publishExchange(ExchangeDefaults.TOPIC_EXCHANGE_NAME) |
| .publishRoutingKey(routingKey) |
| .content(content) |
| .contentHeaderPropertiesHeaders(messageProps) |
| .publishMessage() |
| .tx().commit() |
| .consumeResponse(TxCommitOkBody.class); |
| |
| assertThat(getBrokerAdmin().getQueueDepthMessages(BrokerAdmin.TEST_QUEUE_NAME), is(equalTo(1))); |
| } |
| } |
| } |
