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

2 files changed, 134 insertions, 67 deletions

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();
+    }
+
 }