activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From gtu...@apache.org
Subject svn commit: r792220 - in /activemq/trunk/activemq-core/src/test/java/org/apache/activemq: broker/virtual/VirtualTopicPubSubTest.java spring/ConsumerBean.java
Date Wed, 08 Jul 2009 16:40:54 GMT
Author: gtully
Date: Wed Jul  8 16:40:54 2009
New Revision: 792220

URL: http://svn.apache.org/viewvc?rev=792220&view=rev
Log:
patched test from: https://issues.apache.org/activemq/browse/AMQ-1763 - still cannot reproduce

Modified:
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/virtual/VirtualTopicPubSubTest.java
    activemq/trunk/activemq-core/src/test/java/org/apache/activemq/spring/ConsumerBean.java

Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/virtual/VirtualTopicPubSubTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/virtual/VirtualTopicPubSubTest.java?rev=792220&r1=792219&r2=792220&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/virtual/VirtualTopicPubSubTest.java
(original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/virtual/VirtualTopicPubSubTest.java
Wed Jul  8 16:40:54 2009
@@ -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.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 {
-        if (connection == null) {
-            connection = createConnection();
-        }
-        connection.start();
+
+    private boolean doneTwice = false;
+
+	public void testVirtualTopicCreation() throws Exception {
+	  doTestVirtualTopicCreation(10);
+	}
+
+	public void doTestVirtualTopicCreation(int total) throws Exception {
 
         ConsumerBean messageList = new ConsumerBean() {
             public synchronized void onMessage(Message message) {
@@ -63,48 +67,61 @@
                         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();

Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/spring/ConsumerBean.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/spring/ConsumerBean.java?rev=792220&r1=792219&r2=792220&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/spring/ConsumerBean.java
(original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/spring/ConsumerBean.java
Wed Jul  8 16:40:54 2009
@@ -41,7 +41,7 @@
 
     /**
      * Constructor, initialized semaphore object.
-     * 
+     *
      * @param semaphore
      */
     public ConsumerBean(Object semaphore) {
@@ -59,7 +59,7 @@
 
     /**
      * Method implemented from MessageListener interface.
-     * 
+     *
      * @param message
      */
     public synchronized void onMessage(Message message) {
@@ -96,7 +96,7 @@
 
     /**
      * Used to wait for a message to arrive given a particular message count.
-     * 
+     *
      * @param messageCount
      */
     public void waitForMessagesToArrive(int messageCount) {
@@ -106,12 +106,12 @@
 
         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 @@
 
     /**
      * Identifies if the message is empty.
-     * 
+     *
      * @return
      */
     protected boolean hasReceivedMessage() {
@@ -149,7 +149,7 @@
 
     /**
      * Identifies if the message count has reached the total size of message.
-     * 
+     *
      * @param messageCount
      * @return
      */



Mime
View raw message