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