broswer consumer close is now pulled by IO rather than pushed by queue, fixing browser tests

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/QPID-6262-JavaBrokerNIO@1659232 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 24 insertions, 3 deletions

diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
index 209f6663ec..a540318452 100644
--- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
@@ -661,4 +661,10 @@ public class ConsumerTarget_0_10 extends AbstractConsumerTarget implements FlowC
     {
         return _unacknowledgedCount.longValue();
     }
+
+    @Override
+    protected void processClosed()
+    {
+
+    }
 }
diff --git a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
index 3659d6ce01..579c885053 100644
--- a/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
+++ b/qpid/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
@@ -1140,7 +1140,7 @@ public class ServerSession extends Session
     {
         for(ConsumerTarget target : getSubscriptions())
         {
-            target.processPendingMessages();
+            target.processPending();
         }
     }
 
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
index be28024d13..c0cc7d55b0 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
@@ -3613,7 +3613,7 @@ public class AMQChannel
 
         for(ConsumerTarget target : _tag2SubscriptionTargetMap.values())
         {
-            target.processPendingMessages();
+            target.processPending();
         }
     }
 }
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
index f9560fa0d2..2bf2fc6d27 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
@@ -75,6 +75,7 @@ public abstract class ConsumerTarget_0_8 extends AbstractConsumerTarget implemen
     private final AtomicLong _unacknowledgedCount = new AtomicLong(0);
     private final AtomicLong _unacknowledgedBytes = new AtomicLong(0);
     private final List<ConsumerImpl> _consumers = new CopyOnWriteArrayList<>();
+    private final AtomicBoolean _needToClose = new AtomicBoolean();
 
 
     public static ConsumerTarget_0_8 createBrowserTarget(AMQChannel channel,
@@ -513,6 +514,15 @@ public abstract class ConsumerTarget_0_8 extends AbstractConsumerTarget implemen
     {
         if (isAutoClose())
         {
+            _needToClose.set(true);
+        }
+    }
+
+    @Override
+    protected void processClosed()
+    {
+        if (_needToClose.get() && getState() != State.CLOSED)
+        {
             close();
             confirmAutoClose();
         }
diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
index ebd23f31ae..f06f70b362 100644
--- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
+++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
@@ -535,4 +535,9 @@ class ConsumerTarget_1_0 extends AbstractConsumerTarget
         return 0;
     }
 
+    @Override
+    protected void processClosed()
+    {
+
+    }
 }
diff --git a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
index dd03469d0f..0b613d9a5a 100644
--- a/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
+++ b/qpid/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
@@ -905,7 +905,7 @@ public class Session_1_0 implements SessionEventListener, AMQSessionModel<Sessio
         for(Consumer<?> consumer : getConsumers())
         {
 
-            ((ConsumerImpl)consumer).getTarget().processPendingMessages();
+            ((ConsumerImpl)consumer).getTarget().processPending();
         }
     }