svn commit: r967052 - in /activemq/trunk/activemq-core/src: main/java/org/apache/activemq/SimplePriorityMessageDispatchChannel.java test/java/org/apache/activemq/transport/failover/FailoverConsumerUnconsumedTest.java

[gtully]

Author: gtully
Date: Fri Jul 23 11:42:31 2010
New Revision: 967052

URL: http://svn.apache.org/viewvc?rev=967052&view=rev
Log:
fix issue with size of SimplePriorityMessageDispatchChannel, cause intermittent 
failure of FailoverConsumerUnconsumedTest

Modified:
    
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/SimplePriorityMessageDispatchChannel.java
    
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverConsumerUnconsumedTest.java

Modified: 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/SimplePriorityMessageDispatchChannel.java
URL: 
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/SimplePriorityMessageDispatchChannel.java?rev=967052&r1=967051&r2=967052&view=diff
==============================================================================
--- 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/SimplePriorityMessageDispatchChannel.java
 (original)
+++ 
activemq/trunk/activemq-core/src/main/java/org/apache/activemq/SimplePriorityMessageDispatchChannel.java
 Fri Jul 23 11:42:31 2010
@@ -216,6 +216,7 @@ public class SimplePriorityMessageDispat
             for (int i = MAX_PRIORITY - 1; i >= 0; i--) {
                 List<MessageDispatch> list = lists[i];
                 result.addAll(list);
+                size -= list.size();
                 list.clear();
             }
             return result;

Modified: 
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverConsumerUnconsumedTest.java
URL: 
http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverConsumerUnconsumedTest.java?rev=967052&r1=967051&r2=967052&view=diff
==============================================================================
--- 
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverConsumerUnconsumedTest.java
 (original)
+++ 
activemq/trunk/activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverConsumerUnconsumedTest.java
 Fri Jul 23 11:42:31 2010
@@ -168,6 +168,20 @@ public class FailoverConsumerUnconsumedT
                
         // will be stopped by the plugin
         broker.waitUntilStopped();
+
+        // verify interrupt
+        assertTrue("add messages dispatched and unconsumed are cleaned up", 
Wait.waitFor(new Wait.Condition() {
+            public boolean isSatisified() throws Exception {
+                int totalUnconsumed = 0;
+                for (TestConsumer testConsumer : testConsumers) {
+                    long unconsumed = testConsumer.unconsumedSize();
+                    LOG.info(testConsumer.getConsumerId() + " unconsumed: " + 
unconsumed);
+                    totalUnconsumed += unconsumed;
+                }
+                return totalUnconsumed == 0;
+            }
+        }));
+
         broker = createBroker(false);
         broker.start();