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 2011/08/03 22:31:56 UTC
svn commit: r1153649 [2/2] - in
/activemq/trunk/activemq-core/src/test/java/org/apache/activemq: ./
advisory/ broker/jmx/ bugs/ usecases/
Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/QueueWorkerPrefetchTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/QueueWorkerPrefetchTest.java?rev=1153649&r1=1153648&r2=1153649&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/QueueWorkerPrefetchTest.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/QueueWorkerPrefetchTest.java Wed Aug 3 20:31:53 2011
@@ -41,21 +41,28 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * Test case demonstrating situation where messages are not delivered to consumers.
+ * Test case demonstrating situation where messages are not delivered to
+ * consumers.
*/
-public class QueueWorkerPrefetchTest extends TestCase implements MessageListener
-{
- private static final Logger LOG = LoggerFactory.getLogger(QueueWorkerPrefetchTest.class);
+public class QueueWorkerPrefetchTest extends TestCase implements
+ MessageListener {
+ private static final Logger LOG = LoggerFactory
+ .getLogger(QueueWorkerPrefetchTest.class);
private static final int BATCH_SIZE = 10;
- private static final long WAIT_TIMEOUT = 1000*10;
+ private static final long WAIT_TIMEOUT = 1000 * 10;
/** The connection URL. */
- private static final String CONNECTION_URL = "tcp://localhost:61616";
+ private static final String BROKER_BIND_ADDRESS = "tcp://localhost:0";
- /** The queue prefetch size to use. A value greater than 1 seems to make things work. */
+ /**
+ * The queue prefetch size to use. A value greater than 1 seems to make
+ * things work.
+ */
private static final int QUEUE_PREFETCH_SIZE = 1;
- /** The number of workers to use. A single worker with a prefetch of 1 works. */
+ /**
+ * The number of workers to use. A single worker with a prefetch of 1 works.
+ */
private static final int NUM_WORKERS = 2;
/** Embedded JMS broker. */
@@ -68,32 +75,37 @@ public class QueueWorkerPrefetchTest ext
private MessageConsumer masterItemConsumer;
/** The number of acks received by the master. */
- private AtomicLong acksReceived = new AtomicLong(0);
+ private final AtomicLong acksReceived = new AtomicLong(0);
- private AtomicReference<CountDownLatch> latch = new AtomicReference<CountDownLatch>();
+ private final AtomicReference<CountDownLatch> latch = new AtomicReference<CountDownLatch>();
+
+ private String connectionUri;
/** Messages sent to the work-item queue. */
- private static class WorkMessage implements Serializable
- {
+ private static class WorkMessage implements Serializable {
+ private static final long serialVersionUID = 1L;
private final int id;
+
public WorkMessage(int id) {
this.id = id;
}
+
@Override
public String toString() {
- return "Work: "+id;
+ return "Work: " + id;
}
}
/**
- * The worker process. Consume messages from the work-item queue, possibly creating
- * more messages to submit to the work-item queue. For each work item, send an ack
- * to the master.
+ * The worker process. Consume messages from the work-item queue, possibly
+ * creating more messages to submit to the work-item queue. For each work
+ * item, send an ack to the master.
*/
- private static class Worker implements MessageListener
- {
-
- /** Counter shared between workers to decided when new work-item messages are created. */
+ private static class Worker implements MessageListener {
+ /**
+ * Counter shared between workers to decided when new work-item messages
+ * are created.
+ */
private static AtomicInteger counter = new AtomicInteger(0);
/** Session to use. */
@@ -105,54 +117,48 @@ public class QueueWorkerPrefetchTest ext
/** Producer for sending new work items to the work-items queue. */
private MessageProducer workItemProducer;
- public Worker(Session session)
- throws JMSException
- {
+ public Worker(Session session) throws JMSException {
this.session = session;
- masterItemProducer = session.createProducer(session.createQueue("master-item"));
+ masterItemProducer = session.createProducer(session
+ .createQueue("master-item"));
Queue workItemQueue = session.createQueue("work-item");
workItemProducer = session.createProducer(workItemQueue);
- MessageConsumer workItemConsumer = session.createConsumer(workItemQueue);
+ MessageConsumer workItemConsumer = session
+ .createConsumer(workItemQueue);
workItemConsumer.setMessageListener(this);
}
- public void onMessage(javax.jms.Message message)
- {
- try
- {
- WorkMessage work = (WorkMessage)((ObjectMessage)message).getObject();
-
+ public void onMessage(javax.jms.Message message) {
+ try {
+ WorkMessage work = (WorkMessage) ((ObjectMessage) message)
+ .getObject();
+
long c = counter.incrementAndGet();
-
+
// Don't create a new work item for every BATCH_SIZE message. */
- if (c % BATCH_SIZE != 0)
- {
+ if (c % BATCH_SIZE != 0) {
// Send new work item to work-item queue.
- workItemProducer.send(session.createObjectMessage(
- new WorkMessage(work.id+1)));
+ workItemProducer.send(session
+ .createObjectMessage(new WorkMessage(work.id + 1)));
}
// Send ack to master.
masterItemProducer.send(session.createObjectMessage(work));
- }
- catch (JMSException e)
- {
+ } catch (JMSException e) {
throw new IllegalStateException("Something has gone wrong", e);
}
}
/** Close of JMS resources used by worker. */
- public void close() throws JMSException
- {
+ public void close() throws JMSException {
masterItemProducer.close();
workItemProducer.close();
session.close();
}
}
- /** Master message handler. Process ack messages. */
- public void onMessage(javax.jms.Message message)
- {
+ /** Master message handler. Process ack messages. */
+ public void onMessage(javax.jms.Message message) {
long acks = acksReceived.incrementAndGet();
latch.get().countDown();
if (acks % 1 == 0) {
@@ -160,68 +166,72 @@ public class QueueWorkerPrefetchTest ext
}
}
- protected void setUp() throws Exception
- {
+ protected void setUp() throws Exception {
// Create the message broker.
super.setUp();
broker = new BrokerService();
broker.setPersistent(false);
broker.setUseJmx(true);
- broker.addConnector(CONNECTION_URL);
+ broker.addConnector(BROKER_BIND_ADDRESS);
broker.start();
+ broker.waitUntilStarted();
+
+ connectionUri = broker.getTransportConnectors().get(0).getPublishableConnectString();
}
- protected void tearDown() throws Exception
- {
+ protected void tearDown() throws Exception {
// Shut down the message broker.
broker.deleteAllMessages();
broker.stop();
super.tearDown();
}
- public void testActiveMQ()
- throws Exception
- {
+ public void testActiveMQ() throws Exception {
// Create the connection to the broker.
- ActiveMQConnectionFactory connectionFactory =
- new ActiveMQConnectionFactory(CONNECTION_URL);
+ ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(connectionUri);
ActiveMQPrefetchPolicy prefetchPolicy = new ActiveMQPrefetchPolicy();
prefetchPolicy.setQueuePrefetch(QUEUE_PREFETCH_SIZE);
connectionFactory.setPrefetchPolicy(prefetchPolicy);
Connection connection = connectionFactory.createConnection();
connection.start();
- Session masterSession = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- workItemProducer = masterSession.createProducer(masterSession.createQueue("work-item"));
- masterItemConsumer = masterSession.createConsumer(masterSession.createQueue("master-item"));
+ Session masterSession = connection.createSession(false,
+ Session.AUTO_ACKNOWLEDGE);
+ workItemProducer = masterSession.createProducer(masterSession
+ .createQueue("work-item"));
+ masterItemConsumer = masterSession.createConsumer(masterSession
+ .createQueue("master-item"));
masterItemConsumer.setMessageListener(this);
// Create the workers.
Worker[] workers = new Worker[NUM_WORKERS];
- for (int i = 0; i < NUM_WORKERS; i++)
- {
- workers[i] = new Worker(connection.createSession(false, Session.AUTO_ACKNOWLEDGE));
+ for (int i = 0; i < NUM_WORKERS; i++) {
+ workers[i] = new Worker(connection.createSession(false,
+ Session.AUTO_ACKNOWLEDGE));
}
- // Send a message to the work queue, and wait for the BATCH_SIZE acks from the workers.
+ // Send a message to the work queue, and wait for the BATCH_SIZE acks
+ // from the workers.
acksReceived.set(0);
latch.set(new CountDownLatch(BATCH_SIZE));
- workItemProducer.send(masterSession.createObjectMessage(new WorkMessage(1)));
-
+ workItemProducer.send(masterSession
+ .createObjectMessage(new WorkMessage(1)));
+
if (!latch.get().await(WAIT_TIMEOUT, TimeUnit.MILLISECONDS)) {
fail("First batch only received " + acksReceived + " messages");
}
- LOG.info("First batch received");
+ LOG.info("First batch received");
- // Send another message to the work queue, and wait for the next 1000 acks. It is
+ // Send another message to the work queue, and wait for the next 1000 acks. It is
// at this point where the workers never get notified of this message, as they
- // have a large pending queue. Creating a new worker at this point however will
+ // have a large pending queue. Creating a new worker at this point however will
// receive this new message.
acksReceived.set(0);
latch.set(new CountDownLatch(BATCH_SIZE));
- workItemProducer.send(masterSession.createObjectMessage(new WorkMessage(1)));
-
+ workItemProducer.send(masterSession
+ .createObjectMessage(new WorkMessage(1)));
+
if (!latch.get().await(WAIT_TIMEOUT, TimeUnit.MILLISECONDS)) {
fail("Second batch only received " + acksReceived + " messages");
}
Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/SlowConsumerTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/SlowConsumerTest.java?rev=1153649&r1=1153648&r2=1153649&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/SlowConsumerTest.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/SlowConsumerTest.java Wed Aug 3 20:31:53 2011
@@ -41,8 +41,8 @@ public class SlowConsumerTest extends Te
private static final Logger LOG = LoggerFactory.getLogger(SlowConsumerTest.class);
private static final int MESSAGES_COUNT = 10000;
- protected int messageLogFrequency = 2500;
- protected long messageReceiveTimeout = 10000L;
+ private final int messageLogFrequency = 2500;
+ private final long messageReceiveTimeout = 10000L;
private Socket stompSocket;
private ByteArrayOutputStream inputBuffer;
@@ -58,9 +58,10 @@ public class SlowConsumerTest extends Te
broker.setUseJmx(true);
broker.setDeleteAllMessagesOnStartup(true);
- broker.addConnector("tcp://localhost:61616").setName("Default");
+ broker.addConnector("tcp://localhost:0").setName("Default");
broker.start();
- ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
+ ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(
+ broker.getTransportConnectors().get(0).getPublishableConnectString());
final Connection connection = factory.createConnection();
connection.start();
Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/TransactionNotStartedErrorTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/TransactionNotStartedErrorTest.java?rev=1153649&r1=1153648&r2=1153649&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/TransactionNotStartedErrorTest.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/bugs/TransactionNotStartedErrorTest.java Wed Aug 3 20:31:53 2011
@@ -43,8 +43,8 @@ import org.slf4j.LoggerFactory;
public class TransactionNotStartedErrorTest extends TestCase {
private static final Logger LOG = LoggerFactory.getLogger(TransactionNotStartedErrorTest.class);
-
- private static int counter = 500;
+
+ private static final int counter = 500;
private static int hectorToHaloCtr;
private static int xenaToHaloCtr;
@@ -54,14 +54,13 @@ public class TransactionNotStartedErrorT
private static int haloToXenaCtr;
private static int haloToTroyCtr;
- private String hectorToHalo = "hectorToHalo";
- private String xenaToHalo = "xenaToHalo";
- private String troyToHalo = "troyToHalo";
-
- private String haloToHector = "haloToHector";
- private String haloToXena = "haloToXena";
- private String haloToTroy = "haloToTroy";
-
+ private final String hectorToHalo = "hectorToHalo";
+ private final String xenaToHalo = "xenaToHalo";
+ private final String troyToHalo = "troyToHalo";
+
+ private final String haloToHector = "haloToHector";
+ private final String haloToXena = "haloToXena";
+ private final String haloToTroy = "haloToTroy";
private BrokerService broker;
@@ -72,8 +71,9 @@ public class TransactionNotStartedErrorT
private final Object lock = new Object();
- public Connection createConnection() throws JMSException {
- ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
+ public Connection createConnection() throws Exception {
+ ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(
+ broker.getTransportConnectors().get(0).getPublishableConnectString());
return factory.createConnection();
}
@@ -86,7 +86,7 @@ public class TransactionNotStartedErrorT
broker.setDeleteAllMessagesOnStartup(true);
broker.setPersistent(true);
broker.setUseJmx(true);
- broker.addConnector("tcp://localhost:61616").setName("Default");
+ broker.addConnector("tcp://localhost:0").setName("Default");
broker.start();
LOG.info("Starting broker..");
}
@@ -234,12 +234,10 @@ public class TransactionNotStartedErrorT
}
public MessageSender buildTransactionalProducer(String queueName, Connection connection) throws Exception {
-
return new MessageSender(queueName, connection, true, false);
}
public Thread buildProducer(Connection connection, final String queueName) throws Exception {
-
final Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
final MessageSender producer = new MessageSender(queueName, connection, false, false);
Thread thread = new Thread() {
Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/ConcurrentProducerDurableConsumerTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/ConcurrentProducerDurableConsumerTest.java?rev=1153649&r1=1153648&r2=1153649&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/ConcurrentProducerDurableConsumerTest.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/ConcurrentProducerDurableConsumerTest.java Wed Aug 3 20:31:53 2011
@@ -326,7 +326,7 @@ public class ConcurrentProducerDurableCo
protected BrokerService createBroker() throws Exception {
BrokerService brokerService = new BrokerService();
brokerService.setEnableStatistics(false);
- brokerService.addConnector("tcp://0.0.0.0:61616");
+ brokerService.addConnector("tcp://0.0.0.0:0");
brokerService.setDeleteAllMessagesOnStartup(true);
PolicyEntry policy = new PolicyEntry();
@@ -343,9 +343,9 @@ public class ConcurrentProducerDurableCo
policyMap.setDefaultEntry(policy);
brokerService.setDestinationPolicy(policyMap);
- if (false) {
- // external mysql works a lot faster
- //
+// if (false) {
+// // external mysql works a lot faster
+// //
// JDBCPersistenceAdapter jdbc = new JDBCPersistenceAdapter();
// BasicDataSource ds = new BasicDataSource();
// com.mysql.jdbc.Driver d = new com.mysql.jdbc.Driver();
@@ -358,28 +358,29 @@ public class ConcurrentProducerDurableCo
// jdbc.setDataSource(ds);
// brokerService.setPersistenceAdapter(jdbc);
-/* add mysql bits to the pom in the testing dependencies
-<dependency>
- <groupId>mysql</groupId>
- <artifactId>mysql-connector-java</artifactId>
- <version>5.1.10</version>
- <scope>test</scope>
-</dependency>
-<dependency>
- <groupId>commons-dbcp</groupId>
- <artifactId>commons-dbcp</artifactId>
- <version>1.2.2</version>
- <scope>test</scope>
-</dependency>
+ /* add mysql bits to the pom in the testing dependencies
+ <dependency>
+ <groupId>mysql</groupId>
+ <artifactId>mysql-connector-java</artifactId>
+ <version>5.1.10</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>commons-dbcp</groupId>
+ <artifactId>commons-dbcp</artifactId>
+ <version>1.2.2</version>
+ <scope>test</scope>
+ </dependency>
*/
- } else {
+// } else {
setDefaultPersistenceAdapter(brokerService);
- }
+// }
return brokerService;
}
protected ActiveMQConnectionFactory createConnectionFactory() throws Exception {
- ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("tcp://localhost:61616");
+ ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(
+ broker.getTransportConnectors().get(0).getPublishableConnectString());
ActiveMQPrefetchPolicy prefetchPolicy = new ActiveMQPrefetchPolicy();
prefetchPolicy.setAll(1);
factory.setPrefetchPolicy(prefetchPolicy);
Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/DispatchMultipleConsumersTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/DispatchMultipleConsumersTest.java?rev=1153649&r1=1153648&r2=1153649&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/DispatchMultipleConsumersTest.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/DispatchMultipleConsumersTest.java Wed Aug 3 20:31:53 2011
@@ -1,4 +1,3 @@
-
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
@@ -16,6 +15,7 @@
* limitations under the License.
*/
package org.apache.activemq.usecases;
+
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.Connection;
@@ -32,10 +32,8 @@ import org.apache.activemq.command.Activ
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
/**
* @author Rajani Chennamaneni
- *
*/
public class DispatchMultipleConsumersTest extends TestCase {
private final static Logger logger = LoggerFactory.getLogger(DispatchMultipleConsumersTest.class);
@@ -50,10 +48,10 @@ public class DispatchMultipleConsumersTe
AtomicInteger consumedCount;
CountDownLatch producerLatch;
CountDownLatch consumerLatch;
- String brokerURL = "tcp://localhost:61616";
+ String brokerURL;
String userName = "";
String password = "";
-
+
@Override
protected void setUp() throws Exception {
super.setUp();
@@ -61,42 +59,36 @@ public class DispatchMultipleConsumersTe
broker.setPersistent(true);
broker.setUseJmx(true);
broker.deleteAllMessages();
- broker.addConnector("tcp://localhost:61616");
+ broker.addConnector("tcp://localhost:0");
broker.start();
+ broker.waitUntilStarted();
dest = new ActiveMQQueue(destinationName);
resetCounters();
+ brokerURL = broker.getTransportConnectors().get(0).getPublishableConnectString();
}
@Override
protected void tearDown() throws Exception {
-// broker.stop();
+ broker.stop();
+ broker.waitUntilStopped();
super.tearDown();
}
-
+
private void resetCounters() {
sentCount = new AtomicInteger(0);
consumedCount = new AtomicInteger(0);
producerLatch = new CountDownLatch(producerThreads);
consumerLatch = new CountDownLatch(consumerCount);
}
-
+
public void testDispatch1() {
for (int i = 1; i <= 5; i++) {
resetCounters();
dispatch();
- /*try {
- System.out.print("Press Enter to continue/finish:");
- //pause to check the counts on JConsole
- System.in.read();
- System.in.read();
- } catch (IOException e) {
- e.printStackTrace();
- }*/
- //check for consumed messages count
assertEquals("Incorrect messages in Iteration " + i, sentCount.get(), consumedCount.get());
}
}
-
+
private void dispatch() {
startConsumers();
startProducers();
@@ -130,15 +122,13 @@ public class DispatchMultipleConsumersTe
}
private class ConsumerThread extends Thread {
- Connection conn;
Session session;
MessageConsumer consumer;
public ConsumerThread(Connection conn, String name) {
super();
- this.conn = conn;
this.setName(name);
- logger.info("Created new consumer thread:" + name);
+ logger.trace("Created new consumer thread:" + name);
try {
session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
consumer = session.createConsumer(dest);
@@ -170,24 +160,25 @@ public class DispatchMultipleConsumersTe
nullCount = 0;
}
Thread.sleep(100);
- logger.info("Message received:" + msg.getJMSMessageID());
+ if (logger.isTraceEnabled()) {
+ logger.trace("Message received:" + msg.getJMSMessageID());
+ }
msgCount++;
} catch (JMSException e) {
- logger.error("Failed to consume:", e);
+ logger.error("Failed to consume:", e);
} catch (InterruptedException e) {
- logger.error("Interrupted!", e);
+ logger.error("Interrupted!", e);
}
}
try {
consumer.close();
} catch (JMSException e) {
- logger.error("Failed to close consumer " + getName(), e);
+ logger.error("Failed to close consumer " + getName(), e);
}
consumedCount.addAndGet(msgCount);
consumerLatch.countDown();
- logger.info("Consumed " + msgCount + " messages using thread " + getName());
+ logger.trace("Consumed " + msgCount + " messages using thread " + getName());
}
-
}
private class ProducerThread extends Thread {
@@ -195,12 +186,12 @@ public class DispatchMultipleConsumersTe
Connection conn;
Session session;
MessageProducer producer;
-
+
public ProducerThread(ActiveMQConnectionFactory connFactory, int count, String name) {
super();
this.count = count;
this.setName(name);
- logger.info("Created new producer thread:" + name);
+ logger.trace("Created new producer thread:" + name);
try {
conn = connFactory.createConnection();
conn.start();
@@ -224,12 +215,13 @@ public class DispatchMultipleConsumersTe
} catch (JMSException e) {
logger.error(e.getMessage(), e);
} catch (InterruptedException e) {
- logger.error("Interrupted!", e);
+ logger.error("Interrupted!", e);
}
sentCount.addAndGet(i);
producerLatch.countDown();
- logger.info("Sent " + i + " messages from thread " + getName());
+ if (logger.isTraceEnabled()) {
+ logger.trace("Sent " + i + " messages from thread " + getName());
+ }
}
}
-
}
Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/MessageGroupCloseTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/MessageGroupCloseTest.java?rev=1153649&r1=1153648&r2=1153649&view=diff
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/MessageGroupCloseTest.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/usecases/MessageGroupCloseTest.java Wed Aug 3 20:31:53 2011
@@ -48,7 +48,6 @@ public class MessageGroupCloseTest exten
private HashSet<String> closedGroups2 = new HashSet<String>();
// with the prefetch too high, this bug is not realized
private static final String connStr =
- //"tcp://localhost:61616";
"vm://localhost?broker.persistent=false&broker.useJmx=false&jms.prefetchPolicy.all=1";
public void testNewConsumer() throws JMSException, InterruptedException {