You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ta...@apache.org on 2013/07/23 18:33:14 UTC
svn commit: r1506149 -
/activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ2021Test.java
Author: tabish
Date: Tue Jul 23 16:33:13 2013
New Revision: 1506149
URL: http://svn.apache.org/r1506149
Log:
convert to JUnit 4 test and add a timeout
Modified:
activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ2021Test.java
Modified: activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ2021Test.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ2021Test.java?rev=1506149&r1=1506148&r2=1506149&view=diff
==============================================================================
--- activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ2021Test.java (original)
+++ activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/bugs/AMQ2021Test.java Tue Jul 23 16:33:13 2013
@@ -16,6 +16,10 @@
*/
package org.apache.activemq.bugs;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
import java.lang.Thread.UncaughtExceptionHandler;
import java.util.ArrayList;
import java.util.Vector;
@@ -32,28 +36,33 @@ import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
-import junit.framework.TestCase;
-
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQTopic;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestName;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * This is a test case for the issue reported at:
- * 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
+ * This is a test case for the issue reported at: 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
*/
-public class AMQ2021Test extends TestCase implements ExceptionListener, UncaughtExceptionHandler {
+public class AMQ2021Test implements ExceptionListener, UncaughtExceptionHandler {
private static final Logger log = LoggerFactory.getLogger(AMQ2021Test.class);
BrokerService brokerService;
ArrayList<Thread> threads = new ArrayList<Thread>();
Vector<Throwable> exceptions;
+ @Rule
+ public TestName name = new TestName();
+
AMQ2021Test testCase;
private final String ACTIVEMQ_BROKER_BIND = "tcp://localhost:0";
@@ -62,14 +71,14 @@ public class AMQ2021Test extends TestCas
private final int numMessages = 1000;
private final int numConsumers = 2;
- private final int dlqMessages = numMessages/2;
+ private final int dlqMessages = numMessages / 2;
private CountDownLatch receivedLatch;
private ActiveMQTopic destination;
private CountDownLatch started;
- @Override
- protected void setUp() throws Exception {
+ @Before
+ public void setUp() throws Exception {
Thread.setDefaultUncaughtExceptionHandler(this);
testCase = this;
@@ -78,19 +87,18 @@ public class AMQ2021Test extends TestCas
brokerService.setDeleteAllMessagesOnStartup(true);
brokerService.addConnector(ACTIVEMQ_BROKER_BIND);
brokerService.start();
- destination = new ActiveMQTopic(getName());
+ destination = new ActiveMQTopic(name.getMethodName());
exceptions = new Vector<Throwable>();
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));
+ receivedLatch = new CountDownLatch(numConsumers * (numMessages + dlqMessages));
started = new CountDownLatch(1);
}
- @Override
- protected void tearDown() throws Exception {
+ @After
+ public void tearDown() throws Exception {
for (Thread t : threads) {
t.interrupt();
t.join();
@@ -98,9 +106,10 @@ public class AMQ2021Test extends TestCas
brokerService.stop();
}
+ @Test(timeout=240000)
public void testConcurrentTopicResendToDLQ() throws Exception {
- for (int i=0; i<numConsumers;i++) {
+ for (int i = 0; i < numConsumers; i++) {
ConsumerThread c1 = new ConsumerThread("Consumer-" + i);
threads.add(c1);
c1.start();
@@ -121,7 +130,7 @@ public class AMQ2021Test extends TestCas
producer.start();
boolean allGood = receivedLatch.await(90, TimeUnit.SECONDS);
- for (Throwable t: exceptions) {
+ for (Throwable t : exceptions) {
log.error("failing test with first exception", t);
fail("exception during test : " + t);
}
@@ -129,10 +138,10 @@ public class AMQ2021Test extends TestCas
assertEquals(0, exceptions.size());
- for (int i=0; i<numConsumers; i++) {
+ 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);
+ 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
@@ -141,16 +150,15 @@ public class AMQ2021Test extends TestCas
}
- private void consumeFromDLQ( int messageCount) throws Exception {
- ActiveMQConnectionFactory connectionFactory =
- new ActiveMQConnectionFactory(CONSUMER_BROKER_URL);
+ private void consumeFromDLQ(int messageCount) throws Exception {
+ 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;
- for (int i=0; i< messageCount; i++) {
+ for (int i = 0; i < messageCount; i++) {
if (dlqConsumer.receive(1000) == null) {
break;
}
@@ -160,7 +168,7 @@ public class AMQ2021Test extends TestCas
}
public void produce(int count) throws Exception {
- Connection connection=null;
+ Connection connection = null;
try {
ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(PRODUCER_BROKER_URL);
connection = factory.createConnection();
@@ -169,9 +177,9 @@ public class AMQ2021Test extends TestCas
producer.setTimeToLive(0);
connection.start();
- for (int i=0 ; i< count; i++) {
- int id = i+1;
- TextMessage message = session.createTextMessage(getName()+" Message "+ id);
+ for (int i = 0; i < count; i++) {
+ int id = i + 1;
+ TextMessage message = session.createTextMessage(name.getMethodName() + " Message " + id);
message.setIntProperty("MsgNumber", id);
producer.send(message);
@@ -192,7 +200,7 @@ public class AMQ2021Test extends TestCas
}
}
- public class ConsumerThread extends Thread implements MessageListener {
+ public class ConsumerThread extends Thread implements MessageListener {
public long counter = 0;
public long recoveries = 0;
private Session session;
@@ -201,10 +209,10 @@ public class AMQ2021Test extends TestCas
super(threadId);
}
+ @Override
public void run() {
try {
- ActiveMQConnectionFactory connectionFactory =
- new ActiveMQConnectionFactory(CONSUMER_BROKER_URL);
+ ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(CONSUMER_BROKER_URL);
Connection connection = connectionFactory.createConnection();
connection.setExceptionListener(testCase);
connection.setClientID(getName());
@@ -213,7 +221,7 @@ public class AMQ2021Test extends TestCas
consumer.setMessageListener(this);
connection.start();
- started .countDown();
+ started.countDown();
} catch (JMSException exception) {
log.error("unexpected ex in consumer run", exception);
@@ -221,11 +229,12 @@ public class AMQ2021Test extends TestCas
}
}
+ @Override
public void onMessage(Message message) {
try {
counter++;
- int messageNumber=message.getIntProperty("MsgNumber");
- if(messageNumber%2==0){
+ int messageNumber = message.getIntProperty("MsgNumber");
+ if (messageNumber % 2 == 0) {
session.recover();
recoveries++;
} else {
@@ -236,7 +245,7 @@ public class AMQ2021Test extends TestCas
log.info("recoveries:" + recoveries + ", Received " + counter + ", counter'th " + message);
}
receivedLatch.countDown();
- }catch (Exception e) {
+ } catch (Exception e) {
log.error("unexpected ex on onMessage", e);
exceptions.add(e);
}
@@ -244,14 +253,15 @@ public class AMQ2021Test extends TestCas
}
+ @Override
public void onException(JMSException exception) {
log.info("Unexpected JMSException", exception);
exceptions.add(exception);
}
+ @Override
public void uncaughtException(Thread thread, Throwable exception) {
log.info("Unexpected exception from thread " + thread + ", ex: " + exception);
exceptions.add(exception);
}
-
}