Author: tabish
Date: Fri Sep 23 13:13:33 2011
New Revision: 1174734

URL: http://svn.apache.org/viewvc?rev=1174734&view=rev
Log:
Remove the dependency on the fixed 61616 port number

Modified:
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2021Test.java

Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2021Test.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2021Test.java?rev=1174734&r1=1174733&r2=1174734&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2021Test.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2021Test.java Fri Sep 23 13:13:33 2011
@@ -43,7 +43,7 @@ import org.slf4j.LoggerFactory;
 
 /**
  * This is a test case for the issue reported at:
- * https://issues.apache.org/activemq/browse/AMQ-2021 
+ * https://issues.apache.org/activemq/browse/AMQ-2021
  * Bug is modification of inflight message properties so the failure can manifest itself in a bunch
  * or ways, from message receipt with null properties to marshall errors
  */
@@ -53,25 +53,26 @@ public class AMQ2021Test extends TestCas
     BrokerService brokerService;
     ArrayList<Thread> threads = new ArrayList<Thread>();
     Vector<Throwable> exceptions;
-    
+
     AMQ2021Test testCase;
-    
-    String ACTIVEMQ_BROKER_BIND = "tcp://localhost:61616";    
-    String ACTIVEMQ_BROKER_URL = ACTIVEMQ_BROKER_BIND + "?jms.redeliveryPolicy.maximumRedeliveries=1&jms.redeliveryPolicy.initialRedeliveryDelay=0";
-    
-    private int numMessages = 1000;
-    private int numConsumers = 2;
-    private int dlqMessages = numMessages/2;
-    
-    CountDownLatch receivedLatch;
+
+    private final String ACTIVEMQ_BROKER_BIND = "tcp://localhost:0";
+    private String CONSUMER_BROKER_URL = "?jms.redeliveryPolicy.maximumRedeliveries=1&jms.redeliveryPolicy.initialRedeliveryDelay=0";
+    private String PRODUCER_BROKER_URL;
+
+    private final int numMessages = 1000;
+    private final int numConsumers = 2;
+    private final int dlqMessages = numMessages/2;
+
+    private CountDownLatch receivedLatch;
     private ActiveMQTopic destination;
-    public CountDownLatch started;
+    private CountDownLatch started;
 
     @Override
     protected void setUp() throws Exception {
         Thread.setDefaultUncaughtExceptionHandler(this);
         testCase = this;
-        
+
         // Start an embedded broker up.
         brokerService = new BrokerService();
         brokerService.setDeleteAllMessagesOnStartup(true);
@@ -79,18 +80,21 @@ public class AMQ2021Test extends TestCas
         brokerService.start();
         destination = new ActiveMQTopic(getName());
         exceptions = new Vector<Throwable>();
-        
-        receivedLatch = 
+
+        CONSUMER_BROKER_URL = brokerService.getTransportConnectors().get(0).getPublishableConnectString() + CONSUMER_BROKER_URL;
+        PRODUCER_BROKER_URL = brokerService.getTransportConnectors().get(0).getPublishableConnectString();
+
+        receivedLatch =
             new CountDownLatch(numConsumers * (numMessages + dlqMessages));
         started = new CountDownLatch(1);
     }
-    
+
     @Override
     protected void tearDown() throws Exception {
         for (Thread t : threads) {
             t.interrupt();
             t.join();
-        }        
+        }
         brokerService.stop();
     }
 
@@ -101,9 +105,9 @@ public class AMQ2021Test extends TestCas
             threads.add(c1);
             c1.start();
         }
-        
+
         assertTrue(started.await(10, TimeUnit.SECONDS));
-        
+
         Thread producer = new Thread() {
             @Override
             public void run() {
@@ -115,34 +119,34 @@ public class AMQ2021Test extends TestCas
         };
         threads.add(producer);
         producer.start();
-        
-        boolean allGood = receivedLatch.await(30, TimeUnit.SECONDS);
+
+        boolean allGood = receivedLatch.await(90, TimeUnit.SECONDS);
         for (Throwable t: exceptions) {
             log.error("failing test with first exception", t);
             fail("exception during test : " + t);
-        }        
+        }
         assertTrue("excepted messages received within time limit", allGood);
