From fa83413bbca3b76e810292939c18ab20b026142e Mon Sep 17 00:00:00 2001 From: Keith Wall Date: Mon, 19 Mar 2012 14:00:37 +0000 Subject: QPID-3895: Remove blocked channel/session from the list of blocked channels on channel/session close This patch adds the fllowing: - fixes AMQChannel to stop sending flow commands if channel is closing - fixes AMQChannel#compareTo ServerSession#compareTo - removes AMQSessionModel#getID() method from AMQChannel and Server session in order to avoid confusions Applied patch from Oleksandr Rudyy . git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1302455 13f79535-47bb-0310-9956-ffa450edef68 --- .../qpid/server/queue/ProducerFlowControlTest.java | 36 ++++++++++++++++++++-- 1 file changed, 34 insertions(+), 2 deletions(-) (limited to 'java/systests') diff --git a/java/systests/src/main/java/org/apache/qpid/server/queue/ProducerFlowControlTest.java b/java/systests/src/main/java/org/apache/qpid/server/queue/ProducerFlowControlTest.java index ad8c856a74..13053d02df 100644 --- a/java/systests/src/main/java/org/apache/qpid/server/queue/ProducerFlowControlTest.java +++ b/java/systests/src/main/java/org/apache/qpid/server/queue/ProducerFlowControlTest.java @@ -357,13 +357,45 @@ public class ProducerFlowControlTest extends AbstractTestLogging consumer.receive(); } + public void testQueueDeleteWithBlockedFlow() throws Exception + { + String queueName = getTestQueueName(); + createAndBindQueueWithFlowControlEnabled(producerSession, queueName, 1000, 800, true, false); + + producer = producerSession.createProducer(queue); + + // try to send 5 messages (should block after 4) + sendMessagesAsync(producer, producerSession, 5, 50L); + + Thread.sleep(5000); + + assertEquals("Incorrect number of message sent before blocking", 4, _sentMessages.get()); + + // close blocked producer session and connection + producerConnection.close(); + + // delete queue with a consumer session + ((AMQSession) consumerSession).sendQueueDelete(new AMQShortString(queueName)); + + consumer = consumerSession.createConsumer(queue); + consumerConnection.start(); + + Message message = consumer.receive(1000l); + assertNull("Unexpected message", message); + } + private void createAndBindQueueWithFlowControlEnabled(Session session, String queueName, int capacity, int resumeCapacity) throws Exception + { + createAndBindQueueWithFlowControlEnabled(session, queueName, capacity, resumeCapacity, false, true); + } + + private void createAndBindQueueWithFlowControlEnabled(Session session, String queueName, int capacity, int resumeCapacity, boolean durable, boolean autoDelete) throws Exception { final Map arguments = new HashMap(); arguments.put("x-qpid-capacity",capacity); arguments.put("x-qpid-flow-resume-capacity",resumeCapacity); - ((AMQSession) session).createQueue(new AMQShortString(queueName), true, false, false, arguments); - queue = session.createQueue("direct://amq.direct/"+queueName+"/"+queueName+"?durable='false'&autodelete='true'"); + ((AMQSession) session).createQueue(new AMQShortString(queueName), autoDelete, durable, false, arguments); + queue = session.createQueue("direct://amq.direct/"+queueName+"/"+queueName+"?durable='" + durable + "'&autodelete='" + autoDelete + "'"); ((AMQSession) session).declareAndBind((AMQDestination)queue); } -- cgit v1.2.1