summaryrefslogtreecommitdiff
path: root/java/systests
diff options
context:
space:
mode:
authorKeith Wall <kwall@apache.org>2012-03-19 14:00:37 +0000
committerKeith Wall <kwall@apache.org>2012-03-19 14:00:37 +0000
commitfa83413bbca3b76e810292939c18ab20b026142e (patch)
treea13f7d9fcccb2ee03181038790538f59a401d714 /java/systests
parent0f111b05b2376ced2c470f30b2d5b700e9ae1b5a (diff)
downloadqpid-python-fa83413bbca3b76e810292939c18ab20b026142e.tar.gz
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 <orudyy@gmail.com>. git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1302455 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'java/systests')
-rw-r--r--java/systests/src/main/java/org/apache/qpid/server/queue/ProducerFlowControlTest.java36
1 files changed, 34 insertions, 2 deletions
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<String,Object> arguments = new HashMap<String, Object>();
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);
}