Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

resolve intermitent failure on loaded machines, consumer could complete before producer thread got to do any work at all 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@903758 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Tully 2010-01-27 17:46:08 +00:00
parent 19bfd9d3b5
commit 0d9c588ed7
1 changed files with 19 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
activemq-core/src/test/java/org/apache/activemq/bugs/TempStorageBlockedBrokerTest.java
View File

@ -17,7 +17,11 @@
package org.apache.activemq.bugs; package org.apache.activemq.bugs;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.File; import java.io.File;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.Connection; import javax.jms.Connection;
import javax.jms.DeliveryMode; import javax.jms.DeliveryMode;
@ -80,6 +84,7 @@ public class TempStorageBlockedBrokerTest {
final Connection producerConnection = factory.createConnection(); final Connection producerConnection = factory.createConnection();
producerConnection.start(); producerConnection.start();
final CountDownLatch producerHasSentTenMessages = new CountDownLatch(10);
Thread producingThread = new Thread("Producing thread") { Thread producingThread = new Thread("Producing thread") {
@Override @Override
public void run() { public void run() {
@ -92,10 +97,12 @@ public class TempStorageBlockedBrokerTest {
producer.send(message); producer.send(message);
messagesSent.incrementAndGet(); messagesSent.incrementAndGet();
producerHasSentTenMessages.countDown();
Thread.sleep(10); Thread.sleep(10);
LOG.info("Sent Message " + idx); if (idx != 0 && idx%100 == 0) {
LOG.info("Temp Store Usage " + broker.getSystemUsage().getTempUsage().getUsage()); LOG.info("Sent Message " + idx);
LOG.info("Temp Store Usage " + broker.getSystemUsage().getTempUsage().getUsage());
}
} }
producer.close(); producer.close();
session.close(); session.close();
@ -106,12 +113,16 @@ public class TempStorageBlockedBrokerTest {
}; };
producingThread.start(); producingThread.start();
assertTrue("producer has sent 10 in a reasonable time", producerHasSentTenMessages.await(30, TimeUnit.SECONDS));
int count = 0; int count = 0;
Message m = null; Message m = null;
while ((m = consumer.receive(messageReceiveTimeout)) != null) { while ((m = consumer.receive(messageReceiveTimeout)) != null) {
count++; count++;
LOG.info("Recieved Message (" + count + "):" + m); if (count != 0 && count%10 == 0) {
LOG.info("Recieved Message (" + count + "):" + m);
}
messagesConsumed.incrementAndGet(); messagesConsumed.incrementAndGet();
try { try {
Thread.sleep(100); Thread.sleep(100);
@ -120,19 +131,13 @@ public class TempStorageBlockedBrokerTest {
} }
} }
LOG.info("Connection Timeout: Retrying"); LOG.info("Connection Timeout: Retrying.. count: " + count);
// session.close();
// consumerConnection.close();
//
// consumerConnection2.start();
// session2 = consumerConnection2.createSession(false,
// Session.AUTO_ACKNOWLEDGE);
// consumer = session2.createConsumer(destination);
while ((m = consumer.receive(messageReceiveTimeout)) != null) { while ((m = consumer.receive(messageReceiveTimeout)) != null) {
count++; count++;
LOG.info("Recieved Message (" + count + "):" + m); if (count != 0 && count%10 == 0) {
LOG.info("Recieved Message (" + count + "):" + m);
}
messagesConsumed.incrementAndGet(); messagesConsumed.incrementAndGet();
try { try {
Thread.sleep(100); Thread.sleep(100);