Apache/activemq
1
0
Fork 0
mirror of https://github.com/apache/activemq.git synced 2025-02-28 21:29:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Added a leveldb version of the queue browsing test.

Browse Source 
git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1484319 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Hiram R. Chirino 2013-05-19 17:20:38 +00:00
parent e168b3cd6e
commit 665c3ab33b
2 changed files with 23 additions and 194 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

191
activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueBrowsingLevelDBTest.java
View File

@ -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;
import org.apache.activemq.leveldb.LevelDBStore;
public class QueueBrowsingLevelDBTest {
import java.io.File;
import java.io.IOException;
private static final Logger LOG = LoggerFactory.getLogger(QueueBrowsingLevelDBTest.class);
public class QueueBrowsingLevelDBTest extends QueueBrowsingTest {
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);
@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;
}
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));
}
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(messageToSend, received);
}
@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);
}
}

26
activemq-unit-tests/src/test/java/org/apache/activemq/usecases/QueueBrowsingTest.java
View File

@ -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();