Author: chirino
Date: Sun May 19 17:20:38 2013
New Revision: 1484319
URL: http://svn.apache.org/r1484319
Log:
Added a leveldb version of the queue browsing test.
Modified:
activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueBrowsingLevelDBTest.java
activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueBrowsingTest.java
Modified:
activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueBrowsingLevelDBTest.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueBrowsingLevelDBTest.java?rev=1484319&r1=1484318&r2=1484319&view=diff
==============================================================================
---
activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueBrowsingLevelDBTest.java
(original)
+++
activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueBrowsingLevelDBTest.java
Sun May 19 17:20:38 2013
@@ -16,190 +16,21 @@
*/
package org.apache.activemq.usecases;
-import static org.junit.Assert.assertEquals;
-
-import java.net.URI;
-import java.util.Enumeration;
-
-import javax.jms.Connection;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.QueueBrowser;
-import javax.jms.Session;
-
-import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.broker.TransportConnector;
-import org.apache.activemq.command.ActiveMQQueue;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class QueueBrowsingLevelDBTest {
-
- private static final Logger LOG =
LoggerFactory.getLogger(QueueBrowsingLevelDBTest.class);
-
- private BrokerService broker;
- private URI connectUri;
- private ActiveMQConnectionFactory factory;
-
-
- @Before
- public void startBroker() throws Exception {
- createBroker();
- TransportConnector connector = broker.addConnector("tcp://0.0.0.0:0");
- broker.deleteAllMessages();
- broker.start();
- broker.waitUntilStarted();
- connectUri = connector.getConnectUri();
- factory = new ActiveMQConnectionFactory(connectUri);
- }
-
- private void createBroker() {
- broker = new BrokerService();
- }
-
- @After
- public void stopBroker() throws Exception {
- broker.stop();
- broker.waitUntilStopped();
- }
-
- @Test
- public void testBrowsing() throws JMSException {
-
- int messageToSend = 370;
-
- ActiveMQQueue queue = new ActiveMQQueue("TEST");
- Connection connection = factory.createConnection();
- connection.start();
- Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
- MessageProducer producer = session.createProducer(queue);
-
- String data = "";
- for( int i=0; i < 1024*2; i++ ) {
- data += "x";
- }
-
- for( int i=0; i < messageToSend; i++ ) {
- producer.send(session.createTextMessage(data));
- }
+import org.apache.activemq.leveldb.LevelDBStore;
- QueueBrowser browser = session.createBrowser(queue);
- Enumeration<?> enumeration = browser.getEnumeration();
- int received = 0;
- while (enumeration.hasMoreElements()) {
- Message m = (Message) enumeration.nextElement();
- received++;
- LOG.info("Browsed message " + received + ": " +
m.getJMSMessageID());
- }
+import java.io.File;
+import java.io.IOException;
- browser.close();
+public class QueueBrowsingLevelDBTest extends QueueBrowsingTest {
- assertEquals(messageToSend, received);
+ @Override
+ public BrokerService createBroker() throws IOException {
+ BrokerService broker = super.createBroker();
+ LevelDBStore store = new LevelDBStore();
+ store.setDirectory(new File("target/test-data/leveldb"));
+ broker.setPersistenceAdapter(store);
+ return broker;
}
- @Test
- public void testBrowseConcurrent() throws Exception {
- final int messageToSend = 370;
-
- final ActiveMQQueue queue = new ActiveMQQueue("TEST");
- Connection connection = factory.createConnection();
- connection.start();
- final Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
-
- MessageProducer producer = session.createProducer(queue);
-
- String data = "";
- for( int i=0; i < 1024*2; i++ ) {
- data += "x";
- }
-
- for( int i=0; i < messageToSend; i++ ) {
- producer.send(session.createTextMessage(data));
- }
-
- Thread browserThread = new Thread() {
- @Override
- public void run() {
- try {
- QueueBrowser browser = session.createBrowser(queue);
- Enumeration<?> enumeration = browser.getEnumeration();
- int received = 0;
- while (enumeration.hasMoreElements()) {
- Message m = (Message) enumeration.nextElement();
- received++;
- LOG.info("Browsed message " + received + ": " +
m.getJMSMessageID());
- }
- assertEquals("Browsed all messages", messageToSend,
received);
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
- };
-
- browserThread.start();
-
- Thread consumerThread = new Thread() {
- @Override
- public void run() {
- try {
- MessageConsumer consumer = session.createConsumer(queue);
- int received = 0;
- while (true) {
- Message m = consumer.receive(1000);
- if (m == null)
- break;
- received++;
- }
- assertEquals("Consumed all messages", messageToSend,
received);
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
- };
-
- consumerThread.start();
-
- browserThread.join();
- consumerThread.join();
- }
-
- @Test
- public void testMemoryLimit() throws Exception {
- broker.getSystemUsage().getMemoryUsage().setLimit(10 * 1024);
-
- int messageToSend = 370;
-
- ActiveMQQueue queue = new ActiveMQQueue("TEST");
- Connection connection = factory.createConnection();
- connection.start();
- Session session = connection.createSession(false,
Session.AUTO_ACKNOWLEDGE);
- MessageProducer producer = session.createProducer(queue);
-
- String data = "";
- for( int i=0; i < 1024*2; i++ ) {
- data += "x";
- }
-
- for( int i=0; i < messageToSend; i++ ) {
- producer.send(session.createTextMessage(data));
- }
-
- QueueBrowser browser = session.createBrowser(queue);
- Enumeration<?> enumeration = browser.getEnumeration();
- int received = 0;
- while (enumeration.hasMoreElements()) {
- Message m = (Message) enumeration.nextElement();
- received++;
- LOG.info("Browsed message " + received + ": " +
m.getJMSMessageID());
- }
-
- browser.close();
- assertEquals(3, received);
- }
}
Modified:
activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueBrowsingTest.java
URL:
http://svn.apache.org/viewvc/activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueBrowsingTest.java?rev=1484319&r1=1484318&r2=1484319&view=diff
==============================================================================
---
activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueBrowsingTest.java
(original)
+++
activemq/trunk/activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueBrowsingTest.java
Sun May 19 17:20:38 2013
@@ -16,19 +16,6 @@
*/
package org.apache.activemq.usecases;
-import static org.junit.Assert.assertEquals;
-
-import java.net.URI;
-import java.util.Enumeration;
-
-import javax.jms.Connection;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.QueueBrowser;
-import javax.jms.Session;
-
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.TransportConnector;
@@ -39,6 +26,13 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.jms.*;
+import java.io.IOException;
+import java.net.URI;
+import java.util.Enumeration;
+
+import static org.junit.Assert.assertEquals;
+
public class QueueBrowsingTest {
private static final Logger LOG =
LoggerFactory.getLogger(QueueBrowsingTest.class);
@@ -50,7 +44,7 @@ public class QueueBrowsingTest {
@Before
public void startBroker() throws Exception {
- broker = new BrokerService();
+ broker = createBroker();
TransportConnector connector = broker.addConnector("tcp://0.0.0.0:0");
broker.deleteAllMessages();
broker.start();
@@ -59,6 +53,10 @@ public class QueueBrowsingTest {
factory = new ActiveMQConnectionFactory(connectUri);
}
+ public BrokerService createBroker() throws IOException {
+ return new BrokerService();
+ }
+
@After
public void stopBroker() throws Exception {
broker.stop();
