QPID-1315:

   BasicBytesFlowControl doesn't wait long enough to determine if the 3rd message is going to be delivered accidently. It also ack'd every message, which was not it's intent, so use acknowledgeThis() instead.

   Refactor common code out of processQueue and flushSubscription into attemptDelivery.

   Make sure sendLock is held when closing the consumer.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@705657 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 10 insertions, 12 deletions

diff --git a/qpid/java/systests/src/main/java/org/apache/qpid/test/client/FlowControlTest.java b/qpid/java/systests/src/main/java/org/apache/qpid/test/client/FlowControlTest.java
index 910d546034..91ed9766f6 100644
--- a/qpid/java/systests/src/main/java/org/apache/qpid/test/client/FlowControlTest.java
+++ b/qpid/java/systests/src/main/java/org/apache/qpid/test/client/FlowControlTest.java
@@ -21,6 +21,7 @@
 package org.apache.qpid.test.client;
 
 import org.apache.qpid.client.AMQSession_0_8;
+import org.apache.qpid.client.message.AbstractJMSMessage;
 import org.apache.qpid.test.utils.QpidTestCase;
 import org.apache.log4j.Logger;
 
@@ -91,25 +92,22 @@ public class FlowControlTest extends QpidTestCase
         assertNotNull("Second message not received", r2);
         assertEquals("Messages in wrong order", 2, r2.getIntProperty("msg"));
 
-        Message r3 = recv.receiveNoWait();
+        Message r3 = recv.receive(RECEIVE_TIMEOUT);
         assertNull("Third message incorrectly delivered", r3);
 
-        r1.acknowledge();
+        ((AbstractJMSMessage)r1).acknowledgeThis();
 
-        r3 = recv.receiveNoWait();
+        r3 = recv.receive(RECEIVE_TIMEOUT);
         assertNull("Third message incorrectly delivered", r3);
 
-        r2.acknowledge();
+        ((AbstractJMSMessage)r2).acknowledgeThis();
 
         r3 = recv.receive(RECEIVE_TIMEOUT);
         assertNotNull("Third message not received", r3);
         assertEquals("Messages in wrong order", 3, r3.getIntProperty("msg"));
 
-        r3.acknowledge();
-        recv.close();
-        consumerSession.close();
+        ((AbstractJMSMessage)r3).acknowledgeThis();
         consumerConnection.close();
-
     }
 
     public void testTwoConsumersBytesFlowControl() throws Exception
@@ -161,21 +159,21 @@ public class FlowControlTest extends QpidTestCase
         assertNotNull("First message not received", r1);
         assertEquals("Messages in wrong order", 1, r1.getIntProperty("msg"));
 
-        Message r2 = recv1.receiveNoWait();
+        Message r2 = recv1.receive(RECEIVE_TIMEOUT);
         assertNull("Second message incorrectly delivered", r2);
 
         Session consumerSession2 = consumerConnection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
         ((AMQSession_0_8) consumerSession2).setPrefetchLimits(0, 256);
         MessageConsumer recv2 = consumerSession2.createConsumer(_queue);
 
-        r2 = recv2.receive(100000L);//RECEIVE_TIMEOUT);
+        r2 = recv2.receive(RECEIVE_TIMEOUT);
         assertNotNull("Second message not received", r2);
         assertEquals("Messages in wrong order", 2, r2.getIntProperty("msg"));
 
-        Message r3 = recv2.receiveNoWait();
+        Message r3 = recv2.receive(RECEIVE_TIMEOUT);
         assertNull("Third message incorrectly delivered", r3);
 
-        r3 = recv1.receive(100000L);//RECEIVE_TIMEOUT);
+        r3 = recv1.receive(RECEIVE_TIMEOUT);
         assertNotNull("Third message not received", r3);
         assertEquals("Messages in wrong order", 3, r3.getIntProperty("msg"));