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

Don't use a fixed port and turn of JMX in the broker as its not needed in this test. 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1431033 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Timothy A. Bish 2013-01-09 19:55:43 +00:00
parent 0309eb3373
commit e724226033
1 changed files with 26 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
activemq-core/src/test/java/org/apache/activemq/bugs/AMQ2401Test.java
View File

@ -31,6 +31,8 @@ import javax.jms.MessageListener;
import javax.jms.MessageProducer; import javax.jms.MessageProducer;
import javax.jms.Session; import javax.jms.Session;
import junit.framework.TestCase;
import org.apache.activemq.ActiveMQConnection; import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory; import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService; import org.apache.activemq.broker.BrokerService;
@ -40,14 +42,11 @@ import org.apache.activemq.broker.region.policy.VMPendingQueueMessageStoragePoli
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import junit.framework.TestCase;
/** /**
* An AMQ-2401 Test * An AMQ-2401 Test
*
*/ */
public class AMQ2401Test extends TestCase implements MessageListener{ public class AMQ2401Test extends TestCase implements MessageListener {
BrokerService broker; private BrokerService broker;
private ActiveMQConnectionFactory factory; private ActiveMQConnectionFactory factory;
private static final int SEND_COUNT = 500; private static final int SEND_COUNT = 500;
private static final int CONSUMER_COUNT = 50; private static final int CONSUMER_COUNT = 50;
@ -56,15 +55,16 @@ public class AMQ2401Test extends TestCase implements MessageListener{
private static final Logger LOG = LoggerFactory.getLogger(AMQ2401Test.class); private static final Logger LOG = LoggerFactory.getLogger(AMQ2401Test.class);
private ArrayList<Service> services = new ArrayList<Service>(CONSUMER_COUNT + PRODUCER_COUNT); private final ArrayList<Service> services = new ArrayList<Service>(CONSUMER_COUNT + PRODUCER_COUNT);
int count = 0; private int count = 0;
CountDownLatch latch; private CountDownLatch latch;
@Override
protected void setUp() throws Exception { protected void setUp() throws Exception {
broker = new BrokerService(); broker = new BrokerService();
broker.setDataDirectory("target" + File.separator + "test-data" + File.separator + "AMQ2401Test"); broker.setDataDirectory("target" + File.separator + "test-data" + File.separator + "AMQ2401Test");
broker.setDeleteAllMessagesOnStartup(true); broker.setDeleteAllMessagesOnStartup(true);
broker.addConnector("tcp://0.0.0.0:2401"); String connectionUri = broker.addConnector("tcp://0.0.0.0:0").getPublishableConnectString();
PolicyMap policies = new PolicyMap(); PolicyMap policies = new PolicyMap();
PolicyEntry entry = new PolicyEntry(); PolicyEntry entry = new PolicyEntry();
entry.setMemoryLimit(1024 * 100); entry.setMemoryLimit(1024 * 100);
@ -73,15 +73,18 @@ public class AMQ2401Test extends TestCase implements MessageListener{
entry.setQueue(">"); entry.setQueue(">");
policies.setDefaultEntry(entry); policies.setDefaultEntry(entry);
broker.setDestinationPolicy(policies); broker.setDestinationPolicy(policies);
broker.setUseJmx(false);
broker.start(); broker.start();
broker.waitUntilStarted(); broker.waitUntilStarted();
factory = new ActiveMQConnectionFactory("tcp://0.0.0.0:2401"); factory = new ActiveMQConnectionFactory(connectionUri);
super.setUp(); super.setUp();
} }
@Override
protected void tearDown() throws Exception { protected void tearDown() throws Exception {
broker.stop(); broker.stop();
broker.waitUntilStopped();
} }
public void testDupsOk() throws Exception { public void testDupsOk() throws Exception {
@ -89,23 +92,20 @@ public class AMQ2401Test extends TestCase implements MessageListener{
TestProducer p = null; TestProducer p = null;
TestConsumer c = null; TestConsumer c = null;
try { try {
latch = new CountDownLatch(SEND_COUNT); latch = new CountDownLatch(SEND_COUNT);
for(int i = 0; i < CONSUMER_COUNT; i++) for (int i = 0; i < CONSUMER_COUNT; i++) {
{
TestConsumer consumer = new TestConsumer(); TestConsumer consumer = new TestConsumer();
consumer.start(); consumer.start();
services.add(consumer); services.add(consumer);
} }
for(int i = 0; i < PRODUCER_COUNT; i++) for (int i = 0; i < PRODUCER_COUNT; i++) {
{
TestProducer producer = new TestProducer(); TestProducer producer = new TestProducer();
producer.start(); producer.start();
services.add(producer); services.add(producer);
} }
waitForMessageReceipt(TimeUnit.SECONDS.toMillis(30));
waitForMessageReceipt(TimeUnit.SECONDS.toMillis(30));
} finally { } finally {
if (p != null) { if (p != null) {
p.close(); p.close();
@ -115,43 +115,34 @@ public class AMQ2401Test extends TestCase implements MessageListener{
c.close(); c.close();
} }
} }
} }
/* @Override
* (non-Javadoc)
*
* @see javax.jms.MessageListener#onMessage(javax.jms.Message)
*/
public void onMessage(Message message) { public void onMessage(Message message) {
latch.countDown(); latch.countDown();
if (++count % LOG_INTERVAL == 0) { if (++count % LOG_INTERVAL == 0) {
LOG.debug("Received message " + count); LOG.debug("Received message " + count);
} }
try { try {
Thread.sleep(1); Thread.sleep(1);
} catch (InterruptedException e) { } catch (InterruptedException e) {
Thread.currentThread().interrupt(); Thread.currentThread().interrupt();
} }
} }
/** /**
* @throws InterruptedException * @throws InterruptedException
* @throws TimeoutException * @throws TimeoutException
*
*/ */
private void waitForMessageReceipt(long timeout) throws InterruptedException, TimeoutException { private void waitForMessageReceipt(long timeout) throws InterruptedException, TimeoutException {
if (!latch.await(timeout, TimeUnit.MILLISECONDS)) { if (!latch.await(timeout, TimeUnit.MILLISECONDS)) {
throw new TimeoutException(String.format( throw new TimeoutException(String.format("Consumner didn't receive expected # of messages, %d of %d received.", latch.getCount(), SEND_COUNT));
"Consumner didn't receive expected # of messages, %d of %d received.",
latch.getCount(), SEND_COUNT ));
} }
} }
private interface Service { private interface Service {
public void start() throws Exception; public void start() throws Exception;
public void close(); public void close();
} }
@ -169,19 +160,20 @@ public class AMQ2401Test extends TestCase implements MessageListener{
connection.start(); connection.start();
session = connection.createSession(false, Session.DUPS_OK_ACKNOWLEDGE); session = connection.createSession(false, Session.DUPS_OK_ACKNOWLEDGE);
producer = session.createProducer(session.createQueue("AMQ2401Test")); producer = session.createProducer(session.createQueue("AMQ2401Test"));
} }
@Override
public void start() { public void start() {
thread.start(); thread.start();
} }
@Override
public void run() { public void run() {
int count = SEND_COUNT / PRODUCER_COUNT; int count = SEND_COUNT / PRODUCER_COUNT;
for (int i = 1; i <= count; i++) { for (int i = 1; i <= count; i++) {
try { try {
if( (i% LOG_INTERVAL)==0 ) { if ((i % LOG_INTERVAL) == 0) {
LOG.debug("Sending: " + i); LOG.debug("Sending: " + i);
} }
message = session.createBytesMessage(); message = session.createBytesMessage();
@ -194,6 +186,7 @@ public class AMQ2401Test extends TestCase implements MessageListener{
} }
} }
@Override
public void close() { public void close() {
try { try {
connection.close(); connection.close();
@ -217,10 +210,12 @@ public class AMQ2401Test extends TestCase implements MessageListener{
consumer.setMessageListener(AMQ2401Test.this); consumer.setMessageListener(AMQ2401Test.this);
} }
@Override
public void start() throws Exception { public void start() throws Exception {
connection.start(); connection.start();
} }
@Override
public void close() { public void close() {
try { try {
connection.close(); connection.close();
@ -228,11 +223,7 @@ public class AMQ2401Test extends TestCase implements MessageListener{
} }
} }
/* @Override
* (non-Javadoc)
*
* @see java.lang.Runnable#run()
*/
public void run() { public void run() {
while (latch.getCount() > 0) { while (latch.getCount() > 0) {
try { try {