[jboss-cvs] JBoss Messaging SVN: r3011 - in trunk: tests/src/org/jboss/test/messaging/jms/clustering and 1 other directory.

[jboss-cvs-commits at lists.jboss.org]

Author: timfox
Date: 2007-08-14 13:15:35 -0400 (Tue, 14 Aug 2007)
New Revision: 3011

Modified:
   trunk/src/main/org/jboss/messaging/core/impl/MessagingQueue.java
   trunk/tests/src/org/jboss/test/messaging/jms/clustering/DistributedRequestResponseTest.java
Log:
http://jira.jboss.com/jira/browse/JBMESSAGING-1024


Modified: trunk/src/main/org/jboss/messaging/core/impl/MessagingQueue.java
===================================================================
--- trunk/src/main/org/jboss/messaging/core/impl/MessagingQueue.java	2007-08-14 14:51:55 UTC (rev 3010)
+++ trunk/src/main/org/jboss/messaging/core/impl/MessagingQueue.java	2007-08-14 17:15:35 UTC (rev 3011)
@@ -385,7 +385,7 @@
    	{
 	   	boolean removed = suckers.remove(sucker);
 	   	
-	   	if (removed)
+	   	if (removed && suckers.isEmpty())
 	   	{
 	   		handleFlowControlForConsumers = false;
 	   	}
@@ -511,9 +511,10 @@
 		super.deliverInternal();
 		
 		if (trace) { log.trace(this + " deliverInternal"); }
-   			
+		
 		if (handleFlowControlForConsumers && getReceiversReady() && localDistributor.getNumberOfReceivers() > 0)
 		{
+			if (trace) { log.trace("Informing suckers"); }
 			//The receivers are still ready for more messages but there is nothing left in the local queue
 			//so we inform the message suckers to start consuming (if they aren't already)
 			informSuckers(true);

Modified: trunk/tests/src/org/jboss/test/messaging/jms/clustering/DistributedRequestResponseTest.java
===================================================================
--- trunk/tests/src/org/jboss/test/messaging/jms/clustering/DistributedRequestResponseTest.java	2007-08-14 14:51:55 UTC (rev 3010)
+++ trunk/tests/src/org/jboss/test/messaging/jms/clustering/DistributedRequestResponseTest.java	2007-08-14 17:15:35 UTC (rev 3011)
@@ -32,6 +32,8 @@
 import javax.jms.Session;
 import javax.jms.TextMessage;
 
+import org.jboss.test.messaging.tools.ServerManagement;
+
 /**
  * A test for distributed request-response pattern
  *
@@ -76,6 +78,16 @@
    {
    	distributedRequestResponse(true, false);
    }
+   
+   // http://jira.jboss.com/jira/browse/JBMESSAGING-1024
+   public void testSuckAfterKill() throws Exception
+   {
+   	ServerManagement.kill(2);
+   	
+   	Thread.sleep(3000);
+   	
+   	distributedRequestResponse(false, true);
+   }
 
    // Package protected ----------------------------------------------------------------------------
 
@@ -100,7 +112,7 @@
    {
    	Connection conn0 = null;   	
    	Connection conn1 = null;
-      
+   	
       try
       {      	
          conn0 = this.createConnectionOnServer(cf, 0);
@@ -139,6 +151,7 @@
 				{
 					try
 					{
+						log.info("Received message in listener");
 						Destination dest = msg.getJMSReplyTo();
 						MessageProducer prod = sess.createProducer(dest);
 						prod.setDeliveryMode(persistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);
@@ -181,11 +194,15 @@
             
             prod.send(sm);
             
+            log.info("Sent message");
+            
             TextMessage tm = (TextMessage)cons0.receive(60000);
             
             assertNotNull(tm);
             
             assertEquals(sm.getText() + "reply", tm.getText());
+            
+            log.info("Got reply");
          }   
       }
       finally