Merged revisions 608477,609961,610475,610479,610806,611146 via svnmerge from

https://svn.apache.org/repos/asf/incubator/qpid/branches/M2

........
  r608477 | rgodfrey | 2008-01-03 13:23:04 +0000 (Thu, 03 Jan 2008) | 1 line
  
  QPID-499 : Added per-virtual host timed tasks to inspect queues (with no consumers) for expired messages
........
  r609961 | ritchiem | 2008-01-08 12:59:01 +0000 (Tue, 08 Jan 2008) | 2 lines
  
  QPID-499 : Patch to update the queue size statistics when the Active TTL process runs 
  Removed old single commented out code line from AMQSession.
........
  r610475 | ritchiem | 2008-01-09 17:32:43 +0000 (Wed, 09 Jan 2008) | 1 line
  
  Qpid-723 Added exec to qpid.start
........
  r610479 | ritchiem | 2008-01-09 17:39:54 +0000 (Wed, 09 Jan 2008) | 1 line
  
  Qpid-690 : Provide configurable delay between re-connecion attempts.
........
  r610806 | ritchiem | 2008-01-10 14:41:37 +0000 (Thu, 10 Jan 2008) | 1 line
  
  QPID-690 : Relaxed the timings on failover as Thread.sleep is accurate to 10ms so may finish the sleep 10ms early. Resulting in erratic failures as 9.9s < 10s.
........
  r611146 | ritchiem | 2008-01-11 11:33:31 +0000 (Fri, 11 Jan 2008) | 1 line
  
  Patch by Aidan Skinner to make third constructor public. This is done so that the BDBMessageStore tests can still run with the addition of the VirtualHost reaper thread.
........


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/M2.1@615943 13f79535-47bb-0310-9956-ffa450edef68

11 files changed, 575 insertions, 335 deletions

diff --git a/java/broker/etc/config.xml b/java/broker/etc/config.xml
index b3b6a2877f..6c2d84c5d8 100644
--- a/java/broker/etc/config.xml
+++ b/java/broker/etc/config.xml
@@ -100,6 +100,10 @@
                     <class>org.apache.qpid.server.store.MemoryMessageStore</class>
                 </store>
 
+                <housekeeping>
+                    <expiredMessageCheckPeriod>20000</expiredMessageCheckPeriod>
+                </housekeeping>
+
                 <security>
                     <!-- Need protocol changes to allow this-->
                     <authentication>
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 53c36d9718..e1c1de29bd 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
@@ -26,6 +26,7 @@ import org.apache.qpid.configuration.Configured;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.FieldTable;
 import org.apache.qpid.server.AMQChannel;
+import org.apache.qpid.server.registry.ApplicationRegistry;
 import org.apache.qpid.server.exchange.Exchange;
 import org.apache.qpid.server.management.Managable;
 import org.apache.qpid.server.management.ManagedObject;
@@ -37,6 +38,8 @@ import org.apache.qpid.server.virtualhost.VirtualHost;
 import javax.management.JMException;
 import java.text.MessageFormat;
 import java.util.List;
+import java.util.Timer;
+import java.util.TimerTask;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -154,10 +157,7 @@ public class AMQQueue implements Managable, Comparable
     /** total messages received by the queue since startup. */
     public AtomicLong _totalMessagesReceived = new AtomicLong();
 
-    public int compareTo(Object o)
-    {
-        return _name.compareTo(((AMQQueue) o).getName());
-    }
+
 
     public AMQQueue(AMQShortString name, boolean durable, AMQShortString owner, boolean autoDelete, VirtualHost virtualHost)
             throws AMQException
@@ -950,4 +950,20 @@ public class AMQQueue implements Managable, Comparable
         return new QueueEntry(this, amqMessage);
     }
 
+    public int compareTo(Object o)
+    {
+        return _name.compareTo(((AMQQueue) o).getName());
+    }
+
+
+    public void removeExpiredIfNoSubscribers() throws AMQException
+    {
+        synchronized(_subscribers.getChangeLock())
+        {
+            if(_subscribers.isEmpty())
+            {
+                _deliveryMgr.removeExpired();
+            }
+        }
+    }
 }
diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java b/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
index 3cf2cb9b12..a7cef1b983 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/queue/ConcurrentSelectorDeliveryManager.java
@@ -212,6 +212,30 @@ public class ConcurrentSelectorDeliveryManager implements DeliveryManager
         }
     }
 
