qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ai...@apache.org
Subject svn commit: r707672 - in /incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server: AMQChannel.java queue/SimpleAMQQueue.java
Date Fri, 24 Oct 2008 16:20:17 GMT
Author: aidan
Date: Fri Oct 24 09:20:16 2008
New Revision: 707672

URL: http://svn.apache.org/viewvc?rev=707672&view=rev
Log:
QPID-1315: Fix style issue, iterator control usage as per review comments from rgodfrey.

Modified:
    incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
    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/AMQChannel.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java?rev=707672&r1=707671&r2=707672&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
(original)
+++ incubator/qpid/trunk/qpid/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
Fri Oct 24 09:20:16 2008
@@ -333,7 +333,8 @@
         Subscription sub = _tag2SubscriptionMap.remove(consumerTag);
         if (sub != null)
         {
-            try {
+            try 
+            {
                 sub.getSendLock();
                 sub.getQueue().unregisterSubscription(sub);
             }

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=707672&r1=707671&r2=707672&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
Fri Oct 24 09:20:16 2008
@@ -1204,16 +1204,16 @@
         flushSubscription(sub, Long.MAX_VALUE);
     }
 
-    public boolean flushSubscription(Subscription sub, Long deliveries) throws AMQException
+    public boolean flushSubscription(Subscription sub, Long iterations) throws AMQException
     {
         boolean atTail = false;
 
-        while (!sub.isSuspended() && !atTail && deliveries != 0)
+        while (!sub.isSuspended() && !atTail && iterations != 0)
         {
             try 
             {
                 sub.getSendLock();
-                atTail =  attemptDelivery(sub, deliveries);
+                atTail =  attemptDelivery(sub);
                 if (atTail && sub.isAutoClose())
                 {
                     unregisterSubscription(sub);
@@ -1221,6 +1221,10 @@
                     ProtocolOutputConverter converter = sub.getChannel().getProtocolSession().getProtocolOutputConverter();
                     converter.confirmConsumerAutoClose(sub.getChannel().getChannelId(), sub.getConsumerTag());
                 }
+                else if (!atTail)
+                {
+                    iterations--;
+                }
             }
             finally
             {
@@ -1239,7 +1243,7 @@
         return atTail;
     }
 
-    private boolean attemptDelivery(Subscription sub, Long deliveries) throws AMQException
+    private boolean attemptDelivery(Subscription sub) throws AMQException
     {
         boolean atTail = false;
         boolean advanced = false;
@@ -1258,11 +1262,9 @@
                             if (!sub.isBrowser() && !node.acquire(sub))
                             {
                                 sub.restoreCredit(node);
-
                             }
                             else
                             {
-                                deliveries--;
                                 deliverMessage(sub, node);
 
                                 if (sub.isBrowser())
@@ -1352,11 +1354,11 @@
         boolean deliveryIncomplete = true;
 
         int extraLoops = 1;
-        Long deliveries = new Long(MAX_ASYNC_DELIVERIES);
+        Long iterations = new Long(MAX_ASYNC_DELIVERIES);
 
         _asynchronousRunner.compareAndSet(runner, null);
 
-        while (deliveries != 0 && ((previousStateChangeCount != (stateChangeCount
= _stateChangeCount.get())) || deliveryIncomplete) && _asynchronousRunner.compareAndSet(null,
runner))
+        while (iterations != 0 && ((previousStateChangeCount != (stateChangeCount
= _stateChangeCount.get())) || deliveryIncomplete) && _asynchronousRunner.compareAndSet(null,
runner))
         {
             // we want to have one extra loop after every subscription has reached the point
where it cannot move
             // further, just in case the advance of one subscription in the last loop allows
a different subscription to
@@ -1386,7 +1388,7 @@
                         QueueEntry node = moveSubscriptionToNextNode(sub);
                         if (node != null)
                         {
-                            done = attemptDelivery(sub, deliveries);
+                            done = attemptDelivery(sub);
                         }
                     }
                     if (done)
@@ -1409,6 +1411,7 @@
                     }
                     else
                     {
+                        iterations--;
                         extraLoops = 1;
                     }
                 }
@@ -1422,7 +1425,7 @@
 
         // If deliveries == 0 then the limitting factor was the time-slicing rather than
available messages or credit
         // therefore we should schedule this runner again (unless someone beats us to it
:-) ).
-        if (deliveries == 0 && _asynchronousRunner.compareAndSet(null, runner))
+        if (iterations == 0 && _asynchronousRunner.compareAndSet(null, runner))
         {
             _asyncDelivery.execute(runner);
         }



Mime
View raw message