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

merging 903758 - 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/branches/activemq-5.3@906104 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Bosanac Dejan 2010-02-03 16:13:46 +00:00
parent 37e1cdd4b9
commit 159a52bd51
1 changed files with 18 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

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