+    /**
+     *  NOTE : This method should only be called when there are no active subscribers
+     */
+    public void removeExpired() throws AMQException
+    {
+        _lock.lock();
+
+
+	    for(Iterator<QueueEntry> iter = _messages.iterator(); iter.hasNext();)
+        {
+            QueueEntry entry = iter.next();
+            if(entry.expired())
+            {
+                // fixme: Currently we have to update the total byte size here for the data in the queue  
+                _totalMessageSize.addAndGet(-entry.getSize());
+                _queue.dequeue(_reapingStoreContext,entry);
+                iter.remove();
+            }
+	    }
+
+
+        _lock.unlock();
+    }
+
     /** @return the state of the async processor. */
     public boolean isProcessingAsync()
     {
diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/DeliveryManager.java b/java/broker/src/main/java/org/apache/qpid/server/queue/DeliveryManager.java
index f7f35a9319..1568f58e2e 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/queue/DeliveryManager.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/queue/DeliveryManager.java
@@ -97,4 +97,6 @@ interface DeliveryManager
     long getOldestMessageArrival();
 
     void subscriberHasPendingResend(boolean hasContent, Subscription subscription, QueueEntry msg);
+
+    void removeExpired() throws AMQException;
 }
diff --git a/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionSet.java b/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionSet.java
index b73b8d7e07..b2f8cae8ff 100644
--- a/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionSet.java
+++ b/java/broker/src/main/java/org/apache/qpid/server/queue/SubscriptionSet.java
@@ -37,7 +37,8 @@ class SubscriptionSet implements WeightedSubscriptionManager
 
     /** Used to control the round robin delivery of content */
     private int _currentSubscriber;
-    private final Object _subscriptionsChange = new Object();
+
+    private final Object _changeLock = new Object();
 
 
     /** Accessor for unit tests. */
