QPID-1266 - Provide a stop() method on AMQQueue to stop all processing on that queue thus shutting down the ThreadPool.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@691263 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 18 insertions, 2 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
index 90d7109df8..03ccbe7ce4 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
@@ -161,6 +161,8 @@ public interface AMQQueue extends Managable, Comparable<AMQQueue>
 
     void deliverAsync();
 
+    void stop();
+
 
     /**
      * ExistingExclusiveSubscription signals a failure to create a subscription, because an exclusive subscription
diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
index 29c3f68286..1184ba1d19 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
@@ -1102,12 +1102,17 @@ public class SimpleAMQQueue implements AMQQueue, Subscription.StateListener
             }
 
             _deleteTaskList.clear();
-            ReferenceCountingExecutorService.getInstance().releaseExecutorService();
+            stop();
         }
         return getMessageCount();
 
     }
 
+    public void stop()
+    {
+        ReferenceCountingExecutorService.getInstance().releaseExecutorService();
+    }
+
     public void deliverAsync()
     {
         _stateChangeCount.incrementAndGet();
diff --git a/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
index b25a56344e..71f6c8ed44 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java
@@ -301,11 +301,20 @@ public class VirtualHost implements Accessable
 
     public void close() throws Exception
     {
+        //Stop the Queues processing
+        if (_queueRegistry != null)
+        {
+            for (AMQQueue queue : _queueRegistry.getQueues())
+            {
+                queue.stop();
+            }
+        }        
+
         //Stop Housekeeping
         if (_houseKeepingTimer != null)
         {
             _houseKeepingTimer.cancel();
-        }
+        }        
 
         //Stop Connections
         _connectionRegistry.close();