You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by gt...@apache.org on 2009/07/08 18:40:54 UTC
svn commit: r792220 - in
/activemq/trunk/activemq-core/src/test/java/org/apache/activemq:
broker/virtual/VirtualTopicPubSubTest.java spring/ConsumerBean.java
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
*/