more testing of http inactivity monitor

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

activemq-optional/src/test/java/org/apache/activemq/bugs/AMQ2764Test.java

@@ -22,13 +22,7 @@ import java.util.Iterator;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
-import javax.jms.Connection;
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
+import javax.jms.*;
 
 import junit.framework.TestCase;
 import org.apache.activemq.ActiveMQConnectionFactory;
@@ -50,6 +44,48 @@ public class AMQ2764Test extends TestCase {
     private Destination destination;
     private ArrayList<Connection> connections = new ArrayList<Connection>();
 
+    public void testInactivityMonitor() throws Exception {
+
+        startBrokerTwo();
+        brokerTwo.waitUntilStarted();
+
+        startBrokerOne();
+        brokerOne.waitUntilStarted();
+
+        Thread.sleep(2000);
+
+        ActiveMQConnectionFactory secondProducerConnectionFactory = createBrokerTwoHttpConnectionFactory();
+        ActiveMQConnectionFactory consumerConnectionFactory = createBrokerOneHttpConnectionFactory();
+
+        MessageConsumer consumer = createConsumer(consumerConnectionFactory);
+        AtomicInteger counter = createConsumerCounter(consumerConnectionFactory);
+        waitForConsumerToArrive(counter);
+
+        Connection connection = secondProducerConnectionFactory.createConnection();
+        Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        MessageProducer producer = session.createProducer(destination);
+        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
+
+
+        final int expectedMessagesReceived = 2000;
+
+        for (int i = 0; i < expectedMessagesReceived; i++) {
+            Message message = session.createMessage();
+            producer.send(message);            
+            LOG.info("sent message " + i);
+        }
+
+        for (int i = 0; i < expectedMessagesReceived; i++) {
+            Message message = consumer.receive(2000);
+            if (message == null) {
+                fail("Didn't receive a message");
+            }
+            LOG.info("received message " + i);
+        }
+
+
+    }
+
     public void testBrokerRestart() throws Exception {
 
         startBrokerTwo();
@@ -141,6 +177,14 @@ public class AMQ2764Test extends TestCase {
         return new ActiveMQConnectionFactory("vm://broker2");
     }
 
+    protected ActiveMQConnectionFactory createBrokerOneHttpConnectionFactory() {
+        return new ActiveMQConnectionFactory("http://localhost:61616");
+    }
+
+    protected ActiveMQConnectionFactory createBrokerTwoHttpConnectionFactory() {
+        return new ActiveMQConnectionFactory("http://localhost:61617");
+    }
+
     protected void setUp() throws Exception {
 
         LOG.info("===============================================================================");

activemq-optional/src/test/resources/org/apache/activemq/bugs/amq2764/reconnect-broker2.xml

@@ -29,6 +29,7 @@
       <!-- configure a low inactivity monitor check time to ensure all messages are received quickly, ask for a response as
       peer will have 30 second inactivity delay so won't satisfy our read check till 30+ delay -->
       <transportConnector uri="http://localhost:61617?transport.readCheckTime=4000&amp;transport.initialDelayTime=4000&amp;transport.keepAliveResponseRequired=true"/>
+      <!--<transportConnector uri="http://localhost:61617"/>-->
     </transportConnectors>
 
   </broker>