-        
+
         assertEquals(0, exceptions.size());
-    
+
         for (int i=0; i<numConsumers; i++) {
             // last recovery sends message to deq so is not received again
             assertEquals(dlqMessages*2, ((ConsumerThread)threads.get(i)).recoveries);
             assertEquals(numMessages + dlqMessages, ((ConsumerThread)threads.get(i)).counter);
         }
-       
+
         // half of the messages for each consumer should go to the dlq but duplicates will
         // be suppressed
         consumeFromDLQ(dlqMessages);
-        
-    }    
-    
+
+    }
+
     private void consumeFromDLQ( int messageCount) throws Exception {
-        ActiveMQConnectionFactory connectionFactory = 
-            new ActiveMQConnectionFactory(ACTIVEMQ_BROKER_URL);
+        ActiveMQConnectionFactory connectionFactory =
+            new ActiveMQConnectionFactory(CONSUMER_BROKER_URL);
         Connection connection = connectionFactory.createConnection();
         connection.start();
-                    
+
         Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
         MessageConsumer dlqConsumer = session.createConsumer(new ActiveMQQueue("ActiveMQ.DLQ"));
         int count = 0;
@@ -158,19 +162,19 @@ public class AMQ2021Test extends TestCas
     public void produce(int count) throws Exception {
         Connection connection=null;
         try {
-            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(ACTIVEMQ_BROKER_BIND);
+            ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(PRODUCER_BROKER_URL);
             connection = factory.createConnection();
             Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
             MessageProducer producer = session.createProducer(destination);
             producer.setTimeToLive(0);
             connection.start();
-            
+
             for (int i=0 ; i< count; i++) {
                 int id = i+1;
                 TextMessage message = session.createTextMessage(getName()+" Message "+ id);
                 message.setIntProperty("MsgNumber", id);
                 producer.send(message);
-                
+
                 if (id % 500 == 0) {
                     log.info("sent " + id + ", ith " + message);
                 }
@@ -187,7 +191,7 @@ public class AMQ2021Test extends TestCas
             }
         }
     }
-    
+
     public class ConsumerThread extends Thread implements MessageListener   {
         public long counter = 0;
         public long recoveries = 0;
@@ -199,24 +203,24 @@ public class AMQ2021Test extends TestCas
 
         public void run() {
             try {
-                ActiveMQConnectionFactory connectionFactory = 
-                    new ActiveMQConnectionFactory(ACTIVEMQ_BROKER_URL);
+                ActiveMQConnectionFactory connectionFactory =
+                    new ActiveMQConnectionFactory(CONSUMER_BROKER_URL);
                 Connection connection = connectionFactory.createConnection();
                 connection.setExceptionListener(testCase);
-                connection.setClientID(getName());            
+                connection.setClientID(getName());
                 session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
-                MessageConsumer consumer = session.createDurableSubscriber(destination, getName());                            
+                MessageConsumer consumer = session.createDurableSubscriber(destination, getName());
                 consumer.setMessageListener(this);
                 connection.start();
-                
+
                 started .countDown();
-            
+
             } catch (JMSException exception) {
                 log.error("unexpected ex in consumer run", exception);
                 exceptions.add(exception);
             }
         }
-        
+
         public void onMessage(Message message) {
             try {
                 counter++;
@@ -226,8 +230,8 @@ public class AMQ2021Test extends TestCas
                     recoveries++;
                 } else {
                     message.acknowledge();
-                }     
-                
+                }
+
                 if (counter % 200 == 0) {
                     log.info("recoveries:" + recoveries + ", Received " + counter + ", counter'th " + message);
                 }
@@ -237,7 +241,7 @@ public class AMQ2021Test extends TestCas
                 exceptions.add(e);
             }
         }
-        
+
     }
 
     public void onException(JMSException exception) {