@@ -48,7 +49,7 @@ class SubscriptionSet implements WeightedSubscriptionManager
 
     public void addSubscriber(Subscription subscription)
     {
-        synchronized (_subscriptionsChange)
+        synchronized (_changeLock)
         {
             _subscriptions.add(subscription);
         }
@@ -66,7 +67,7 @@ class SubscriptionSet implements WeightedSubscriptionManager
         // TODO: possibly need O(1) operation here.
 
         Subscription sub = null;
-        synchronized (_subscriptionsChange)
+        synchronized (_changeLock)
         {
             int subIndex = _subscriptions.indexOf(subscription);
 
@@ -226,4 +227,11 @@ class SubscriptionSet implements WeightedSubscriptionManager
     {
         return _subscriptions.size();
     }
+
+
+    public Object getChangeLock()
+    {
+        return _changeLock;
+    }
+    
 }
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 8d6a26fdbc..9addf83e01 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
@@ -1,260 +1,303 @@
-/*

- *

- * 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;

-

-import javax.management.NotCompliantMBeanException;

-

-import org.apache.commons.configuration.Configuration;

-import org.apache.log4j.Logger;

-import org.apache.qpid.server.AMQBrokerManagerMBean;

-import org.apache.qpid.server.security.access.AccessManager;

-import org.apache.qpid.server.security.access.AccessManagerImpl;

-import org.apache.qpid.server.security.access.Accessable;

-import org.apache.qpid.server.security.auth.manager.PrincipalDatabaseAuthenticationManager;

-import org.apache.qpid.server.security.auth.manager.AuthenticationManager;

-import org.apache.qpid.server.configuration.Configurator;

-import org.apache.qpid.server.exchange.DefaultExchangeFactory;

-import org.apache.qpid.server.exchange.DefaultExchangeRegistry;

-import org.apache.qpid.server.exchange.ExchangeFactory;

-import org.apache.qpid.server.exchange.ExchangeRegistry;

-import org.apache.qpid.server.management.AMQManagedObject;

-import org.apache.qpid.server.management.ManagedObject;

-import org.apache.qpid.server.queue.DefaultQueueRegistry;

-import org.apache.qpid.server.queue.QueueRegistry;

-import org.apache.qpid.server.registry.ApplicationRegistry;

-import org.apache.qpid.server.store.MessageStore;

-

-public class VirtualHost implements Accessable

-{

-    private static final Logger _logger = Logger.getLogger(VirtualHost.class);

-

-

-    private final String _name;

-

-    private QueueRegistry _queueRegistry;

-

-    private ExchangeRegistry _exchangeRegistry;

-

-    private ExchangeFactory _exchangeFactory;

-

-    private MessageStore _messageStore;

-

-    protected VirtualHostMBean _virtualHostMBean;

-

-    private AMQBrokerManagerMBean _brokerMBean;

-

-    private AuthenticationManager _authenticationManager;

-

-    private AccessManager _accessManager;

-

-

-    public void setAccessableName(String name)

-    {

-        _logger.warn("Setting Accessable Name for VirualHost is not allowed. ("

-                     + name + ") ignored remains :" + getAccessableName());

-    }

-

-    public String getAccessableName()

-    {

-        return _name;

-    }

-

-

-    /**

-     * Abstract MBean class. This has some of the methods implemented from management intrerface for exchanges. Any

-     * implementaion of an Exchange MBean should extend this class.

-     */

-    public class VirtualHostMBean extends AMQManagedObject implements ManagedVirtualHost

-    {

-        public VirtualHostMBean() throws NotCompliantMBeanException

-        {

-            super(ManagedVirtualHost.class, "VirtualHost");

-        }

-

-        public String getObjectInstanceName()

-        {

-            return _name.toString();

-        }

-

-        public String getName()

-        {

-            return _name.toString();

-        }

-

-        public VirtualHost getVirtualHost()

-        {

-            return VirtualHost.this;

-        }

-

-

-    } // End of MBean class

-

-    /**

-     * Used for testing only

-     * @param name

-     * @param store

-     * @throws Exception

-     */

-    public VirtualHost(String name, MessageStore store) throws Exception

-    {

-        this(name, null, store);

-    }

-

-    /**

-     * Normal Constructor

-     * @param name

-     * @param hostConfig

-     * @throws Exception

-     */

-    public VirtualHost(String name, Configuration hostConfig) throws Exception

-    {

-        this(name, hostConfig, null);

-    }

-

-    private VirtualHost(String name, Configuration hostConfig, MessageStore store) throws Exception

-    {

-        _name = name;

-

-        _virtualHostMBean = new VirtualHostMBean();

-        // This isn't needed to be registered

-        //_virtualHostMBean.register();

-

-        _queueRegistry = new DefaultQueueRegistry(this);

-        _exchangeFactory = new DefaultExchangeFactory(this);

-        _exchangeFactory.initialise(hostConfig);

-        _exchangeRegistry = new DefaultExchangeRegistry(this);

-

-        if (store != null)

-        {

-            _messageStore = store;

-        }

-        else

-        {

-            if (hostConfig == null)

-            {

-                throw new IllegalAccessException("HostConfig and MessageStore cannot be null");

-            }

-            initialiseMessageStore(hostConfig);

-        }

-

-        _exchangeRegistry.initialise();

-

-        _authenticationManager = new PrincipalDatabaseAuthenticationManager(name, hostConfig);

-

-        _accessManager = new AccessManagerImpl(name, hostConfig);

-

-        _brokerMBean = new AMQBrokerManagerMBean(_virtualHostMBean);

-        _brokerMBean.register();

-    }

-

-    private void initialiseMessageStore(Configuration config) throws Exception

-    {

-        String messageStoreClass = config.getString("store.class");

-

-        Class clazz = Class.forName(messageStoreClass);

-        Object o = clazz.newInstance();

-

-        if (!(o instanceof MessageStore))

-        {

-            throw new ClassCastException("Message store class must implement " + MessageStore.class + ". Class " + clazz +

-                                         " does not.");

-        }

-        _messageStore = (MessageStore) o;

-        _messageStore.configure(this, "store", config);

-    }

-

-

-    public <T> T getConfiguredObject(Class<T> instanceType, Configuration config)

-    {

-        T instance;

-        try

-        {

-            instance = instanceType.newInstance();

-        }

-        catch (Exception e)

-        {

-            _logger.error("Unable to instantiate configuration class " + instanceType + " - ensure it has a public default constructor");

-            throw new IllegalArgumentException("Unable to instantiate configuration class " + instanceType + " - ensure it has a public default constructor", e);

-        }

-        Configurator.configure(instance);

-

-        return instance;

-    }

-

-

-    public String getName()

-    {

-        return _name;

-    }

-

-    public QueueRegistry getQueueRegistry()

-    {

-        return _queueRegistry;

-    }

-

-    public ExchangeRegistry getExchangeRegistry()

-    {

-        return _exchangeRegistry;

-    }

-

-    public ExchangeFactory getExchangeFactory()

-    {

-        return _exchangeFactory;

-    }

-

-    public ApplicationRegistry getApplicationRegistry()

-    {

-        throw new UnsupportedOperationException();

-    }

-

-    public MessageStore getMessageStore()

-    {

-        return _messageStore;

-    }

-

-    public AuthenticationManager getAuthenticationManager()

-    {

-        return _authenticationManager;

-    }

-

-    public AccessManager getAccessManager()

-    {

-        return _accessManager;

-    }

-

-    public void close() throws Exception

-    {

-        if (_messageStore != null)

-        {

-            _messageStore.close();

-        }

-    }

-

-    public ManagedObject getBrokerMBean()

-    {

-        return _brokerMBean;

-    }

-

-    public ManagedObject getManagedObject()

-    {

-        return _virtualHostMBean;

-    }

-}

