svn commit: r1667152 - in /qpid/trunk/qpid/java/broker-plugins: amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ amqp-1-0-protocol/src/main/java/org/apache/qpi...

[rgodfrey]

Author: rgodfrey
Date: Mon Mar 16 23:26:20 2015
New Revision: 1667152

URL: http://svn.apache.org/r1667152
Log:
QPID-6429 : ensure the consumer target queues are emptied before we perform 
async actions like close

Modified:
    
qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
    
qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
    
qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java

Modified: 
qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
URL: 
http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java?rev=1667152&r1=1667151&r2=1667152&view=diff
==============================================================================
--- 
qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
 (original)
+++ 
qpid/trunk/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnection.java
 Mon Mar 16 23:26:20 2015
@@ -718,17 +718,17 @@ public class ServerConnection extends Co
 
     public void processPending()
     {
-        while(_asyncTaskList.peek() != null)
-        {
-            Action<? super ServerConnection> asyncAction = 
_asyncTaskList.poll();
-            asyncAction.performAction(this);
-        }
 
         for (AMQSessionModel session : getSessionModels())
         {
             session.processPending();
         }
 
+        while(_asyncTaskList.peek() != null)
+        {
+            Action<? super ServerConnection> asyncAction = 
_asyncTaskList.poll();
+            asyncAction.performAction(this);
+        }
     }
 
     public void closeAndIgnoreFutureInput()

Modified: 
qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
URL: 
http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java?rev=1667152&r1=1667151&r2=1667152&view=diff
==============================================================================
--- 
qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
 (original)
+++ 
qpid/trunk/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
 Mon Mar 16 23:26:20 2015
@@ -2034,7 +2034,10 @@ public class AMQProtocolEngine implement
     @Override
     public void processPending()
     {
-
+        for (AMQSessionModel session : getSessionModels())
+        {
+            session.processPending();
+        }
 
         while(_asyncTaskList.peek() != null)
         {
@@ -2042,10 +2045,6 @@ public class AMQProtocolEngine implement
             asyncAction.performAction(this);
         }
 
-        for (AMQSessionModel session : getSessionModels())
-        {
-            session.processPending();
-        }
     }
 
     @Override

Modified: 
qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
URL: 
http://svn.apache.org/viewvc/qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java?rev=1667152&r1=1667151&r2=1667152&view=diff
==============================================================================
--- 
qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
 (original)
+++ 
qpid/trunk/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Connection_1_0.java
 Mon Mar 16 23:26:20 2015
@@ -538,15 +538,15 @@ public class Connection_1_0 implements C
 
     public void processPending()
     {
-        while(_asyncTaskList.peek() != null)
+        for (AMQSessionModel session : getSessionModels())
         {
-            Action<? super Connection_1_0> asyncAction = _asyncTaskList.poll();
-            asyncAction.performAction(this);
+            session.processPending();
         }
 
-        for (AMQSessionModel session : getSessionModels())
+        while(_asyncTaskList.peek() != null)
         {
-            session.processPending();
+            Action<? super Connection_1_0> asyncAction = _asyncTaskList.poll();
+            asyncAction.performAction(this);
         }
 
     }



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@qpid.apache.org
For additional commands, e-mail: commits-h...@qpid.apache.org