diff options
3 files changed, 18 insertions, 2 deletions
diff --git a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java index 90d7109df8..03ccbe7ce4 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java +++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java index 29c3f68286..1184ba1d19 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java +++ b/qpid/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/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java b/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java index b25a56344e..71f6c8ed44 100644 --- a/qpid/java/broker/src/main/java/org/apache/qpid/server/virtualhost/VirtualHost.java +++ b/qpid/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(); |