+/*
+ *
+ * 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;
+
+import javax.management.NotCompliantMBeanException;
+
+import org.apache.commons.configuration.Configuration;
+import org.apache.log4j.Logger;
+import org.apache.qpid.server.AMQBrokerManagerMBean;
+import org.apache.qpid.server.security.access.AccessManager;
+import org.apache.qpid.server.security.access.AccessManagerImpl;
+import org.apache.qpid.server.security.access.Accessable;
+import org.apache.qpid.server.security.auth.manager.PrincipalDatabaseAuthenticationManager;
+import org.apache.qpid.server.security.auth.manager.AuthenticationManager;
+import org.apache.qpid.server.configuration.Configurator;
+import org.apache.qpid.server.exchange.DefaultExchangeFactory;
+import org.apache.qpid.server.exchange.DefaultExchangeRegistry;
+import org.apache.qpid.server.exchange.ExchangeFactory;
+import org.apache.qpid.server.exchange.ExchangeRegistry;
+import org.apache.qpid.server.management.AMQManagedObject;
+import org.apache.qpid.server.management.ManagedObject;
+import org.apache.qpid.server.queue.DefaultQueueRegistry;
+import org.apache.qpid.server.queue.QueueRegistry;
+import org.apache.qpid.server.queue.AMQQueue;
+import org.apache.qpid.server.registry.ApplicationRegistry;
+import org.apache.qpid.server.store.MessageStore;
+import org.apache.qpid.AMQException;
+
+import java.util.Timer;
+import java.util.TimerTask;
+
+public class VirtualHost implements Accessable
+{
+    private static final Logger _logger = Logger.getLogger(VirtualHost.class);
+
+
+    private final String _name;
+
+    private QueueRegistry _queueRegistry;
+
+    private ExchangeRegistry _exchangeRegistry;
+
+    private ExchangeFactory _exchangeFactory;
+
+    private MessageStore _messageStore;
+
+    protected VirtualHostMBean _virtualHostMBean;
+
+    private AMQBrokerManagerMBean _brokerMBean;
+
+    private AuthenticationManager _authenticationManager;
+
+    private AccessManager _accessManager;
+
+    private final Timer _houseKeepingTimer = new Timer("Queue-housekeeping", true);
+     
+    private static final long DEFAULT_HOUSEKEEPING_PERIOD = 30000L;
+    
+    public void setAccessableName(String name)
+    {
+        _logger.warn("Setting Accessable Name for VirualHost is not allowed. ("
+                     + name + ") ignored remains :" + getAccessableName());
+    }
+
+    public String getAccessableName()
+    {
+        return _name;
+    }
+
+
+    /**
+     * Abstract MBean class. This has some of the methods implemented from management intrerface for exchanges. Any
+     * implementaion of an Exchange MBean should extend this class.
+     */
+    public class VirtualHostMBean extends AMQManagedObject implements ManagedVirtualHost
+    {
+        public VirtualHostMBean() throws NotCompliantMBeanException
+        {
+            super(ManagedVirtualHost.class, "VirtualHost");
+        }
+
+        public String getObjectInstanceName()
+        {
+            return _name.toString();
+        }
+
+        public String getName()
+        {
+            return _name.toString();
+        }
+
+        public VirtualHost getVirtualHost()
+        {
+            return VirtualHost.this;
+        }
+
+
+    } // End of MBean class
+
+    /**
+     * Used for testing only
+     * @param name
+     * @param store
+     * @throws Exception
+     */
+    public VirtualHost(String name, MessageStore store) throws Exception
+    {
+        this(name, null, store);
+    }
+
+    /**
+     * Normal Constructor
+     * @param name
+     * @param hostConfig
+     * @throws Exception
+     */
+    public VirtualHost(String name, Configuration hostConfig) throws Exception
+    {
+        this(name, hostConfig, null);
+    }
+
+    public VirtualHost(String name, Configuration hostConfig, MessageStore store) throws Exception
+    {
+        _name = name;
+
+        _virtualHostMBean = new VirtualHostMBean();
+        // This isn't needed to be registered
+        //_virtualHostMBean.register();
+
+        _queueRegistry = new DefaultQueueRegistry(this);
+        _exchangeFactory = new DefaultExchangeFactory(this);
+        _exchangeFactory.initialise(hostConfig);
+        _exchangeRegistry = new DefaultExchangeRegistry(this);
+
+        if (store != null)
+        {
+            _messageStore = store;
+        }
+        else
+        {
+            if (hostConfig == null)
+            {
+                throw new IllegalAccessException("HostConfig and MessageStore cannot be null");
+            }
+            initialiseMessageStore(hostConfig);
+        }
+
+        _exchangeRegistry.initialise();
+
+        _authenticationManager = new PrincipalDatabaseAuthenticationManager(name, hostConfig);
+
+        _accessManager = new AccessManagerImpl(name, hostConfig);
+
+        _brokerMBean = new AMQBrokerManagerMBean(_virtualHostMBean);
+        _brokerMBean.register();
+        initialiseHouseKeeping(hostConfig);
+    }
+
+    private void initialiseHouseKeeping(final Configuration hostConfig)
+    {
+     
+    	long period = hostConfig.getLong("housekeeping.expiredMessageCheckPeriod", DEFAULT_HOUSEKEEPING_PERIOD);
+    
+    	/* add a timer task to iterate over queues, cleaning expired messages from queues with no consumers */
+    	if(period != 0L)
+    	{
+    		class RemoveExpiredMessagesTask extends TimerTask
+    		{
+    			public void run()
+    			{
+    				for(AMQQueue q : _queueRegistry.getQueues())
+    				{
+
+    					try
+    					{
+    						q.removeExpiredIfNoSubscribers();
+    					}
+    					catch (AMQException e)
+    					{
+    						_logger.error("Exception in housekeeping for queue: " + q.getName().toString(),e);
+    						throw new RuntimeException(e);
+    					}
+    				}
+    			}
+    		}
+    		
+    		_houseKeepingTimer.scheduleAtFixedRate(new RemoveExpiredMessagesTask(),
+    				period/2,
+    				period);
+    	}
+    }
+    
+    private void initialiseMessageStore(Configuration config) throws Exception
+    {
+        String messageStoreClass = config.getString("store.class");
+
+        Class clazz = Class.forName(messageStoreClass);
+        Object o = clazz.newInstance();
+
+        if (!(o instanceof MessageStore))
+        {
+            throw new ClassCastException("Message store class must implement " + MessageStore.class + ". Class " + clazz +
+                                         " does not.");
+        }
+        _messageStore = (MessageStore) o;
+        _messageStore.configure(this, "store", config);
+    }
+
+
+    public <T> T getConfiguredObject(Class<T> instanceType, Configuration config)
+    {
+        T instance;
+        try
+        {
+            instance = instanceType.newInstance();
+        }
+        catch (Exception e)
+        {
+            _logger.error("Unable to instantiate configuration class " + instanceType + " - ensure it has a public default constructor");
+            throw new IllegalArgumentException("Unable to instantiate configuration class " + instanceType + " - ensure it has a public default constructor", e);
+        }
+        Configurator.configure(instance);
+
+        return instance;
+    }
+
+
+    public String getName()
+    {
+        return _name;
+    }
+
+    public QueueRegistry getQueueRegistry()
+    {
+        return _queueRegistry;
+    }
+
+    public ExchangeRegistry getExchangeRegistry()
+    {
+        return _exchangeRegistry;
+    }
+
+    public ExchangeFactory getExchangeFactory()
+    {
+        return _exchangeFactory;
+    }
+
+    public ApplicationRegistry getApplicationRegistry()
+    {
+        throw new UnsupportedOperationException();
+    }
+
+    public MessageStore getMessageStore()
+    {
+        return _messageStore;
+    }
+
+    public AuthenticationManager getAuthenticationManager()
+    {
+        return _authenticationManager;
+    }
+
+    public AccessManager getAccessManager()
+    {
+        return _accessManager;
+    }
+
+    public void close() throws Exception
+    {
+        if (_messageStore != null)
+        {
+            _messageStore.close();
+        }
+    }
+
+    public ManagedObject getBrokerMBean()
+    {
+        return _brokerMBean;
+    }
+
+    public ManagedObject getManagedObject()
+    {
+        return _virtualHostMBean;
+    }
+}
diff --git a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
index 42f07f97f9..deaa435d8c 100644
--- a/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
+++ b/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
@@ -39,6 +39,7 @@ import org.apache.qpid.client.message.MessageFactoryRegistry;
 import org.apache.qpid.client.message.UnprocessedMessage;
 import org.apache.qpid.client.protocol.AMQProtocolHandler;
 import org.apache.qpid.client.util.FlowControllingBlockingQueue;
