svn commit: r679059 - /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java

[ritchiem]

Author: ritchiem
Date: Wed Jul 23 03:30:26 2008
New Revision: 679059

URL: http://svn.apache.org/viewvc?rev=679059&view=rev
Log:
QPID-1187 : The broker did not correctly handle subscriptions that would 
suspend due to exhaustion of bytes credit. The processQueue loop would spin, 
this fix marks the subscription inactive for that loop in processQueue so it 
will stop processing that subscription and ultimately the whole processQueue 
loop if required.

Modified:
    
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java

Modified: 
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
URL: 
http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java?rev=679059&r1=679058&r2=679059&view=diff
==============================================================================
--- 
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
 (original)
+++ 
incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
 Wed Jul 23 03:30:26 2008
@@ -1397,8 +1397,12 @@
                                             }
                                             done = false;
                                         }
-                                        else
+                                        else // Not enough Credit for message 
and wouldSuspend
                                         {
+                                            //QPID-1187 - Treat the 
subscription as suspended for this message
+                                            // and wait for the message to be 
removed to continue delivery.
+                                            subActive = false;
+
                                             node.addStateChangeListener(new 
QueueEntryListener(sub, node));
                                         }
                                     }