patched test from: https://issues.apache.org/activemq/browse/AMQ-1763 - still cannot reproduce

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

activemq-core/src/test/java/org/apache/activemq/broker/virtual/VirtualTopicPubSubTest.java

@@ -16,6 +16,8 @@
  */
 package org.apache.activemq.broker.virtual;
 
+import java.util.Vector;
+
 import javax.jms.Connection;
 import javax.jms.JMSException;
 import javax.jms.Message;
@@ -31,27 +33,29 @@ import org.apache.activemq.command.ActiveMQTopic;
 import org.apache.activemq.spring.ConsumerBean;
 
 /**
- * 
+ *
  * @version $Revision: $
  */
 public class VirtualTopicPubSubTest extends EmbeddedBrokerTestSupport {
 
-    private Connection connection;
+    private Vector<Connection> connections = new Vector<Connection>();
     public int ackMode = Session.AUTO_ACKNOWLEDGE;
 
     public static Test suite() {
         return suite(VirtualTopicPubSubTest.class);
     }
-    
+
     public void initCombosForTestVirtualTopicCreation() {
         addCombinationValues("ackMode", new Object[] {new Integer(Session.AUTO_ACKNOWLEDGE), new Integer(Session.CLIENT_ACKNOWLEDGE) });
     }
-    
+
-    public void testVirtualTopicCreation() throws Exception {
+    private boolean doneTwice = false;
-        if (connection == null) {
+
-            connection = createConnection();
+	public void testVirtualTopicCreation() throws Exception {
-        }
+	  doTestVirtualTopicCreation(10);
-        connection.start();
+	}
+
+	public void doTestVirtualTopicCreation(int total) throws Exception {
 
         ConsumerBean messageList = new ConsumerBean() {
             public synchronized void onMessage(Message message) {
@@ -63,48 +67,61 @@ public class VirtualTopicPubSubTest extends EmbeddedBrokerTestSupport {
                         e.printStackTrace();
                     }
                 }
-                
+
             }
         };
         messageList.setVerbose(true);
-        
+
         String queueAName = getVirtualTopicConsumerName();
         // create consumer 'cluster'
         ActiveMQQueue queue1 = new ActiveMQQueue(queueAName);
         ActiveMQQueue queue2 = new ActiveMQQueue(queueAName);
-
+  
-        Session session = connection.createSession(false, ackMode);
+        Session session = createStartAndTrackConnection().createSession(false, ackMode);
         MessageConsumer c1 = session.createConsumer(queue1);
+         
+        session = createStartAndTrackConnection().createSession(false, ackMode);
         MessageConsumer c2 = session.createConsumer(queue2);
 
         c1.setMessageListener(messageList);
         c2.setMessageListener(messageList);
 
         // create topic producer
-        MessageProducer producer = session.createProducer(new ActiveMQTopic(getVirtualTopicName()));
+        Session producerSession = createStartAndTrackConnection().createSession(false, ackMode);
+        MessageProducer producer = producerSession.createProducer(new ActiveMQTopic(getVirtualTopicName()));
         assertNotNull(producer);
 
-        int total = 10;
         for (int i = 0; i < total; i++) {
-            producer.send(session.createTextMessage("message: " + i));
+            producer.send(producerSession.createTextMessage("message: " + i));
         }
-        
+
         messageList.assertMessagesArrived(total);
+
+        // do twice so we confirm messages do not get redelivered after client acknowledgement
+        if( doneTwice == false ) {
+            doneTwice = true;
+            doTestVirtualTopicCreation(0);
+		}
     }
 
+    private Connection createStartAndTrackConnection() throws Exception {
+        Connection connection = createConnection();
+        connection.start();
+        connections.add(connection);
+        return connection;
+    }
 
     protected String getVirtualTopicName() {
         return "VirtualTopic.TEST";
     }
 
-
     protected String getVirtualTopicConsumerName() {
         return "Consumer.A.VirtualTopic.TEST";
     }
 
 
     protected void tearDown() throws Exception {
-        if (connection != null) {
+        for (Connection connection: connections) {
             connection.close();
         }
         super.tearDown();

activemq-core/src/test/java/org/apache/activemq/spring/ConsumerBean.java

@@ -41,7 +41,7 @@ public class ConsumerBean extends Assert implements MessageListener {
 
     /**
      * Constructor, initialized semaphore object.
-     * 
+     *
      * @param semaphore
      */
     public ConsumerBean(Object semaphore) {
@@ -59,7 +59,7 @@ public class ConsumerBean extends Assert implements MessageListener {
 
     /**
      * Method implemented from MessageListener interface.
-     * 
+     *
      * @param message
      */
     public synchronized void onMessage(Message message) {
@@ -96,7 +96,7 @@ public class ConsumerBean extends Assert implements MessageListener {
 
     /**
      * Used to wait for a message to arrive given a particular message count.
-     * 
+     *
      * @param messageCount
      */
     public void waitForMessagesToArrive(int messageCount) {
@@ -106,12 +106,12 @@ public class ConsumerBean extends Assert implements MessageListener {
 
         for (int i = 0; i < 10; i++) {
             try {
-                if (hasReceivedMessages(messageCount)) {
-                    break;
-                }
                 synchronized (semaphore) {
                     semaphore.wait(1000);
                 }
+                if (hasReceivedMessages(messageCount)) {
+                    break;
+                }
             } catch (InterruptedException e) {
                 LOG.info("Caught: " + e);
             }
@@ -140,7 +140,7 @@ public class ConsumerBean extends Assert implements MessageListener {
 
     /**
      * Identifies if the message is empty.
-     * 
+     *
      * @return
      */
     protected boolean hasReceivedMessage() {
@@ -149,7 +149,7 @@ public class ConsumerBean extends Assert implements MessageListener {
 
     /**
      * Identifies if the message count has reached the total size of message.
-     * 
+     *
      * @param messageCount
      * @return
      */