QPID-2682: Move slow consumer disconnection mechanism to the broker

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

12 files changed, 903 insertions, 40 deletions

diff --git a/java/broker/src/main/java/org/apache/qpid/server/configuration/plugin/SlowConsumerDetectionConfiguration.java b/java/broker/src/main/java/org/apache/qpid/server/configuration/plugin/SlowConsumerDetectionConfiguration.java
new file mode 100644
index 0000000000..dd63c9b698
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/configuration/plugin/SlowConsumerDetectionConfiguration.java
@@ -0,0 +1,92 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.configuration.plugin;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.configuration.ConversionException;
+import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
+import org.apache.qpid.server.configuration.plugins.ConfigurationPluginFactory;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+public class SlowConsumerDetectionConfiguration extends ConfigurationPlugin
+{
+    public static class SlowConsumerDetectionConfigurationFactory implements ConfigurationPluginFactory
+    {
+        public ConfigurationPlugin newInstance(String path, Configuration config) throws ConfigurationException
+        {
+            SlowConsumerDetectionConfiguration slowConsumerConfig = new SlowConsumerDetectionConfiguration();
+            slowConsumerConfig.setConfiguration(path, config);
+            return slowConsumerConfig;
+        }
+
+        public List<String> getParentPaths()
+        {
+            return Arrays.asList("virtualhosts.virtualhost.slow-consumer-detection");
+        }
+    }
+
+    //Set Default time unit to seconds
+    TimeUnit _timeUnit = TimeUnit.SECONDS;
+
+    public String[] getElementsProcessed()
+    {
+        return new String[]{"delay",
+                            "timeunit"};
+    }
+
+    public long getDelay()
+    {
+        return getLongValue("delay", 10);
+    }
+
+    public TimeUnit getTimeUnit()
+    {
+        return  _timeUnit;
+    }
+
+    @Override
+    public void validateConfiguration() throws ConfigurationException
+    {
+        validatePositiveLong("delay");
+
+        String timeUnit = getStringValue("timeunit");
+
+        if (timeUnit != null)
+        {
+            try
+            {
+                _timeUnit = TimeUnit.valueOf(timeUnit.toUpperCase());
+            }
+            catch (IllegalArgumentException iae)
+            {
+                throw new ConfigurationException("Unable to configure Slow Consumer Detection invalid TimeUnit:" + timeUnit);
+            }
+        }
+
+        System.out.println("Configured SCDC");
+        System.out.println("Delay:" + getDelay());
+        System.out.println("TimeUnit:" + getTimeUnit());
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/configuration/plugin/SlowConsumerDetectionPolicyConfiguration.java b/java/broker/src/main/java/org/apache/qpid/server/configuration/plugin/SlowConsumerDetectionPolicyConfiguration.java
new file mode 100644
index 0000000000..8e2ecff6fb
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/configuration/plugin/SlowConsumerDetectionPolicyConfiguration.java
@@ -0,0 +1,76 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.configuration.plugin;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
+import org.apache.qpid.server.configuration.plugins.ConfigurationPluginFactory;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class SlowConsumerDetectionPolicyConfiguration extends ConfigurationPlugin
+{
+    public static class SlowConsumerDetectionPolicyConfigurationFactory implements ConfigurationPluginFactory
+    {
+        public ConfigurationPlugin newInstance(String path, Configuration config) throws ConfigurationException
+        {
+            SlowConsumerDetectionPolicyConfiguration slowConsumerConfig = new SlowConsumerDetectionPolicyConfiguration();
+            slowConsumerConfig.setConfiguration(path, config);
+            return slowConsumerConfig;
+        }
+
+        public List<String> getParentPaths()
+        {
+            return Arrays.asList(
+                    "virtualhosts.virtualhost.queues.slow-consumer-detection.policy",
+                    "virtualhosts.virtualhost.queues.queue.slow-consumer-detection.policy",
+                    "virtualhosts.virtualhost.topics.slow-consumer-detection.policy",
+                    "virtualhosts.virtualhost.topics.topic.slow-consumer-detection.policy");
+        }
+    }
+
+    public String[] getElementsProcessed()
+    {
+        return new String[]{"name"};
+    }
+
+    public String getPolicyName()
+    {
+        return getStringValue("name");
+    }
+
+    @Override
+    public void validateConfiguration() throws ConfigurationException
+    {
+        if (getPolicyName() == null)
+        {
+            throw new ConfigurationException("No Slow consumer policy defined.");
+        }
+    }
+
+    @Override
+    public String formatToString()
+    {
+        return "Policy:"+getPolicyName();
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/configuration/plugin/SlowConsumerDetectionQueueConfiguration.java b/java/broker/src/main/java/org/apache/qpid/server/configuration/plugin/SlowConsumerDetectionQueueConfiguration.java
new file mode 100644
index 0000000000..58131760da
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/configuration/plugin/SlowConsumerDetectionQueueConfiguration.java
@@ -0,0 +1,153 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.configuration.plugin;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
+import org.apache.qpid.server.configuration.plugins.ConfigurationPluginFactory;
+import org.apache.qpid.server.plugins.PluginManager;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.slowconsumerdetection.policies.SlowConsumerPolicyPlugin;
+import org.apache.qpid.slowconsumerdetection.policies.SlowConsumerPolicyPluginFactory;
+
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+public class SlowConsumerDetectionQueueConfiguration extends ConfigurationPlugin
+{
+    private SlowConsumerPolicyPlugin _policyPlugin;
+
+    public static class SlowConsumerDetectionQueueConfigurationFactory implements ConfigurationPluginFactory
+    {
+        public ConfigurationPlugin newInstance(String path, Configuration config) throws ConfigurationException
+        {
+            SlowConsumerDetectionQueueConfiguration slowConsumerConfig = new SlowConsumerDetectionQueueConfiguration();
+            slowConsumerConfig.setConfiguration(path, config);
+            return slowConsumerConfig;
+        }
+
+        public List<String> getParentPaths()
+        {
+            return Arrays.asList(
+                    "virtualhosts.virtualhost.queues.slow-consumer-detection",
+                    "virtualhosts.virtualhost.queues.queue.slow-consumer-detection",
+                    "virtualhosts.virtualhost.topics.slow-consumer-detection",
+                    "virtualhosts.virtualhost.topics.topic.slow-consumer-detection");
+        }
+    }
+
+    public String[] getElementsProcessed()
+    {
+        return new String[]{"messageAge",
+                            "depth",
+                            "messageCount"};
+    }
+
+    public long getMessageAge()
+    {
+        return getLongValue("messageAge");
+    }
+
+    public long getDepth()
+    {
+        return getLongValue("depth");
+    }
+
+    public long getMessageCount()
+    {
+        return getLongValue("messageCount");
+    }
+
+    public SlowConsumerPolicyPlugin getPolicy()
+    {
+        return _policyPlugin;
+    }
+
+    @Override
+    public void validateConfiguration() throws ConfigurationException
+    {
+        if (!containsPositiveLong("messageAge") &&
+            !containsPositiveLong("depth") &&
+            !containsPositiveLong("messageCount"))
+        {
+            throw new ConfigurationException("At least one configuration property" +
+                                             "('messageAge','depth' or 'messageCount') must be specified.");
+        }
+
+        SlowConsumerDetectionPolicyConfiguration policyConfig = getConfiguration(SlowConsumerDetectionPolicyConfiguration.class.getName());
+
+        PluginManager pluginManager = ApplicationRegistry.getInstance().getPluginManager();
+        Map<String, SlowConsumerPolicyPluginFactory> factories = pluginManager.getSlowConsumerPlugins();
+
+        if (policyConfig == null)
+        {
+            throw new ConfigurationException("No Slow Consumer Policy specified. Known Policies:" + factories.keySet());
+        }
+
+        if (_logger.isDebugEnabled())
+        {
+            Iterator<?> keys = policyConfig.getConfig().getKeys();
+
+            while (keys.hasNext())
+            {
+                String key = (String) keys.next();
+
+                _logger.debug("Policy Keys:" + key);
+            }
+
+        }
+
+        SlowConsumerPolicyPluginFactory<SlowConsumerPolicyPlugin> pluginFactory = factories.get(policyConfig.getPolicyName().toLowerCase());
+
+        if (pluginFactory == null)
+        {
+            throw new ConfigurationException("Unknown Slow Consumer Policy specified:" + policyConfig.getPolicyName() + " Known Policies:" + factories.keySet());
+        }
+
+        _policyPlugin = pluginFactory.newInstance(policyConfig);
+
+        // Debug the creation of this Config
+        _logger.debug(this);
+    }
+
+    public String formatToString()
+    {
+        StringBuilder sb = new StringBuilder();
+        if (getMessageAge() > 0)
+        {
+            sb.append("Age:").append(getMessageAge()).append(":");
+        }
+        if (getDepth() > 0)
+        {
+            sb.append("Depth:").append(getDepth()).append(":");
+        }
+        if (getMessageCount() > 0)
+        {
+            sb.append("Count:").append(getMessageCount()).append(":");
+        }
+
+        sb.append("Policy[").append(getPolicy()).append("]");
+        return sb.toString();
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java b/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java
index 717f0d1bee..97c43b940b 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/plugins/PluginManager.java
@@ -35,6 +35,9 @@ import org.apache.felix.framework.util.StringMap;
 import org.apache.log4j.Logger;
 import org.apache.qpid.common.Closeable;
 import org.apache.qpid.server.configuration.TopicConfiguration;
+import org.apache.qpid.server.configuration.plugin.SlowConsumerDetectionConfiguration.SlowConsumerDetectionConfigurationFactory;
+import org.apache.qpid.server.configuration.plugin.SlowConsumerDetectionPolicyConfiguration.SlowConsumerDetectionPolicyConfigurationFactory;
+import org.apache.qpid.server.configuration.plugin.SlowConsumerDetectionQueueConfiguration.SlowConsumerDetectionQueueConfigurationFactory;
 import org.apache.qpid.server.configuration.plugins.ConfigurationPluginFactory;
 import org.apache.qpid.server.exchange.ExchangeType;
 import org.apache.qpid.server.security.SecurityManager;
@@ -43,6 +46,9 @@ import org.apache.qpid.server.security.access.plugins.AllowAll;
 import org.apache.qpid.server.security.access.plugins.DenyAll;
 import org.apache.qpid.server.security.access.plugins.LegacyAccess;
 import org.apache.qpid.server.virtualhost.plugins.VirtualHostPluginFactory;
+import org.apache.qpid.server.virtualhost.plugin.SlowConsumerDetection;
+import org.apache.qpid.server.virtualhost.plugin.policies.TopicDeletePolicy;
+import org.apache.qpid.slowconsumerdetection.policies.SlowConsumerPolicyPluginFactory;
 import org.osgi.framework.BundleActivator;
 import org.osgi.framework.BundleException;
 import org.osgi.framework.launch.Framework;
@@ -57,7 +63,6 @@ public class PluginManager implements Closeable
     private static final Logger _logger = Logger.getLogger(PluginManager.class);
 
     private static final int FELIX_STOP_TIMEOUT = 30000;
-    private static final String VERSION = "2.6.0.4";
 
     private Framework _felix;
 
@@ -65,11 +70,14 @@ public class PluginManager implements Closeable
     private ServiceTracker _securityTracker = null;
     private ServiceTracker _configTracker = null;
     private ServiceTracker _virtualHostTracker = null;
+    private ServiceTracker _policyTracker = null;
 
     private Activator _activator;
 
     private Map<String, SecurityPluginFactory> _securityPlugins = new HashMap<String, SecurityPluginFactory>();
     private Map<List<String>, ConfigurationPluginFactory> _configPlugins = new IdentityHashMap<List<String>, ConfigurationPluginFactory>();
+    private Map<String, VirtualHostPluginFactory> _vhostPlugins = new HashMap<String, VirtualHostPluginFactory>();
+    private Map<String, SlowConsumerPolicyPluginFactory> _policyPlugins = new HashMap<String, SlowConsumerPolicyPluginFactory>();
 
     public PluginManager(String pluginPath, String cachePath) throws Exception
     {
@@ -85,10 +93,23 @@ public class PluginManager implements Closeable
                 SecurityManager.SecurityConfiguration.FACTORY,
                 AllowAll.AllowAllConfiguration.FACTORY,
                 DenyAll.DenyAllConfiguration.FACTORY,
-                LegacyAccess.LegacyAccessConfiguration.FACTORY))
+                LegacyAccess.LegacyAccessConfiguration.FACTORY,
+                new SlowConsumerDetectionConfigurationFactory(),
+                new SlowConsumerDetectionPolicyConfigurationFactory(),
+                new SlowConsumerDetectionQueueConfigurationFactory()))
         {
             _configPlugins.put(configFactory.getParentPaths(), configFactory);
         }
+        for (SlowConsumerPolicyPluginFactory pluginFactory : Arrays.asList(
+                new TopicDeletePolicy.TopicDeletePolicyFactory()))
+        {
+            _policyPlugins.put(pluginFactory.getPluginName(), pluginFactory);
+        }
+        for (VirtualHostPluginFactory pluginFactory : Arrays.asList(
+                new SlowConsumerDetection.SlowConsumerFactory()))
+        {
+            _vhostPlugins.put(pluginFactory.getClass().getName(), pluginFactory);
+        }
 
         // Check the plugin directory path is set and exist
         if (pluginPath == null)
@@ -117,6 +138,7 @@ public class PluginManager implements Closeable
                 "org.apache.qpid.common; version=0.7," +
                 "org.apache.qpid.exchange; version=0.7," +
                 "org.apache.qpid.framing; version=0.7," +
+                "org.apache.qpid.management.common.mbeans.annotations; version=0.7," +
                 "org.apache.qpid.protocol; version=0.7," +
                 "org.apache.qpid.server.binding; version=0.7," +
                 "org.apache.qpid.server.configuration; version=0.7," +
@@ -157,7 +179,7 @@ public class PluginManager implements Closeable
         configMap.put(SYSTEMBUNDLE_ACTIVATORS_PROP, activators);
 
         if (cachePath != null)
-            {
+        {
             File cacheDir = new File(cachePath);
             if (!cacheDir.exists() && cacheDir.canWrite())
             {
@@ -204,12 +226,11 @@ public class PluginManager implements Closeable
 
         _virtualHostTracker = new ServiceTracker(_activator.getContext(), VirtualHostPluginFactory.class.getName(), null);
         _virtualHostTracker.open();
-
-        _logger.info("Opened service trackers");
+ 
+        _policyTracker = new ServiceTracker(_activator.getContext(), SlowConsumerPolicyPluginFactory.class.getName(), null);
+        _policyTracker.open();
         
-        // Load security and configuration plugins from their trackers for access
-        _configPlugins.putAll(getConfigurationServices());
-        _securityPlugins.putAll(getPlugins(SecurityPluginFactory.class));
+        _logger.info("Opened service trackers");
     }
 
     private static <T> Map<String, T> getServices(ServiceTracker tracker)
@@ -234,11 +255,18 @@ public class PluginManager implements Closeable
         return services;
     }
 
-    private Map<List<String>, ConfigurationPluginFactory> getConfigurationServices()
+    public static <T> Map<String, T> getServices(ServiceTracker tracker, Map<String, T> plugins)
+    {   
+        Map<String, T> services = getServices(tracker);
+        services.putAll(plugins);
+        return services;
+    }
+
+    public Map<List<String>, ConfigurationPluginFactory> getConfigurationPlugins()
     {   
         Map<List<String>, ConfigurationPluginFactory> services = new IdentityHashMap<List<String>, ConfigurationPluginFactory>();
         
-        if (_configTracker.getServices() != null)
+        if (_configTracker != null && _configTracker.getServices() != null)
         {
             for (Object service : _configTracker.getServices())
             {
@@ -246,49 +274,30 @@ public class PluginManager implements Closeable
                 services.put(factory.getParentPaths(), factory);
             }
         }
+        
+        services.putAll(_configPlugins);
 
         return services;
     }
 
-    public Map<String, ExchangeType<?>> getExchanges()
-    {
-        return getServices(_exchangeTracker);
+    public Map<String, VirtualHostPluginFactory> getVirtualHostPlugins()
+    {   
+        return getServices(_virtualHostTracker, _vhostPlugins);
     }
 
-    public Map<String, VirtualHostPluginFactory> getVirtualHostPlugins()
-    {
-        return getServices(_virtualHostTracker);
+    public Map<String, SlowConsumerPolicyPluginFactory> getSlowConsumerPlugins()
+    {   
+        return getServices(_policyTracker, _policyPlugins);
     }
 
-    public <P extends PluginFactory<?>> Map<String, P> getPlugins(Class<P> plugin)
+    public Map<String, ExchangeType<?>> getExchanges()
     {
-        // If plugins are not configured then return an empty set
-        if (_activator == null)
-        {
-            return new HashMap<String, P>();
-        }
-
-        ServiceTracker tracker = new ServiceTracker(_activator.getContext(), plugin.getName(), null);
-        tracker.open();
-
-        try
-        {
-            return getServices(tracker);
-        }
-        finally
-        {
-            tracker.close();
-        }
+        return getServices(_exchangeTracker);
     }
     
     public Map<String, SecurityPluginFactory> getSecurityPlugins()
     {
-        return _securityPlugins;
-    }
-    
-    public Map<List<String>, ConfigurationPluginFactory> getConfigurationPlugins()
-    {
-        return _configPlugins;
+        return getServices(_securityTracker, _securityPlugins);
     }
 
     public void close()
@@ -302,6 +311,7 @@ public class PluginManager implements Closeable
                 _securityTracker.close();
                 _configTracker.close();
                 _virtualHostTracker.close();
+                _policyTracker.close();
             }
             finally
             {
diff --git a/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/ConfiguredQueueBindingListener.java b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/ConfiguredQueueBindingListener.java
new file mode 100644
index 0000000000..d947e9a367
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/ConfiguredQueueBindingListener.java
@@ -0,0 +1,86 @@
+package org.apache.qpid.server.virtualhost.plugin;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
+
+import org.apache.log4j.Logger;
+import org.apache.qpid.server.binding.Binding;
+import org.apache.qpid.server.configuration.plugin.SlowConsumerDetectionQueueConfiguration;
+import org.apache.qpid.server.exchange.AbstractExchange;
+import org.apache.qpid.server.exchange.Exchange;
+import org.apache.qpid.server.exchange.Exchange.BindingListener;
+import org.apache.qpid.server.queue.AMQQueue;
+
+/**
+ * This is a listener that caches queues that are configured for slow consumer disconnection.
+ * 
+ * There should be one listener per virtual host, which can be added to all exchanges on
+ * that host.
+ * 
+ * TODO In future, it will be possible to configure the policy at runtime, so only the queue
+ * itself is cached, and the configuration looked up by the housekeeping thread. This means
+ * that there may be occasions where the copy of the cache contents retrieved by the thread
+ * does not contain queues that are configured, or that configured queues are not present.
+ * 
+ * @see BindingListener
+ */
+public class ConfiguredQueueBindingListener implements BindingListener
+{
+    private static final Logger _log = Logger.getLogger(ConfiguredQueueBindingListener.class);
+    
+    private String _vhostName;
+    private Set<AMQQueue> _cache = Collections.synchronizedSet(new HashSet<AMQQueue>());
+    
+    public ConfiguredQueueBindingListener(String vhostName)
+    {
+        _vhostName = vhostName;
+    }
+
+    /**
+     * @see BindingListener#bindingAdded(Exchange, Binding)
+     */
+    public void bindingAdded(Exchange exchange, Binding binding)
+    {
+        processBinding(binding);
+    }
+
+    /**
+     * @see BindingListener#bindingRemoved(Exchange, Binding)
+     */
+    public void bindingRemoved(Exchange exchange, Binding binding)
+    {
+        processBinding(binding);
+    }
+    
+    private void processBinding(Binding binding)
+    {
+        AMQQueue queue = binding.getQueue();
+        
+        SlowConsumerDetectionQueueConfiguration config =
+            queue.getConfiguration().getConfiguration(SlowConsumerDetectionQueueConfiguration.class.getName());
+        if (config != null)
+        {
+            _cache.add(queue);
+        }
+        else
+        {
+            _cache.remove(queue);
+        }
+    }
+    
+    /**
+     * Lookup and return the cache of configured {@link AMQQueue}s.
+     * 
+	 * Note that when accessing the cached queues, the {@link Iterator} is not thread safe
+	 * (see the {@link Collections#synchronizedSet(Set)} documentation) so a copy of the
+	 * cache is returned.
+     * 
+     * @return a copy of the cached {@link java.util.Set} of queues
+     */
+    public Set<AMQQueue> getQueueCache()
+    {
+        return new HashSet<AMQQueue>(_cache);
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/SlowConsumerDetection.java b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/SlowConsumerDetection.java
new file mode 100644
index 0000000000..6acb0bc11e
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/SlowConsumerDetection.java
@@ -0,0 +1,161 @@
+/*
+ *
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.virtualhost.plugin;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.server.configuration.plugin.SlowConsumerDetectionConfiguration;
+import org.apache.qpid.server.configuration.plugin.SlowConsumerDetectionQueueConfiguration;
+import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.plugins.Plugin;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.virtualhost.VirtualHost;
+import org.apache.qpid.server.virtualhost.plugin.logging.SlowConsumerDetectionMessages;
+import org.apache.qpid.server.virtualhost.plugins.VirtualHostHouseKeepingPlugin;
+import org.apache.qpid.server.virtualhost.plugins.VirtualHostPluginFactory;
+
+public class SlowConsumerDetection extends VirtualHostHouseKeepingPlugin
+{
+    private SlowConsumerDetectionConfiguration _config;
+    private ConfiguredQueueBindingListener _listener;
+
+    public static class SlowConsumerFactory implements VirtualHostPluginFactory
+    {
+        public SlowConsumerDetection newInstance(VirtualHost vhost)
+        {
+            SlowConsumerDetectionConfiguration config = vhost.getConfiguration().getConfiguration(SlowConsumerDetectionConfiguration.class.getName());
+
+            if (config == null)
+            {
+                return null;
+            }
+
+            SlowConsumerDetection plugin = new SlowConsumerDetection(vhost);
+            plugin.configure(config);
+            return plugin;
+        }
+    }
+
+    /**
+     * Configures the slow consumer disconnect plugin by adding a listener to each exchange on this
+     * cirtual host to record all the configured queues in a cache for processing by the housekeeping
+     * thread.
+     * 
+     * @see Plugin#configure(ConfigurationPlugin)
+     */
+    public void configure(ConfigurationPlugin config)
+    {        
+        _config = (SlowConsumerDetectionConfiguration) config;
+        _listener = new ConfiguredQueueBindingListener(_virtualhost.getName());
+        for (AMQShortString exchangeName : _virtualhost.getExchangeRegistry().getExchangeNames())
+        {
+            _virtualhost.getExchangeRegistry().getExchange(exchangeName).addBindingListener(_listener);
+        }
+    }
+    
+    public SlowConsumerDetection(VirtualHost vhost)
+    {
+        super(vhost);
+    }
+
+    public void execute()
+    {
+        CurrentActor.get().message(SlowConsumerDetectionMessages.RUNNING());
+        
+        Set<AMQQueue> cache = _listener.getQueueCache();
+        for (AMQQueue q : cache)
+        {
+            CurrentActor.get().message(SlowConsumerDetectionMessages.CHECKING_QUEUE(q.getName()));
+            
+            try
+            {
+                SlowConsumerDetectionQueueConfiguration config =
+                    q.getConfiguration().getConfiguration(SlowConsumerDetectionQueueConfiguration.class.getName());
+                if (checkQueueStatus(q, config))
+                {
+                    config.getPolicy().performPolicy(q);
+                }
+            }
+            catch (Exception e)
+            {
+                // Don't throw exceptions as this will stop the house keeping task from running.
+                _logger.error("Exception in SlowConsumersDetection for queue: " + q.getName(), e);
+            }
+        }
+
+        CurrentActor.get().message(SlowConsumerDetectionMessages.COMPLETE());
+    }
+
+    public long getDelay()
+    {
+        return _config.getDelay();
+    }
+
+    public TimeUnit getTimeUnit()
+    {
+        return _config.getTimeUnit();
+    }
+
+    /**
+     * Check the depth,messageSize,messageAge,messageCount values for this q
+     *
+     * @param q      the queue to check
+     * @param config the queue configuration to compare against the queue state
+     *
+     * @return true if the queue has reached a threshold.
+     */
+    private boolean checkQueueStatus(AMQQueue q, SlowConsumerDetectionQueueConfiguration config)
+    {
+        if (config != null)
+        {
+            _logger.info("Retrieved Queue(" + q.getName() + ") Config:" + config);
+
+            int count = q.getMessageCount();
+
+            // First Check message counts
+            if ((config.getMessageCount() != 0 && count >= config.getMessageCount()) ||
+                // The check queue depth
+                (config.getDepth() != 0 && q.getQueueDepth() >= config.getDepth()) ||
+                // finally if we have messages on the queue check Arrival time.
+                // We must check count as OldestArrival time is Long.MAX_LONG when
+                // there are no messages.
+                (config.getMessageAge() != 0 &&
+                 ((count > 0) && q.getOldestMessageArrivalTime() >= config.getMessageAge())))
+            {
+                
+                if (_logger.isDebugEnabled())
+                {
+                    _logger.debug("Detected Slow Consumer on Queue(" + q.getName() + ")");
+                    _logger.debug("Queue Count:" + q.getMessageCount() + ":" + config.getMessageCount());
+                    _logger.debug("Queue Depth:" + q.getQueueDepth() + ":" + config.getDepth());
+                    _logger.debug("Queue Arrival:" + q.getOldestMessageArrivalTime() + ":" + config.getMessageAge());
+                }
+
+                return true;
+            }
+        }
+        return false;
+    }
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/logging/SlowConsumerDetection_logmessages.properties b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/logging/SlowConsumerDetection_logmessages.properties
new file mode 100644
index 0000000000..2714935a71
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/logging/SlowConsumerDetection_logmessages.properties
@@ -0,0 +1,4 @@
+#SlowConsumerDetection.logMessages
+RUNNING = SCD-1001 : Running
+COMPLETE = SCD-1002 : Complete
+CHECKING_QUEUE = SCD-1003 : Checking Status of Queue {0}
\ No newline at end of file
diff --git a/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/logging/TopicDeletePolicy_logmessages.properties b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/logging/TopicDeletePolicy_logmessages.properties
new file mode 100644
index 0000000000..d0f5965c39
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/logging/TopicDeletePolicy_logmessages.properties
@@ -0,0 +1,3 @@
+#TopicDeletePolicy.logMessages
+DELETING_QUEUE = TDP-1001 : Deleting Queue
+DISCONNECTING = TDP-1002 : Disconnecting Session
\ No newline at end of file
diff --git a/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/policies/TopicDeletePolicy.java b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/policies/TopicDeletePolicy.java
new file mode 100644
index 0000000000..3bd4ae8d4e
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/policies/TopicDeletePolicy.java
@@ -0,0 +1,141 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.virtualhost.plugin.policies;
+
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.log4j.Logger;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.protocol.AMQConstant;
+import org.apache.qpid.server.binding.Binding;
+import org.apache.qpid.server.configuration.plugin.SlowConsumerDetectionPolicyConfiguration;
+import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
+import org.apache.qpid.server.exchange.TopicExchange;
+import org.apache.qpid.server.logging.actors.CurrentActor;
+import org.apache.qpid.server.protocol.AMQSessionModel;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.virtualhost.plugin.logging.TopicDeletePolicyMessages;
+import org.apache.qpid.slowconsumerdetection.policies.SlowConsumerPolicyPlugin;
+import org.apache.qpid.slowconsumerdetection.policies.SlowConsumerPolicyPluginFactory;
+
+public class TopicDeletePolicy implements SlowConsumerPolicyPlugin
+{
+    Logger _logger = Logger.getLogger(TopicDeletePolicy.class);
+    private TopicDeletePolicyConfiguration _configuration;
+
+    public static class TopicDeletePolicyFactory implements SlowConsumerPolicyPluginFactory
+    {
+        public TopicDeletePolicy newInstance(ConfigurationPlugin configuration) throws ConfigurationException
+        {
+            TopicDeletePolicyConfiguration config =
+                    configuration.getConfiguration(TopicDeletePolicyConfiguration.class.getName());
+
+            TopicDeletePolicy policy = new TopicDeletePolicy();
+            policy.configure(config);
+            return policy;
+        }
+
+        public String getPluginName()
+        {
+            return "topicdelete";
+        }
+
+        public Class<TopicDeletePolicy> getPluginClass()
+        {
+            return TopicDeletePolicy.class;
+        }
+    }
+
+    public void performPolicy(AMQQueue q)
+    {
+        if (q == null)
+        {
+            return;
+        }
+
+        AMQSessionModel owner = q.getExclusiveOwningSession();
+
+        // Only process exclusive queues
+        if (owner == null)
+        {
+            return;
+        }
+
+        //Only process Topics
+        if (!validateQueueIsATopic(q))
+        {
+            return;
+        }
+
+        try
+        {
+            CurrentActor.get().message(owner.getLogSubject(),TopicDeletePolicyMessages.DISCONNECTING());
+            // Close the consumer . this will cause autoDelete Queues to be purged
+            owner.getConnectionModel().
+                    closeSession(owner, AMQConstant.RESOURCE_ERROR,
+                                 "Consuming to slow.");
+
+            // Actively delete non autoDelete queues if deletePersistent is set
+            if (!q.isAutoDelete() && (_configuration != null && _configuration.deletePersistent()))
+            {
+                CurrentActor.get().message(q.getLogSubject(), TopicDeletePolicyMessages.DELETING_QUEUE());
+                q.delete();
+            }
+
+        }
+        catch (AMQException e)
+        {
+            _logger.warn("Unable to close consumer:" + owner + ", on queue:" + q.getName());
+        }
+
+    }
+
+    /**
+     * Check the queue bindings to validate the queue is bound to the
+     * topic exchange.
+     *
+     * @param q the Queue
+     *
+     * @return true iff Q is bound to a TopicExchange
+     */
+    private boolean validateQueueIsATopic(AMQQueue q)
+    {
+        for (Binding binding : q.getBindings())
+        {
+            if (binding.getExchange() instanceof TopicExchange)
+            {
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+    public void configure(ConfigurationPlugin config)
+    {
+        _configuration = (TopicDeletePolicyConfiguration) config;
+    }
+
+    @Override
+    public String toString()
+    {
+        return "TopicDelete" + (_configuration == null ? "" : "[" + _configuration + "]");
+    }
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/policies/TopicDeletePolicyConfiguration.java b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/policies/TopicDeletePolicyConfiguration.java
new file mode 100644
index 0000000000..e6ad1cbcc3
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/server/virtualhost/plugin/policies/TopicDeletePolicyConfiguration.java
@@ -0,0 +1,81 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.virtualhost.plugin.policies;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.qpid.server.configuration.plugins.ConfigurationPlugin;
+import org.apache.qpid.server.configuration.plugins.ConfigurationPluginFactory;
+
+public class TopicDeletePolicyConfiguration extends ConfigurationPlugin
+{
+
+    public static class TopicDeletePolicyConfigurationFactory
+            implements ConfigurationPluginFactory
+    {
+        public ConfigurationPlugin newInstance(String path,
+                                               Configuration config)
+                throws ConfigurationException
+        {
+            TopicDeletePolicyConfiguration slowConsumerConfig =
+                    new TopicDeletePolicyConfiguration();
+            slowConsumerConfig.setConfiguration(path, config);
+            return slowConsumerConfig;
+        }
+
+        public List<String> getParentPaths()
+        {
+            return Arrays.asList(
+                    "virtualhosts.virtualhost.queues.slow-consumer-detection.policy.topicDelete",
+                    "virtualhosts.virtualhost.queues.queue.slow-consumer-detection.policy.topicDelete",
+                    "virtualhosts.virtualhost.topics.slow-consumer-detection.policy.topicDelete",
+                    "virtualhosts.virtualhost.topics.topic.slow-consumer-detection.policy.topicDelete");
+        }
+    }
+
+    public String[] getElementsProcessed()
+    {
+        return new String[]{"delete-persistent"};
+    }
+
+    @Override
+    public void validateConfiguration() throws ConfigurationException
+    {
+        // No validation required.
+    }
+
+    public boolean deletePersistent()
+    {
+        // If we don't have configuration then we don't deletePersistent Queues 
+        return (hasConfiguration() && contains("delete-persistent"));
+    }
+
+    @Override
+    public String formatToString()
+    {
+        return (deletePersistent()?"delete-durable":"");
+    }
+
+
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/slowconsumerdetection/policies/SlowConsumerPolicyPlugin.java b/java/broker/src/main/java/org/apache/qpid/slowconsumerdetection/policies/SlowConsumerPolicyPlugin.java
new file mode 100644
index 0000000000..7f600abdc9
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/slowconsumerdetection/policies/SlowConsumerPolicyPlugin.java
@@ -0,0 +1,29 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.slowconsumerdetection.policies;
+
+import org.apache.qpid.server.plugins.Plugin;
+import org.apache.qpid.server.queue.AMQQueue;
+
+public interface SlowConsumerPolicyPlugin extends Plugin
+{
+    public void performPolicy(AMQQueue Queue);
+}
diff --git a/java/broker/src/main/java/org/apache/qpid/slowconsumerdetection/policies/SlowConsumerPolicyPluginFactory.java b/java/broker/src/main/java/org/apache/qpid/slowconsumerdetection/policies/SlowConsumerPolicyPluginFactory.java
new file mode 100644
index 0000000000..b2fe6766a6
--- /dev/null
+++ b/java/broker/src/main/java/org/apache/qpid/slowconsumerdetection/policies/SlowConsumerPolicyPluginFactory.java
@@ -0,0 +1,27 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.slowconsumerdetection.policies;
+
+import org.apache.qpid.server.plugins.PluginFactory;
+
+public interface SlowConsumerPolicyPluginFactory<P extends SlowConsumerPolicyPlugin> extends PluginFactory<P>
+{
+}