+import org.apache.qpid.client.state.listener.SpecificMethodFrameListener;
 import org.apache.qpid.common.AMQPFilterTypes;
 import org.apache.qpid.framing.*;
 import org.apache.qpid.framing.amqp_0_9.MethodRegistry_0_9;
@@ -2148,6 +2149,70 @@ public class AMQSession extends Closeable implements Session, QueueSession, Topi
         declareExchange(amqd.getExchangeName(), amqd.getExchangeClass(), protocolHandler, nowait);
     }
 
+
+    /**
+     * Returns the number of messages currently queued for the given destination.
+     *
+     * <p/>Note that this operation automatically retries in the event of fail-over.
+     *
+     * @param amqd            The destination to be checked
+     *
+     * @return the number of queued messages.
+     *
+     * @throws AMQException If the queue cannot be declared for any reason.
+     */
+    public long getQueueDepth(final AMQDestination amqd)
+            throws AMQException
+    {
+
+        class QueueDeclareOkHandler extends SpecificMethodFrameListener
+        {
+
+            private long _messageCount;
+            private long _consumerCount;
+
+            public QueueDeclareOkHandler()
+            {
+                super(getChannelId(), QueueDeclareOkBody.class);
+            }
+
+            public boolean processMethod(int channelId, AMQMethodBody frame) //throws AMQException
+            {
+                boolean matches = super.processMethod(channelId, frame);
+                QueueDeclareOkBody declareOk = (QueueDeclareOkBody) frame;
+                _messageCount = declareOk.getMessageCount();
+                _consumerCount = declareOk.getConsumerCount();
+                return matches;
+            }
+
+        }
+
+        return new FailoverNoopSupport<Long, AMQException>(
+                new FailoverProtectedOperation<Long, AMQException>()
+                {
+                    public Long execute() throws AMQException, FailoverException
+                    {
+
+                    	AMQFrame queueDeclare =
+                    		getMethodRegistry().createQueueDeclareBody(getTicket(),
+                    												   amqd.getAMQQueueName(),
+                    												   true,
+                    												   amqd.isDurable(),
+                    												   amqd.isExclusive(),
+                    												   amqd.isAutoDelete(),
+                    												   false,
+                    												   null).generateFrame(_channelId);
+                        QueueDeclareOkHandler okHandler = new QueueDeclareOkHandler();
+                        getProtocolHandler().writeCommandFrameAndWaitForReply(queueDeclare, okHandler);
+
+                        return okHandler._messageCount;
+                    }
+                }, _connection).execute();
+
+    }
+
+
+
     /**
      * Declares the named exchange and type of exchange.
      *
diff --git a/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java b/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java
index e8a220f5e9..c1116ca01e 100644
--- a/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java
+++ b/java/client/src/main/java/org/apache/qpid/client/transport/TransportConnection.java
@@ -283,7 +283,10 @@ public class TransportConnection
     public static void killAllVMBrokers()
     {
         _logger.info("Killing all VM Brokers");
-        _acceptor.unbindAll();
+        if (_acceptor != null)
+        {
+        	_acceptor.unbindAll();
+        }
         synchronized (_inVmPipeAddress)
         {
             _inVmPipeAddress.clear();
diff --git a/java/common/src/main/java/org/apache/qpid/util/ConcurrentLinkedMessageQueueAtomicSize.java b/java/common/src/main/java/org/apache/qpid/util/ConcurrentLinkedMessageQueueAtomicSize.java
index 461cf9591d..633cf4fe3a 100644
--- a/java/common/src/main/java/org/apache/qpid/util/ConcurrentLinkedMessageQueueAtomicSize.java
+++ b/java/common/src/main/java/org/apache/qpid/util/ConcurrentLinkedMessageQueueAtomicSize.java
@@ -215,6 +215,14 @@ public class ConcurrentLinkedMessageQueueAtomicSize<E> extends ConcurrentLinkedQ
                 public void remove()
                 {
                     last.remove();
+                    if(last == _mainIterator)
+                    {
+                        _size.decrementAndGet();
+                    }
+                    else
+                    {
+                        _messageHeadSize.decrementAndGet();                        
+                    }
                 }
             };
     }
diff --git a/java/systests/src/main/java/org/apache/qpid/server/failover/FailoverMethodTest.java b/java/systests/src/main/java/org/apache/qpid/server/failover/FailoverMethodTest.java
index d3f79f84b6..2c293527d9 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/failover/FailoverMethodTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/failover/FailoverMethodTest.java
@@ -1,19 +1,4 @@
-package org.apache.qpid.server.failover;
-
-import junit.framework.TestCase;
-import org.apache.qpid.AMQDisconnectedException;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.client.AMQConnection;
-import org.apache.qpid.client.AMQConnectionURL;
-import org.apache.qpid.client.transport.TransportConnection;
-import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
-import org.apache.qpid.url.URLSyntaxException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.jms.ExceptionListener;
-import javax.jms.JMSException;
-import java.util.concurrent.CountDownLatch;/*
+/*
  *
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
@@ -33,26 +18,51 @@ import java.util.concurrent.CountDownLatch;/*
  * under the License.
  *
  */
+package org.apache.qpid.server.failover;
+
+import junit.framework.TestCase;
+import org.apache.qpid.AMQDisconnectedException;
+import org.apache.qpid.AMQException;
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.AMQConnectionURL;
+import org.apache.qpid.client.transport.TransportConnection;
+import org.apache.qpid.client.vmbroker.AMQVMBrokerCreationException;
+import org.apache.qpid.url.URLSyntaxException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.jms.ExceptionListener;
+import javax.jms.JMSException;
+import java.util.concurrent.CountDownLatch;
 
 public class FailoverMethodTest extends TestCase implements ExceptionListener
 {
-    private static final Logger _logger = LoggerFactory.getLogger(FailoverMethodTest.class);
     private CountDownLatch _failoverComplete = new CountDownLatch(1);
 
     public void setUp() throws AMQVMBrokerCreationException
     {
+        TransportConnection.createVMBroker(1);
     }
 
     public void tearDown() throws AMQVMBrokerCreationException
     {
+        TransportConnection.killAllVMBrokers();
     }
 
-    public void testFailoverRoundRobinDelay() throws URLSyntaxException, AMQVMBrokerCreationException, InterruptedException, JMSException
+    /**
+     * Test that the round robin method has the correct delays.
+     * The first connection to vm://:1 will work but the localhost connection should fail but the duration it takes
+     * to report the failure is what is being tested.
+     *
+     * @throws URLSyntaxException
+     * @throws InterruptedException
+     * @throws JMSException
+     */
+    public void testFailoverRoundRobinDelay() throws URLSyntaxException, InterruptedException, JMSException
     {
         String connectionString = "amqp://guest:guest@/test?brokerlist='vm://:1;tcp://localhost:5670?connectdelay='2000',retries='3''";
 
         AMQConnectionURL url = new AMQConnectionURL(connectionString);
-        TransportConnection.createVMBroker(1);
 
         try
         {
@@ -64,9 +74,15 @@ public class FailoverMethodTest extends TestCase implements ExceptionListener
             TransportConnection.killAllVMBrokers();
 
             _failoverComplete.await();
+
             long end = System.currentTimeMillis();
 
-            assertTrue("Failover took at over 10seconds", (end - start) > 6000);
+            //Failover should take less that 10 seconds.
+            // This is calculated by vm://:1 two retries left after initial connection (4s)
+            // localhost get three retries so (6s) so 10s in total for connection dropping
+            assertTrue("Failover took less than 9.5 seconds:"+(end - start), (end - start) > 9500);
+            // The sleep method is not 100% accurate under windows so with 5 sleeps and a 10ms accuracy then there is
+            // the potential for the tests to finish in 500ms sooner than the predicted 10s.
 
         }
         catch (AMQException e)
@@ -80,7 +96,6 @@ public class FailoverMethodTest extends TestCase implements ExceptionListener
         String connectionString = "amqp://guest:guest@/test?brokerlist='vm://:1?connectdelay='2000',retries='3''";
 
         AMQConnectionURL url = new AMQConnectionURL(connectionString);
-        TransportConnection.createVMBroker(1);
 
         try
         {
@@ -92,9 +107,16 @@ public class FailoverMethodTest extends TestCase implements ExceptionListener
             TransportConnection.killAllVMBrokers();
 
             _failoverComplete.await();
+
             long end = System.currentTimeMillis();
 
-            assertTrue("Failover took at over 10seconds", (end - start) > 6000);
+            //Failover should take less that 10 seconds.
+            // This is calculated by vm://:1 two retries left after initial connection
+            // so 4s in total for connection dropping
+
+            assertTrue("Failover took less than 3.7 seconds", (end - start) > 3700);
+            // The sleep method is not 100% accurate under windows so with 3 sleeps and a 10ms accuracy then there is
+            // the potential for the tests to finish in 300ms sooner than the predicted 4s.
 
         }
         catch (AMQException e)
diff --git a/java/systests/src/main/java/org/apache/qpid/server/queue/TimeToLiveTest.java b/java/systests/src/main/java/org/apache/qpid/server/queue/TimeToLiveTest.java
index 06956ba52f..3afecb47d3 100644
--- a/java/systests/src/main/java/org/apache/qpid/server/queue/TimeToLiveTest.java
+++ b/java/systests/src/main/java/org/apache/qpid/server/queue/TimeToLiveTest.java
@@ -25,7 +25,11 @@ import junit.framework.TestCase;
 import junit.framework.Assert;
 import org.apache.qpid.client.transport.TransportConnection;
 import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.AMQDestination;
 import org.apache.qpid.jndi.PropertiesFileInitialContextFactory;
+import org.apache.qpid.url.URLSyntaxException;
+import org.apache.qpid.AMQException;
 import org.apache.log4j.Logger;
 
 import javax.jms.JMSException;
@@ -38,6 +42,7 @@ import javax.jms.Connection;
 import javax.jms.Message;
 import javax.naming.spi.InitialContextFactory;
 import javax.naming.Context;
+import javax.naming.NamingException;
 import java.util.Hashtable;
 
 
@@ -53,21 +58,37 @@ public class TimeToLiveTest extends TestCase
 
     private final long TIME_TO_LIVE = 1000L;
 
-    Context _context;
-
-    private Connection _clientConnection, _producerConnection;
-
-    private MessageConsumer _consumer;
-    MessageProducer _producer;
-    Session _clientSession, _producerSession;
     private static final int MSG_COUNT = 50;
+    private static final long SERVER_TTL_TIMEOUT = 60000L;
 
     protected void setUp() throws Exception
     {
-        if (BROKER.startsWith("vm://"))
+        super.setUp();
+
+        if (usingInVMBroker())
         {
             TransportConnection.createVMBroker(1);
         }
+
+
+    }
+
+    private boolean usingInVMBroker()
+    {
+        return BROKER.startsWith("vm://");
+    }
+
+    protected void tearDown() throws Exception
+    {
+        if (usingInVMBroker())
+        {
+            TransportConnection.killAllVMBrokers();
+        }
+        super.tearDown();
+    }
+
+    public void testPassiveTTL() throws JMSException, NamingException
+    {
         InitialContextFactory factory = new PropertiesFileInitialContextFactory();
 
         Hashtable<String, String> env = new Hashtable<String, String>();
@@ -75,56 +96,40 @@ public class TimeToLiveTest extends TestCase
         env.put("connectionfactory.connection", "amqp://guest:guest@TTL_TEST_ID" + VHOST + "?brokerlist='" + BROKER + "'");
         env.put("queue.queue", QUEUE);
 
-        _context = factory.getInitialContext(env);
+        Context context = factory.getInitialContext(env);
 
-        Queue queue = (Queue) _context.lookup("queue");
+        Queue queue = (Queue) context.lookup("queue");
 
         //Create Client 1
-        _clientConnection = ((ConnectionFactory) _context.lookup("connection")).createConnection();
+        Connection clientConnection = ((ConnectionFactory) context.lookup("connection")).createConnection();
 
-        _clientSession = _clientConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Session clientSession = clientConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
 
-        _consumer = _clientSession.createConsumer(queue);
+        MessageConsumer consumer = clientSession.createConsumer(queue);
 
         //Create Producer
-        _producerConnection = ((ConnectionFactory) _context.lookup("connection")).createConnection();
+        Connection producerConnection = ((ConnectionFactory) context.lookup("connection")).createConnection();
 
-        _producerConnection.start();
+        producerConnection.start();
 
-        _producerSession = _producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Session producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
 
-        _producer = _producerSession.createProducer(queue);
-    }
+        MessageProducer producer = producerSession.createProducer(queue);
 
-    protected void tearDown() throws Exception
-    {
-        _clientConnection.close();
-
-        _producerConnection.close();
-        super.tearDown();
-        
-        if (BROKER.startsWith("vm://"))
-        {
-            TransportConnection.killAllVMBrokers();
-        }
-    }
-
-    public void test() throws JMSException
-    {
         //Set TTL
         int msg = 0;
-        _producer.send(nextMessage(String.valueOf(msg), true));
+        producer.send(nextMessage(String.valueOf(msg), true, producerSession, producer));
 
-        _producer.setTimeToLive(TIME_TO_LIVE);
+        producer.setTimeToLive(TIME_TO_LIVE);
 
         for (; msg < MSG_COUNT - 2; msg++)
         {
-            _producer.send(nextMessage(String.valueOf(msg), false));
+            producer.send(nextMessage(String.valueOf(msg), false, producerSession, producer));
         }
 
         //Reset TTL
-        _producer.setTimeToLive(0L);
-        _producer.send(nextMessage(String.valueOf(msg), false));
+        producer.setTimeToLive(0L);
+        producer.send(nextMessage(String.valueOf(msg), false, producerSession, producer));
 
          try
         {
@@ -136,31 +141,71 @@ public class TimeToLiveTest extends TestCase
 
         }
 
-        _clientConnection.start();
+        clientConnection.start();
 
         //Receive Message 0
-        Message received = _consumer.receive(100);
+        Message received = consumer.receive(100);
         Assert.assertNotNull("First message not received", received);
         Assert.assertTrue("First message doesn't have first set.", received.getBooleanProperty("first"));
         Assert.assertEquals("First message has incorrect TTL.", 0L, received.getLongProperty("TTL"));
 
 
-        received = _consumer.receive(100);
+        received = consumer.receive(100);
         Assert.assertNotNull("Final message not received", received);
         Assert.assertFalse("Final message has first set.", received.getBooleanProperty("first"));
         Assert.assertEquals("Final message has incorrect TTL.", 0L, received.getLongProperty("TTL"));
 
-        received = _consumer.receive(100);
+        received = consumer.receive(100);
         Assert.assertNull("More messages received", received);
+
+        clientConnection.close();
+
+        producerConnection.close();
     }
 
-    private Message nextMessage(String msg, boolean first) throws JMSException
+    private Message nextMessage(String msg, boolean first, Session producerSession, MessageProducer producer) throws JMSException
     {
-        Message send = _producerSession.createTextMessage("Message " + msg);
+        Message send = producerSession.createTextMessage("Message " + msg);
         send.setBooleanProperty("first", first);
-        send.setLongProperty("TTL", _producer.getTimeToLive());
+        send.setLongProperty("TTL", producer.getTimeToLive());
         return send;
     }
 
 
+    /**
+     * Tests the expired messages get actively deleted even on queues which have no consumers
+     */
+    public void testActiveTTL() throws URLSyntaxException, AMQException, JMSException, InterruptedException
+    {
+        Connection producerConnection = new AMQConnection(BROKER,"guest","guest","activeTTLtest","test");
+        AMQSession producerSession = (AMQSession) producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        Queue queue = producerSession.createTemporaryQueue();
+        producerSession.declareAndBind((AMQDestination) queue);
+        MessageProducer producer = producerSession.createProducer(queue);
+        producer.setTimeToLive(1000L);
+
+        // send Messages
+        for(int i = 0; i < MSG_COUNT; i++)
+        {
+            producer.send(producerSession.createTextMessage("Message: "+i));
+        }
+        long failureTime = System.currentTimeMillis() + 2*SERVER_TTL_TIMEOUT;
+
+        // check Queue depth for up to TIMEOUT seconds
+        long messageCount;
+
+        do
+        {
+            Thread.sleep(100);
+            messageCount = producerSession.getQueueDepth((AMQDestination) queue);
+        }
+        while(messageCount > 0L && System.currentTimeMillis() < failureTime);
+
+        assertEquals("Messages not automatically expired: ", 0L, messageCount);
+
+        producer.close();
+        producerSession.close();
+        producerConnection.close();
+    }
+
 }