You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2012/05/11 09:42:57 UTC
svn commit: r1337045 [5/5] - in
/qpid/branches/java-config-and-management/qpid/java: ./
amqp-1-0-client-jms/ amqp-1-0-client/ amqp-1-0-common/
bdbstore/src/main/java/org/apache/qpid/server/store/berkeleydb/
broker-plugins/access-control/src/main/java/o...
Modified: qpid/branches/java-config-and-management/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/endtoend/EndToEndTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/endtoend/EndToEndTest.java?rev=1337045&r1=1337044&r2=1337045&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/endtoend/EndToEndTest.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/perftests/src/test/java/org/apache/qpid/systest/disttest/endtoend/EndToEndTest.java Fri May 11 07:42:48 2012
@@ -19,7 +19,7 @@
package org.apache.qpid.systest.disttest.endtoend;
import static org.apache.qpid.disttest.AbstractRunner.JNDI_CONFIG_PROP;
-import static org.apache.qpid.disttest.ControllerRunner.OUTPUT_FILE_PROP;
+import static org.apache.qpid.disttest.ControllerRunner.OUTPUT_DIR_PROP;
import static org.apache.qpid.disttest.ControllerRunner.TEST_CONFIG_PROP;
import java.io.File;
@@ -39,18 +39,19 @@ public class EndToEndTest extends QpidBr
public void testRunner() throws Exception
{
- File csvOutputFile = createTemporaryCsvFile();
- assertFalse("CSV output file must exist",csvOutputFile.exists());
+ File csvOutputDir = createTemporaryCsvDirectory();
+ assertTrue("CSV output dir must not exist",csvOutputDir.isDirectory());
final String[] args = new String[] {TEST_CONFIG_PROP + "=" + TEST_CONFIG,
JNDI_CONFIG_PROP + "=" + JNDI_CONFIG_FILE,
- OUTPUT_FILE_PROP + "=" + csvOutputFile.getAbsolutePath()};
+ OUTPUT_DIR_PROP + "=" + csvOutputDir.getAbsolutePath()};
_runner = new ControllerRunner();
_runner.parseArgumentsIntoConfig(args);
_runner.runController();
- assertTrue("CSV output file must exist", csvOutputFile.exists());
- final String csvContents = FileUtils.readFileAsString(csvOutputFile);
+ File expectedCsvOutputFile = new File(csvOutputDir, "endtoend.csv");
+ assertTrue("CSV output file must exist", expectedCsvOutputFile.exists());
+ final String csvContents = FileUtils.readFileAsString(expectedCsvOutputFile);
final String[] csvLines = csvContents.split("\n");
int numberOfHeaders = 1;
@@ -82,13 +83,13 @@ public class EndToEndTest extends QpidBr
}
- private File createTemporaryCsvFile() throws IOException
+ private File createTemporaryCsvDirectory() throws IOException
{
- File csvFile = File.createTempFile(getName(), ".csv");
- csvFile.delete();
- csvFile.deleteOnExit();
-
- return csvFile;
+ String tmpDir = System.getProperty("java.io.tmpdir");
+ File csvDir = new File(tmpDir, "csv");
+ csvDir.mkdir();
+ csvDir.deleteOnExit();
+ return csvDir;
}
}
Propchange: qpid/branches/java-config-and-management/qpid/java/systests/src/main/java/org/apache/qpid/server/SupportedProtocolVersionsTest.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/server/SupportedProtocolVersionsTest.java:r1327945-1336832
Modified: qpid/branches/java-config-and-management/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/ConflationQueueTest.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/ConflationQueueTest.java?rev=1337045&r1=1337044&r2=1337045&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/ConflationQueueTest.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/systests/src/main/java/org/apache/qpid/server/queue/ConflationQueueTest.java Fri May 11 07:42:48 2012
@@ -21,8 +21,7 @@
package org.apache.qpid.server.queue;
-import org.apache.log4j.Logger;
-
+import org.apache.qpid.AMQException;
import org.apache.qpid.client.AMQDestination;
import org.apache.qpid.client.AMQQueue;
import org.apache.qpid.client.AMQSession;
@@ -44,45 +43,27 @@ import java.util.Map;
public class ConflationQueueTest extends QpidBrokerTestCase
{
- private static final int TIMEOUT = 1500;
-
-
- private static final Logger _logger = Logger.getLogger(ConflationQueueTest.class);
-
-
-
- protected final String VHOST = "/test";
- protected final String QUEUE = "ConflationQueue";
private static final int MSG_COUNT = 400;
+ private String queueName;
private Connection producerConnection;
private MessageProducer producer;
private Session producerSession;
private Queue queue;
private Connection consumerConnection;
private Session consumerSession;
-
-
private MessageConsumer consumer;
protected void setUp() throws Exception
{
super.setUp();
+ queueName = getTestQueueName();
producerConnection = getConnection();
producerSession = producerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
producerConnection.start();
-
-
- }
-
- protected void tearDown() throws Exception
- {
- producerConnection.close();
- consumerConnection.close();
- super.tearDown();
}
public void testConflation() throws Exception
@@ -90,12 +71,7 @@ public class ConflationQueueTest extends
consumerConnection = getConnection();
consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-
- final Map<String,Object> arguments = new HashMap<String, Object>();
- arguments.put("qpid.last_value_queue_key","key");
- ((AMQSession) producerSession).createQueue(new AMQShortString(QUEUE), false, true, false, arguments);
- queue = new org.apache.qpid.client.AMQQueue("amq.direct",QUEUE);
- ((AMQSession) producerSession).declareAndBind((AMQDestination)queue);
+ createConflationQueue(producerSession);
producer = producerSession.createProducer(queue);
for (int msg = 0; msg < MSG_COUNT; msg++)
@@ -124,22 +100,15 @@ public class ConflationQueueTest extends
Message msg = messages.get(i);
assertEquals("Unexpected message number received", MSG_COUNT - 10 + i, msg.getIntProperty("msg"));
}
-
-
}
-
public void testConflationWithRelease() throws Exception
{
consumerConnection = getConnection();
consumerSession = consumerConnection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
- final Map<String,Object> arguments = new HashMap<String, Object>();
- arguments.put("qpid.last_value_queue_key","key");
- ((AMQSession) producerSession).createQueue(new AMQShortString(QUEUE), false, true, false, arguments);
- queue = new org.apache.qpid.client.AMQQueue("amq.direct",QUEUE);
- ((AMQSession) producerSession).declareAndBind((AMQDestination)queue);
+ createConflationQueue(producerSession);
producer = producerSession.createProducer(queue);
for (int msg = 0; msg < MSG_COUNT/2; msg++)
@@ -149,7 +118,7 @@ public class ConflationQueueTest extends
}
// HACK to do something synchronous
- ((AMQSession)producerSession).sync();
+ ((AMQSession<?,?>)producerSession).sync();
consumer = consumerSession.createConsumer(queue);
consumerConnection.start();
@@ -183,7 +152,7 @@ public class ConflationQueueTest extends
// HACK to do something synchronous
- ((AMQSession)producerSession).sync();
+ ((AMQSession<?,?>)producerSession).sync();
consumer = consumerSession.createConsumer(queue);
consumerConnection.start();
@@ -205,18 +174,13 @@ public class ConflationQueueTest extends
}
-
public void testConflationWithReleaseAfterNewPublish() throws Exception
{
consumerConnection = getConnection();
consumerSession = consumerConnection.createSession(false, Session.CLIENT_ACKNOWLEDGE);
- final Map<String,Object> arguments = new HashMap<String, Object>();
- arguments.put("qpid.last_value_queue_key","key");
- ((AMQSession) producerSession).createQueue(new AMQShortString(QUEUE), false, true, false, arguments);
- queue = new org.apache.qpid.client.AMQQueue("amq.direct",QUEUE);
- ((AMQSession) producerSession).declareAndBind((AMQDestination)queue);
+ createConflationQueue(producerSession);
producer = producerSession.createProducer(queue);
for (int msg = 0; msg < MSG_COUNT/2; msg++)
@@ -225,7 +189,7 @@ public class ConflationQueueTest extends
}
// HACK to do something synchronous
- ((AMQSession)producerSession).sync();
+ ((AMQSession<?,?>)producerSession).sync();
consumer = consumerSession.createConsumer(queue);
consumerConnection.start();
@@ -252,7 +216,7 @@ public class ConflationQueueTest extends
}
// HACK to do something synchronous
- ((AMQSession)producerSession).sync();
+ ((AMQSession<?,?>)producerSession).sync();
// this causes the "old" messages to be released
@@ -284,17 +248,31 @@ public class ConflationQueueTest extends
}
+ public void testConflatedQueueDepth() throws Exception
+ {
+ consumerConnection = getConnection();
+ consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+
+ createConflationQueue(producerSession);
+ producer = producerSession.createProducer(queue);
+
+ for (int msg = 0; msg < MSG_COUNT; msg++)
+ {
+ producer.send(nextMessage(msg, producerSession));
+ }
+
+ final long queueDepth = ((AMQSession<?, ?>)producerSession).getQueueDepth((AMQDestination)queue, true);
+
+ assertEquals(10, queueDepth);
+ }
+
public void testConflationBrowser() throws Exception
{
consumerConnection = getConnection();
consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- final Map<String,Object> arguments = new HashMap<String, Object>();
- arguments.put("qpid.last_value_queue_key","key");
- ((AMQSession) producerSession).createQueue(new AMQShortString(QUEUE), false, true, false, arguments);
- queue = new org.apache.qpid.client.AMQQueue("amq.direct",QUEUE);
- ((AMQSession) producerSession).declareAndBind((AMQDestination)queue);
+ createConflationQueue(producerSession);
producer = producerSession.createProducer(queue);
for (int msg = 0; msg < MSG_COUNT; msg++)
@@ -303,9 +281,9 @@ public class ConflationQueueTest extends
}
- ((AMQSession)producerSession).sync();
+ ((AMQSession<?,?>)producerSession).sync();
- AMQBindingURL url = new AMQBindingURL("direct://amq.direct//"+QUEUE+"?browse='true'&durable='true'");
+ AMQBindingURL url = new AMQBindingURL("direct://amq.direct//"+queueName+"?browse='true'&durable='true'");
AMQQueue browseQueue = new AMQQueue(url);
consumer = consumerSession.createConsumer(browseQueue);
@@ -329,7 +307,7 @@ public class ConflationQueueTest extends
producer.send(nextMessage(MSG_COUNT, producerSession));
- ((AMQSession)producerSession).sync();
+ ((AMQSession<?,?>)producerSession).sync();
while((received = consumer.receive(1000))!=null)
{
@@ -354,22 +332,17 @@ public class ConflationQueueTest extends
consumerSession = consumerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- final Map<String,Object> arguments = new HashMap<String, Object>();
- arguments.put("qpid.last_value_queue_key","key");
- ((AMQSession) producerSession).createQueue(new AMQShortString(QUEUE), false, true, false, arguments);
- queue = new org.apache.qpid.client.AMQQueue("amq.direct",QUEUE);
- ((AMQSession) producerSession).declareAndBind((AMQDestination)queue);
+ createConflationQueue(producerSession);
producer = producerSession.createProducer(queue);
for (int msg = 0; msg < MSG_COUNT; msg++)
{
producer.send(nextMessage(msg, producerSession));
-
}
- ((AMQSession)producerSession).sync();
+ ((AMQSession<?,?>)producerSession).sync();
- AMQBindingURL url = new AMQBindingURL("direct://amq.direct//"+QUEUE+"?browse='true'&durable='true'");
+ AMQBindingURL url = new AMQBindingURL("direct://amq.direct//"+queueName+"?browse='true'&durable='true'");
AMQQueue browseQueue = new AMQQueue(url);
consumer = consumerSession.createConsumer(browseQueue);
@@ -412,12 +385,16 @@ public class ConflationQueueTest extends
producer.close();
producerSession.close();
producerConnection.close();
-
-
-
}
-
+ private void createConflationQueue(Session session) throws AMQException
+ {
+ final Map<String,Object> arguments = new HashMap<String, Object>();
+ arguments.put("qpid.last_value_queue_key","key");
+ ((AMQSession<?,?>) session).createQueue(new AMQShortString(queueName), false, true, false, arguments);
+ queue = new AMQQueue("amq.direct", queueName);
+ ((AMQSession<?,?>) session).declareAndBind((AMQDestination)queue);
+ }
private Message nextMessage(int msg, Session producerSession) throws JMSException
{
@@ -428,8 +405,4 @@ public class ConflationQueueTest extends
return send;
}
-
-
}
-
-
Modified: qpid/branches/java-config-and-management/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java?rev=1337045&r1=1337044&r2=1337045&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java Fri May 11 07:42:48 2012
@@ -17,11 +17,39 @@
*/
package org.apache.qpid.test.utils;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.PrintStream;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import javax.jms.BytesMessage;
+import javax.jms.Connection;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.MapMessage;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.ObjectMessage;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.StreamMessage;
+import javax.jms.TextMessage;
+import javax.jms.Topic;
+import javax.naming.InitialContext;
+import javax.naming.NamingException;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.XMLConfiguration;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
-
import org.apache.qpid.AMQException;
import org.apache.qpid.client.AMQConnectionFactory;
import org.apache.qpid.client.AMQQueue;
@@ -41,37 +69,6 @@ import org.apache.qpid.url.URLSyntaxExce
import org.apache.qpid.util.FileUtils;
import org.apache.qpid.util.LogMonitor;
-import javax.jms.BytesMessage;
-import javax.jms.Connection;
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.MapMessage;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.MessageProducer;
-import javax.jms.ObjectMessage;
-import javax.jms.Queue;
-import javax.jms.Session;
-import javax.jms.StreamMessage;
-import javax.jms.TextMessage;
-import javax.jms.Topic;
-import javax.naming.InitialContext;
-import javax.naming.NamingException;
-import java.io.File;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.PrintStream;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.TimeUnit;
-
/**
* Qpid base class for system testing test cases.
*/
@@ -219,10 +216,6 @@ public class QpidBrokerTestCase extends
out = new PrintStream(new FileOutputStream(_outputFile), true);
err = new PrintStream(String.format("%s/TEST-%s.err", _output, qname));
- // This is relying on behaviour specific to log4j 1.2.12. If we were to upgrade to 1.2.13 or
- // beyond we must change either code (or config) to ensure that ConsoleAppender#setFollow
- // is set to true otherwise log4j logging will not respect the following reassignment.
-
System.setOut(out);
System.setErr(err);
Propchange: qpid/branches/java-config-and-management/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/systests/src/main/java/org/apache/qpid/test/utils/QpidBrokerTestCase.java:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles:r1327945-1336832
Modified: qpid/branches/java-config-and-management/qpid/java/test-profiles/CPPExcludes
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/test-profiles/CPPExcludes?rev=1337045&r1=1337044&r2=1337045&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/test-profiles/CPPExcludes (original)
+++ qpid/branches/java-config-and-management/qpid/java/test-profiles/CPPExcludes Fri May 11 07:42:48 2012
@@ -121,6 +121,8 @@ org.apache.qpid.test.client.message.Sele
//QPID-942 : Implemented Channel.Flow based Producer Side flow control to the Java Broker (not in CPP Broker)
org.apache.qpid.server.queue.ProducerFlowControlTest#*
+//QPID-3986 : Flow control invoked on total store disk usage
+org.apache.qpid.server.store.StoreOverfullTest#*
org.apache.qpid.test.client.RollbackOrderTest#testOrderingAfterRollbackOnMessage#*
@@ -172,3 +174,8 @@ org.apache.qpid.scripts.QpidPasswdTest#*
// QPID-3604: Immediate Prefetch no longer supported by 0-10
org.apache.qpid.client.AsynchMessageListenerTest#testImmediatePrefetchWithMessageListener
+
+// QPID-3936: Exclude Performance Test Framework tests
+org.apache.qpid.systest.disttest.*
+org.apache.qpid.disttest.*
+
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/CPPExcludes
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/CPPExcludes:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/Excludes
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/Excludes:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/JavaBDBExcludes
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/JavaBDBExcludes:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/JavaExcludes
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/JavaExcludes:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/JavaPre010Excludes
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/JavaPre010Excludes:r1327945-1336832
Modified: qpid/branches/java-config-and-management/qpid/java/test-profiles/JavaTransientExcludes
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/test-profiles/JavaTransientExcludes?rev=1337045&r1=1337044&r2=1337045&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/test-profiles/JavaTransientExcludes (original)
+++ qpid/branches/java-config-and-management/qpid/java/test-profiles/JavaTransientExcludes Fri May 11 07:42:48 2012
@@ -30,7 +30,6 @@ org.apache.qpid.test.unit.xa.TopicTest#t
org.apache.qpid.test.unit.xa.TopicTest#testDurSubCrash
org.apache.qpid.test.unit.xa.TopicTest#testRecover
-
org.apache.qpid.server.store.MessageStoreTest#testMessagePersistence
org.apache.qpid.server.store.MessageStoreTest#testMessageRemoval
org.apache.qpid.server.store.MessageStoreTest#testBindingPersistence
@@ -39,6 +38,7 @@ org.apache.qpid.server.store.MessageStor
org.apache.qpid.server.store.MessageStoreTest#testDurableQueueRemoval
org.apache.qpid.server.store.MessageStoreTest#testExchangePersistence
org.apache.qpid.server.store.MessageStoreTest#testDurableExchangeRemoval
+org.apache.qpid.server.store.StoreOverfullTest#*
org.apache.qpid.server.store.berkeleydb.*
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/JavaTransientExcludes
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/JavaTransientExcludes:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/XAExcludes
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/XAExcludes:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/cpp.async.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/cpp.async.testprofile:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/cpp.cluster.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/cpp.cluster.testprofile:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/cpp.noprefetch.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/cpp.noprefetch.testprofile:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/cpp.ssl.excludes
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/cpp.ssl.excludes:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/cpp.ssl.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/cpp.ssl.testprofile:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/cpp.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/cpp.testprofile:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/java-bdb-spawn.0-9-1.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/java-bdb-spawn.0-9-1.testprofile:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/java-bdb.0-9-1.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/java-bdb.0-9-1.testprofile:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/java-dby-spawn.0-9-1.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/java-dby-spawn.0-9-1.testprofile:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/java-dby.0-9-1.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/java-dby.0-9-1.testprofile:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/java-mms-spawn.0-10.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/java-mms-spawn.0-10.testprofile:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/java-mms-spawn.0-9-1.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/java-mms-spawn.0-9-1.testprofile:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/java-mms.0-9-1.testprofile
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/java-mms.0-9-1.testprofile:r1327945-1336832
Modified: qpid/branches/java-config-and-management/qpid/java/test-profiles/log4j-test.xml
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/test-profiles/log4j-test.xml?rev=1337045&r1=1337044&r2=1337045&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/test-profiles/log4j-test.xml (original)
+++ qpid/branches/java-config-and-management/qpid/java/test-profiles/log4j-test.xml Fri May 11 07:42:48 2012
@@ -30,6 +30,7 @@
<appender name="console" class="org.apache.log4j.ConsoleAppender">
<param name="Target" value="System.out"/>
<param name="ImmediateFlush" value="true"/>
+ <param name="Follow" value="true"/>
<layout class="org.apache.log4j.PatternLayout">
<param name="ConversionPattern" value="%t %d %p [%c{4}] %m%n"/>
</layout>
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/log4j-test.xml
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/log4j-test.xml:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/test-provider.properties
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/test-provider.properties:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/test_resources/
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/test_resources:r1327945-1336832
Propchange: qpid/branches/java-config-and-management/qpid/java/test-profiles/testprofile.defaults
------------------------------------------------------------------------------
Merged /qpid/trunk/qpid/java/test-profiles/testprofile.defaults:r1327945-1336832
Modified: qpid/branches/java-config-and-management/qpid/java/tools/bin/Profile-run-from-source
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/tools/bin/Profile-run-from-source?rev=1337045&r1=1337044&r2=1337045&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/tools/bin/Profile-run-from-source (original)
+++ qpid/branches/java-config-and-management/qpid/java/tools/bin/Profile-run-from-source Fri May 11 07:42:48 2012
@@ -49,7 +49,7 @@ export LOG_CONFIG="-Dlog4j.configuration
#------------- Required for qpid-python-testkit -----------------------------------------
-PYTHONPATH=$QPID_CHECKOUT/python/qpid:$QPID_CHECKOUT/cpp/src/test/brokertest.py:$PYTHONPATH
+PYTHONPATH=$QPID_CHECKOUT/python:$QPID_CHECKOUT/cpp/src/test/brokertest.py:$PYTHONPATH
export PATH=$QPID_CHECKOUT/python:$PATH
if [ -x $QPID_CHECKOUT/cpp/src/qpidd ]; then
Propchange: qpid/branches/java-config-and-management/qpid/java/tools/bin/mercury-controller
------------------------------------------------------------------------------
svn:executable = *
Propchange: qpid/branches/java-config-and-management/qpid/java/tools/bin/mercury-start-consumers
------------------------------------------------------------------------------
svn:executable = *
Propchange: qpid/branches/java-config-and-management/qpid/java/tools/bin/mercury-start-producers
------------------------------------------------------------------------------
svn:executable = *
Modified: qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/Clock.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/Clock.java?rev=1337045&r1=1337044&r2=1337045&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/Clock.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/Clock.java Fri May 11 07:42:48 2012
@@ -20,6 +20,9 @@
*/
package org.apache.qpid.tools;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
/**
* In the future this will be replaced by a Clock abstraction
* that can utilize a realtime clock when running in RT Java.
@@ -27,6 +30,8 @@ package org.apache.qpid.tools;
public class Clock
{
+ private static final Logger _logger = LoggerFactory.getLogger(Clock.class);
+
public final static long SEC = 60000;
private static Precision precision;
@@ -54,7 +59,11 @@ public class Clock
precision = Precision.getPrecision(System.getProperty("precision","mili"));
//offset = Long.getLong("offset",-1);
- System.out.println("Using precision : " + precision + " and offset " + offset);
+ if (_logger.isDebugEnabled())
+ {
+ System.out.println("Using precision : " + precision );
+ //+ " and offset " + offset);
+ }
}
public static Precision getPrecision()
Modified: qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/JVMArgConfiguration.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/JVMArgConfiguration.java?rev=1337045&r1=1337044&r2=1337045&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/JVMArgConfiguration.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/JVMArgConfiguration.java Fri May 11 07:42:48 2012
@@ -29,383 +29,422 @@ import org.apache.qpid.client.AMQConnect
public class JVMArgConfiguration implements TestConfiguration
{
- /*
- * By default the connection URL is used.
- * This allows a user to easily specify a fully fledged URL any given property.
- * Ex. SSL parameters
- *
- * By providing a host & port allows a user to simply override the URL.
- * This allows to create multiple clients in test scripts easily,
- * without having to deal with the long URL format.
- */
- private String url = "amqp://guest:guest@clientid/testpath?brokerlist='tcp://localhost:5672'";
-
- private String host = "";
-
- private int port = -1;
-
- private String address = "queue; {create : always}";
-
- private int msg_size = 1024;
-
- private int random_msg_size_start_from = 1;
-
- private boolean cacheMessage = false;
-
- private boolean disableMessageID = false;
-
- private boolean disableTimestamp = false;
-
- private boolean durable = false;
-
- private int transaction_size = 0;
-
- private int ack_mode = Session.AUTO_ACKNOWLEDGE;
-
- private int msg_count = 10;
-
- private int warmup_count = 1;
-
- private boolean random_msg_size = false;
-
- private String msgType = "bytes";
-
- private boolean printStdDev = false;
-
- private int sendRate = 0;
-
- private boolean externalController = false;
-
- private boolean useUniqueDest = false; // useful when using multiple connections.
-
- private int ackFrequency = 100;
-
- private DecimalFormat df = new DecimalFormat("###.##");
-
- private int reportEvery = 0;
-
- private boolean isReportTotal = false;
-
- private boolean isReportHeader = true;
-
- private boolean isReportLatency = false;
-
- private int sendEOS = 0;
-
- private int connectionCount = 1;
-
- private int rollbackFrequency = 0;
-
- private boolean printHeaders;
-
- public JVMArgConfiguration()
- {
-
- url = System.getProperty("url",url);
- host = System.getProperty("host","");
- port = Integer.getInteger("port", -1);
- address = System.getProperty("address",address);
-
- msg_size = Integer.getInteger("msg-size", 1024);
- cacheMessage = Boolean.getBoolean("cache-msg");
- disableMessageID = Boolean.getBoolean("disable-message-id");
- disableTimestamp = Boolean.getBoolean("disable-timestamp");
- durable = Boolean.getBoolean("durable");
- transaction_size = Integer.getInteger("tx",1000);
- ack_mode = Integer.getInteger("ack-mode",Session.AUTO_ACKNOWLEDGE);
- msg_count = Integer.getInteger("msg-count",msg_count);
- warmup_count = Integer.getInteger("warmup-count",warmup_count);
- random_msg_size = Boolean.getBoolean("random-msg-size");
- msgType = System.getProperty("msg-type","bytes");
- printStdDev = Boolean.getBoolean("print-std-dev");
- sendRate = Integer.getInteger("rate",0);
- externalController = Boolean.getBoolean("ext-controller");
- useUniqueDest = Boolean.getBoolean("use-unique-dest");
- random_msg_size_start_from = Integer.getInteger("random-msg-size-start-from", 1);
- reportEvery = Integer.getInteger("report-every");
- isReportTotal = Boolean.getBoolean("report-total");
- isReportHeader = (System.getProperty("report-header") == null) ? true : Boolean.getBoolean("report-header");
- isReportLatency = Boolean.getBoolean("report-latency");
- sendEOS = Integer.getInteger("send-eos");
- connectionCount = Integer.getInteger("con_count",1);
- ackFrequency = Integer.getInteger("ack-frequency");
- rollbackFrequency = Integer.getInteger("rollback-frequency");
- printHeaders = Boolean.getBoolean("print-headers");
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#getUrl()
- */
- @Override
- public String getUrl()
- {
- return url;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#getHost()
- */
- @Override
- public String getHost()
- {
- return host;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#getPort()
- */
- @Override
- public int getPort()
- {
- return port;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#getAddress()
- */
- @Override
- public String getAddress()
- {
- return address;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#getAckMode()
- */
- @Override
- public int getAckMode()
- {
- return ack_mode;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#getMsgCount()
- */
- @Override
- public int getMsgCount()
- {
- return msg_count;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#getMsgSize()
- */
- @Override
- public int getMsgSize()
- {
- return msg_size;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#getRandomMsgSizeStartFrom()
- */
- @Override
- public int getRandomMsgSizeStartFrom()
- {
- return random_msg_size_start_from;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#isDurable()
- */
- @Override
- public boolean isDurable()
- {
- return durable;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#isTransacted()
- */
- @Override
- public boolean isTransacted()
- {
- return transaction_size > 0;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#getTransactionSize()
- */
- @Override
- public int getTransactionSize()
- {
- return transaction_size;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#getWarmupCount()
- */
- @Override
- public int getWarmupCount()
- {
- return warmup_count;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#isCacheMessage()
- */
- @Override
- public boolean isCacheMessage()
- {
- return cacheMessage;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#isDisableMessageID()
- */
- @Override
- public boolean isDisableMessageID()
- {
- return disableMessageID;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#isDisableTimestamp()
- */
- @Override
- public boolean isDisableTimestamp()
- {
- return disableTimestamp;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#isRandomMsgSize()
- */
- @Override
- public boolean isRandomMsgSize()
- {
- return random_msg_size;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#getMessageType()
- */
- @Override
- public String getMessageType()
- {
- return msgType;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#isPrintStdDev()
- */
- @Override
- public boolean isPrintStdDev()
- {
- return printStdDev;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#getSendRate()
- */
- @Override
- public int getSendRate()
- {
- return sendRate;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#isExternalController()
- */
- @Override
- public boolean isExternalController()
- {
- return externalController;
- }
-
- public void setAddress(String addr)
- {
- address = addr;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#isUseUniqueDests()
- */
- @Override
- public boolean isUseUniqueDests()
- {
- return useUniqueDest;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#getAckFrequency()
- */
- @Override
- public int getAckFrequency()
- {
- return ackFrequency;
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#createConnection()
- */
- @Override
- public Connection createConnection() throws Exception
- {
- if (getHost().equals("") || getPort() == -1)
- {
- return new AMQConnection(getUrl());
- }
- else
- {
- return new AMQConnection(getHost(),getPort(),"guest","guest","test","test");
- }
- }
-
- /* (non-Javadoc)
- * @see org.apache.qpid.tools.TestConfiguration#getDecimalFormat()
- */
- @Override
- public DecimalFormat getDecimalFormat()
- {
- return df;
- }
-
- @Override
- public int reportEvery()
- {
- return reportEvery;
- }
-
- @Override
- public boolean isReportTotal()
- {
- return isReportTotal;
- }
-
- @Override
- public boolean isReportHeader()
- {
- return isReportHeader;
- }
-
- @Override
- public boolean isReportLatency()
- {
- return isReportLatency;
- }
-
- @Override
- public int getSendEOS()
- {
- return sendEOS;
- }
-
- @Override
- public int getConnectionCount()
- {
- return connectionCount;
- }
-
- @Override
- public int getRollbackFrequency()
- {
- return rollbackFrequency;
- }
-
- @Override
- public boolean isPrintHeaders()
- {
- return printHeaders;
- }
+ /*
+ * By default the connection URL is used.
+ * This allows a user to easily specify a fully fledged URL any given property.
+ * Ex. SSL parameters
+ *
+ * By providing a host & port allows a user to simply override the URL.
+ * This allows to create multiple clients in test scripts easily,
+ * without having to deal with the long URL format.
+ */
+ private String url = "amqp://guest:guest@clientid/testpath?brokerlist='tcp://localhost:5672'";
+
+ private String host = "";
+
+ private int port = -1;
+
+ private String address = "queue; {create : always}";
+
+ private long timeout = 0;
+
+ private int msg_size = 1024;
+
+ private int random_msg_size_start_from = 1;
+
+ private boolean cacheMessage = false;
+
+ private boolean disableMessageID = false;
+
+ private boolean disableTimestamp = false;
+
+ private boolean durable = false;
+
+ private int transaction_size = 0;
+
+ private int ack_mode = Session.AUTO_ACKNOWLEDGE;
+
+ private int msg_count = 10;
+
+ private int warmup_count = 1;
+
+ private boolean random_msg_size = false;
+
+ private String msgType = "bytes";
+
+ private boolean printStdDev = false;
+
+ private int sendRate = 0;
+
+ private boolean externalController = false;
+
+ private boolean useUniqueDest = false; // useful when using multiple connections.
+
+ private int ackFrequency = 100;
+
+ private DecimalFormat df = new DecimalFormat("###.##");
+
+ private int reportEvery = 0;
+
+ private boolean isReportTotal = false;
+
+ private boolean isReportHeader = true;
+
+ private int sendEOS = 0;
+
+ private int connectionCount = 1;
+
+ private int rollbackFrequency = 0;
+
+ private boolean printHeaders;
+
+ private boolean printContent;
+
+ private long ttl;
+
+ private int priority;
+
+ private String readyAddress;
+
+ public JVMArgConfiguration()
+ {
+
+ url = System.getProperty("url",url);
+ host = System.getProperty("host","");
+ port = Integer.getInteger("port", -1);
+ address = System.getProperty("address",address);
+
+ timeout = Long.getLong("timeout",0);
+ msg_size = Integer.getInteger("msg-size", 0);
+ cacheMessage = true; //Boolean.getBoolean("cache-msg");
+ disableMessageID = Boolean.getBoolean("disable-message-id");
+ disableTimestamp = Boolean.getBoolean("disable-timestamp");
+ durable = Boolean.getBoolean("durable");
+ transaction_size = Integer.getInteger("tx",1000);
+ ack_mode = Integer.getInteger("ack-mode",Session.AUTO_ACKNOWLEDGE);
+ msg_count = Integer.getInteger("msg-count",msg_count);
+ warmup_count = Integer.getInteger("warmup-count",warmup_count);
+ random_msg_size = Boolean.getBoolean("random-msg-size");
+ msgType = System.getProperty("msg-type","bytes");
+ printStdDev = Boolean.getBoolean("print-std-dev");
+ sendRate = Integer.getInteger("rate",0);
+ externalController = Boolean.getBoolean("ext-controller");
+ useUniqueDest = Boolean.getBoolean("use-unique-dest");
+ random_msg_size_start_from = Integer.getInteger("random-msg-size-start-from", 1);
+ reportEvery = Integer.getInteger("report-every",0);
+ isReportTotal = Boolean.getBoolean("report-total");
+ isReportHeader = (System.getProperty("report-header") == null) ? true : Boolean.getBoolean("report-header");
+ sendEOS = Integer.getInteger("send-eos",1);
+ connectionCount = Integer.getInteger("con_count",1);
+ ackFrequency = Integer.getInteger("ack-frequency",100);
+ rollbackFrequency = Integer.getInteger("rollback-frequency",0);
+ printHeaders = Boolean.getBoolean("print-headers");
+ printContent = Boolean.getBoolean("print-content");
+ ttl = Long.getLong("ttl", 0);
+ priority = Integer.getInteger("priority", 0);
+ readyAddress = System.getProperty("ready-address");
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#getUrl()
+ */
+ @Override
+ public String getUrl()
+ {
+ return url;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#getHost()
+ */
+ @Override
+ public String getHost()
+ {
+ return host;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#getPort()
+ */
+ @Override
+ public int getPort()
+ {
+ return port;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#getAddress()
+ */
+ @Override
+ public String getAddress()
+ {
+ return address;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#getTimeout()
+ */
+ @Override
+ public long getTimeout()
+ {
+ return timeout;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#getAckMode()
+ */
+ @Override
+ public int getAckMode()
+ {
+ return ack_mode;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#getMsgCount()
+ */
+ @Override
+ public int getMsgCount()
+ {
+ return msg_count;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#getMsgSize()
+ */
+ @Override
+ public int getMsgSize()
+ {
+ return msg_size;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#getRandomMsgSizeStartFrom()
+ */
+ @Override
+ public int getRandomMsgSizeStartFrom()
+ {
+ return random_msg_size_start_from;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#isDurable()
+ */
+ @Override
+ public boolean isDurable()
+ {
+ return durable;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#isTransacted()
+ */
+ @Override
+ public boolean isTransacted()
+ {
+ return transaction_size > 0;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#getTransactionSize()
+ */
+ @Override
+ public int getTransactionSize()
+ {
+ return transaction_size;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#getWarmupCount()
+ */
+ @Override
+ public int getWarmupCount()
+ {
+ return warmup_count;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#isCacheMessage()
+ */
+ @Override
+ public boolean isCacheMessage()
+ {
+ return cacheMessage;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#isDisableMessageID()
+ */
+ @Override
+ public boolean isDisableMessageID()
+ {
+ return disableMessageID;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#isDisableTimestamp()
+ */
+ @Override
+ public boolean isDisableTimestamp()
+ {
+ return disableTimestamp;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#isRandomMsgSize()
+ */
+ @Override
+ public boolean isRandomMsgSize()
+ {
+ return random_msg_size;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#getMessageType()
+ */
+ @Override
+ public String getMessageType()
+ {
+ return msgType;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#isPrintStdDev()
+ */
+ @Override
+ public boolean isPrintStdDev()
+ {
+ return printStdDev;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#getSendRate()
+ */
+ @Override
+ public int getSendRate()
+ {
+ return sendRate;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#isExternalController()
+ */
+ @Override
+ public boolean isExternalController()
+ {
+ return externalController;
+ }
+
+ public void setAddress(String addr)
+ {
+ address = addr;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#isUseUniqueDests()
+ */
+ @Override
+ public boolean isUseUniqueDests()
+ {
+ return useUniqueDest;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#getAckFrequency()
+ */
+ @Override
+ public int getAckFrequency()
+ {
+ return ackFrequency;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#createConnection()
+ */
+ @Override
+ public Connection createConnection() throws Exception
+ {
+ if (getHost().equals("") || getPort() == -1)
+ {
+ return new AMQConnection(getUrl());
+ }
+ else
+ {
+ return new AMQConnection(getHost(),getPort(),"guest","guest","test","test");
+ }
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.qpid.tools.TestConfiguration#getDecimalFormat()
+ */
+ @Override
+ public DecimalFormat getDecimalFormat()
+ {
+ return df;
+ }
+
+ @Override
+ public int reportEvery()
+ {
+ return reportEvery;
+ }
+
+ @Override
+ public boolean isReportTotal()
+ {
+ return isReportTotal;
+ }
+
+ @Override
+ public boolean isReportHeader()
+ {
+ return isReportHeader;
+ }
+
+ @Override
+ public int getSendEOS()
+ {
+ return sendEOS;
+ }
+
+ @Override
+ public int getConnectionCount()
+ {
+ return connectionCount;
+ }
+
+ @Override
+ public int getRollbackFrequency()
+ {
+ return rollbackFrequency;
+ }
+
+ @Override
+ public boolean isPrintHeaders()
+ {
+ return printHeaders;
+ }
+
+ @Override
+ public boolean isPrintContent()
+ {
+ return printContent;
+ }
+
+ @Override
+ public long getTTL()
+ {
+ return ttl;
+ }
+
+ @Override
+ public int getPriority()
+ {
+ return priority;
+ }
+
+ @Override
+ public String getReadyAddress()
+ {
+ return readyAddress;
+ }
}
Modified: qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/QpidReceive.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/QpidReceive.java?rev=1337045&r1=1337044&r2=1337045&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/QpidReceive.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/QpidReceive.java Fri May 11 07:42:48 2012
@@ -28,10 +28,12 @@ import javax.jms.Destination;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
+import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.qpid.client.AMQDestination;
+import org.apache.qpid.client.message.AbstractJMSMessage;
import org.apache.qpid.tools.TestConfiguration.MessageType;
import org.apache.qpid.tools.report.BasicReporter;
import org.apache.qpid.tools.report.Reporter;
@@ -42,140 +44,164 @@ import org.slf4j.LoggerFactory;
public class QpidReceive implements MessageListener
{
- private static final Logger _logger = LoggerFactory.getLogger(QpidSend.class);
- private final CountDownLatch testCompleted = new CountDownLatch(1);
+ private static final Logger _logger = LoggerFactory.getLogger(QpidReceive.class);
+ private final CountDownLatch testCompleted = new CountDownLatch(1);
- private Connection con;
- private Session session;
- private Destination dest;
- private MessageConsumer consumer;
- private boolean transacted = false;
- private boolean isRollback = false;
- private int txSize = 0;
- private int rollbackFrequency = 0;
- private int ackFrequency = 0;
- private int expected = 0;
- private int received = 0;
- private Reporter report;
- private TestConfiguration config;
-
- public QpidReceive(Reporter report, TestConfiguration config, Connection con, Destination dest)
- {
- this(report,config, con, dest, UUID.randomUUID().toString());
- }
-
- public QpidReceive(Reporter report, TestConfiguration config, Connection con, Destination dest, String prefix)
- {
- //System.out.println("Producer ID : " + id);
- this.report = report;
- this.config = config;
- this.con = con;
- this.dest = dest;
- }
-
- public void setUp() throws Exception
- {
- if (config.isTransacted())
- {
- session = con.createSession(true, Session.SESSION_TRANSACTED);
- }
- else if (config.getAckFrequency() > 0)
- {
- session = con.createSession(false, Session.DUPS_OK_ACKNOWLEDGE);
- }
- else
- {
- session = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
- }
- consumer = session.createConsumer(dest);
- consumer.setMessageListener(this);
- System.out.println("Consumer: " + /*id +*/ " Receiving messages from : " + ((AMQDestination)dest).getAddressName() + "\n");
-
- transacted = config.isTransacted();
- txSize = config.getTransactionSize();
- isRollback = config.getRollbackFrequency() > 0;
- rollbackFrequency = config.getRollbackFrequency();
- ackFrequency = config.getAckFrequency();
- }
-
- public void resetCounters()
- {
- received = 0;
- expected = 0;
- report.clear();
- }
-
- public void onMessage(Message msg)
- {
- try
- {
- if (msg instanceof TextMessage &&
- TestConfiguration.EOS.equals(((TextMessage)msg).getText()))
- {
- testCompleted.countDown();
- return;
- }
-
- received++;
- report.message(msg);
-
- if (transacted && (received % txSize == 0))
- {
- if (isRollback && (received % rollbackFrequency == 0))
- {
- session.rollback();
- }
- else
- {
- session.commit();
- }
- }
- else if (ackFrequency > 0)
- {
- msg.acknowledge();
- }
-
- if (expected >= received)
- {
- testCompleted.countDown();
- }
-
- }
- catch(Exception e)
- {
- _logger.error("Error when receiving messages",e);
- }
-
- }
-
- public void waitforCompletion(int expected) throws Exception
- {
- this.expected = expected;
- testCompleted.await();
- }
-
- public void tearDown() throws Exception
- {
- session.close();
- }
-
- public static void main(String[] args) throws Exception
- {
- TestConfiguration config = new JVMArgConfiguration();
- Reporter reporter = new BasicReporter(config.isReportLatency()? ThroughputAndLatency.class : Throughput.class,
- System.out,
- config.reportEvery(),
- config.isReportHeader()
- );
- Destination dest = AMQDestination.createDestination(config.getAddress());
- QpidReceive receiver = new QpidReceive(reporter,config, config.createConnection(),dest);
- receiver.setUp();
- receiver.waitforCompletion(config.getMsgCount());
- if (config.isReportTotal())
- {
- reporter.report();
- }
- receiver.tearDown();
- }
+ private Connection con;
+ private Session session;
+ private Destination dest;
+ private MessageConsumer consumer;
+ private boolean transacted = false;
+ private boolean isRollback = false;
+ private int txSize = 0;
+ private int rollbackFrequency = 0;
+ private int ackFrequency = 0;
+ private int expected = 0;
+ private int received = 0;
+ private Reporter report;
+ private TestConfiguration config;
+
+ public QpidReceive(Reporter report, TestConfiguration config, Connection con, Destination dest)
+ {
+ this(report,config, con, dest, UUID.randomUUID().toString());
+ }
+
+ public QpidReceive(Reporter report, TestConfiguration config, Connection con, Destination dest, String prefix)
+ {
+ //System.out.println("Producer ID : " + id);
+ this.report = report;
+ this.config = config;
+ this.con = con;
+ this.dest = dest;
+ }
+
+ public void setUp() throws Exception
+ {
+ con.start();
+ if (config.isTransacted())
+ {
+ session = con.createSession(true, Session.SESSION_TRANSACTED);
+ }
+ else if (config.getAckFrequency() > 0)
+ {
+ session = con.createSession(false, Session.DUPS_OK_ACKNOWLEDGE);
+ }
+ else
+ {
+ session = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ }
+ consumer = session.createConsumer(dest);
+ consumer.setMessageListener(this);
+ if (_logger.isDebugEnabled())
+ {
+ System.out.println("Consumer: " + /*id +*/ " Receiving messages from : " + ((AMQDestination)dest).getAddressName() + "\n");
+ }
+
+ transacted = config.isTransacted();
+ txSize = config.getTransactionSize();
+ isRollback = config.getRollbackFrequency() > 0;
+ rollbackFrequency = config.getRollbackFrequency();
+ ackFrequency = config.getAckFrequency();
+
+ _logger.debug("Ready address : " + config.getReadyAddress());
+ if (config.getReadyAddress() != null)
+ {
+ MessageProducer prod = session.createProducer(AMQDestination
+ .createDestination(config.getReadyAddress()));
+ prod.send(session.createMessage());
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("Sending message to ready address " + prod.getDestination());
+ }
+ }
+ }
+
+ public void resetCounters()
+ {
+ received = 0;
+ expected = 0;
+ report.clear();
+ }
+
+ public void onMessage(Message msg)
+ {
+ try
+ {
+ if (msg instanceof TextMessage &&
+ TestConfiguration.EOS.equals(((TextMessage)msg).getText()))
+ {
+ testCompleted.countDown();
+ return;
+ }
+
+ received++;
+ report.message(msg);
+
+ if (config.isPrintHeaders())
+ {
+ System.out.println(((AbstractJMSMessage)msg).toHeaderString());
+ }
+
+ if (config.isPrintContent())
+ {
+ System.out.println(((AbstractJMSMessage)msg).toBodyString());
+ }
+
+ if (transacted && (received % txSize == 0))
+ {
+ if (isRollback && (received % rollbackFrequency == 0))
+ {
+ session.rollback();
+ }
+ else
+ {
+ session.commit();
+ }
+ }
+ else if (ackFrequency > 0)
+ {
+ msg.acknowledge();
+ }
+
+ if (received >= expected)
+ {
+ testCompleted.countDown();
+ }
+
+ }
+ catch(Exception e)
+ {
+ _logger.error("Error when receiving messages",e);
+ }
+ }
+
+ public void waitforCompletion(int expected) throws Exception
+ {
+ this.expected = expected;
+ testCompleted.await();
+ }
+
+ public void tearDown() throws Exception
+ {
+ session.close();
+ }
+
+ public static void main(String[] args) throws Exception
+ {
+ TestConfiguration config = new JVMArgConfiguration();
+ Reporter reporter = new BasicReporter(ThroughputAndLatency.class,
+ System.out,
+ config.reportEvery(),
+ config.isReportHeader());
+ Destination dest = AMQDestination.createDestination(config.getAddress());
+ QpidReceive receiver = new QpidReceive(reporter,config, config.createConnection(),dest);
+ receiver.setUp();
+ receiver.waitforCompletion(config.getMsgCount() + config.getSendEOS());
+ if (config.isReportTotal())
+ {
+ reporter.report();
+ }
+ receiver.tearDown();
+ }
}
Modified: qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/QpidSend.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/QpidSend.java?rev=1337045&r1=1337044&r2=1337045&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/QpidSend.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/QpidSend.java Fri May 11 07:42:48 2012
@@ -43,249 +43,261 @@ import org.slf4j.LoggerFactory;
public class QpidSend
{
- private Connection con;
- private Session session;
- private Destination dest;
- private MessageProducer producer;
- private MessageType msgType;
- private Message msg;
- private Object payload;
- private List<Object> payloads;
- private boolean cacheMsg = false;
- private boolean randomMsgSize = false;
- private boolean durable = false;
- private Random random;
- private int msgSizeRange = 1024;
- private int totalMsgCount = 0;
- private boolean rateLimitProducer = false;
- private boolean transacted = false;
- private int txSize = 0;
-
- private static final Logger _logger = LoggerFactory.getLogger(QpidSend.class);
- Reporter report;
- TestConfiguration config;
-
- public QpidSend(Reporter report, TestConfiguration config, Connection con, Destination dest)
- {
- this(report,config, con, dest, UUID.randomUUID().toString());
- }
-
- public QpidSend(Reporter report, TestConfiguration config, Connection con, Destination dest, String prefix)
- {
- //System.out.println("Producer ID : " + id);
- this.report = report;
- this.config = config;
- this.con = con;
- this.dest = dest;
- }
-
- public void setUp() throws Exception
- {
- if (config.isTransacted())
- {
- session = con.createSession(true, Session.SESSION_TRANSACTED);
- }
- else
- {
- session = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
- }
-
- durable = config.isDurable();
- rateLimitProducer = config.getSendRate() > 0 ? true : false;
- if (_logger.isDebugEnabled() && rateLimitProducer)
- {
- System.out.println("The test will attempt to limit the producer to " + config.getSendRate() + " msg/sec");
- }
-
- transacted = config.isTransacted();
- txSize = config.getTransactionSize();
-
- msgType = MessageType.getType(config.getMessageType());
- // if message caching is enabled we pre create the message
- // else we pre create the payload
- if (config.isCacheMessage())
- {
- cacheMsg = true;
- msg = createMessage(createPayload(config.getMsgSize()));
- msg.setJMSDeliveryMode(durable?
- DeliveryMode.PERSISTENT :
- DeliveryMode.NON_PERSISTENT
- );
- }
- else if (config.isRandomMsgSize())
- {
- random = new Random(20080921);
- randomMsgSize = true;
- msgSizeRange = config.getMsgSize();
- payloads = new ArrayList<Object>(msgSizeRange);
-
- for (int i=0; i < msgSizeRange; i++)
- {
- payloads.add(createPayload(i));
- }
- }
- else
- {
- payload = createPayload(config.getMsgSize());
- }
-
- producer = session.createProducer(dest);
- if (_logger.isDebugEnabled())
- {
- System.out.println("Producer: " + /*id +*/ " Sending messages to: " + ((AMQDestination)dest).getAddressName());
- }
- producer.setDisableMessageID(config.isDisableMessageID());
- producer.setDisableMessageTimestamp(config.isDisableTimestamp());
- }
-
- Object createPayload(int size)
- {
- if (msgType == MessageType.TEXT)
- {
- return MessageFactory.createMessagePayload(size);
- }
- else
- {
- return MessageFactory.createMessagePayload(size).getBytes();
- }
- }
-
- Message createMessage(Object payload) throws Exception
- {
- if (msgType == MessageType.TEXT)
- {
- return session.createTextMessage((String)payload);
- }
- else
- {
- BytesMessage m = session.createBytesMessage();
- m.writeBytes((byte[])payload);
- return m;
- }
- }
-
- protected Message getNextMessage() throws Exception
- {
- if (cacheMsg)
- {
- return msg;
- }
- else
- {
- Message m;
-
- if (!randomMsgSize)
- {
- m = createMessage(payload);
- }
- else
- {
- m = createMessage(payloads.get(random.nextInt(msgSizeRange)));
- }
- m.setJMSDeliveryMode(durable?
- DeliveryMode.PERSISTENT :
- DeliveryMode.NON_PERSISTENT
- );
- return m;
- }
- }
-
- public void commit() throws Exception
- {
- session.commit();
- }
-
- public void send() throws Exception
- {
- send(config.getMsgCount());
- }
-
- public void send(int count) throws Exception
- {
- int sendRate = config.getSendRate();
- if (rateLimitProducer)
- {
- int iterations = count/sendRate;
- int remainder = count%sendRate;
- for (int i=0; i < iterations; i++)
- {
- long iterationStart = Clock.getTime();
- sendMessages(sendRate);
- long elapsed = (Clock.getTime() - iterationStart)*Clock.convertToMiliSecs();
- long diff = Clock.SEC - elapsed;
- if (diff > 0)
- {
- // We have sent more messages in a sec than specified by the rate.
- Thread.sleep(diff);
- }
- }
- sendMessages(remainder);
- }
- else
- {
- sendMessages(count);
- }
- }
-
- private void sendMessages(int count) throws Exception
- {
- boolean isTimestamp = config.isReportLatency();
- for(int i=0; i < count; i++ )
- {
- Message msg = getNextMessage();
- if (isTimestamp)
- {
- msg.setLongProperty(TestConfiguration.TIMESTAMP, Clock.getTime());
- }
- producer.send(msg);
- report.message(msg);
- totalMsgCount++;
-
- if ( transacted && ((totalMsgCount) % txSize == 0))
- {
- session.commit();
- }
- }
- }
-
- public void resetCounters()
- {
- totalMsgCount = 0;
- report.clear();
- }
-
- public void sendEndMessage() throws Exception
- {
- Message msg = session.createMessage();
- msg.setBooleanProperty(TestConfiguration.EOS, true);
- producer.send(msg);
- }
-
- public void tearDown() throws Exception
- {
- session.close();
- }
-
- public static void main(String[] args) throws Exception
- {
- TestConfiguration config = new JVMArgConfiguration();
- Reporter reporter = new BasicReporter(Throughput.class,
- System.out,
- config.reportEvery(),
- config.isReportHeader()
- );
- Destination dest = AMQDestination.createDestination(config.getAddress());
- QpidSend sender = new QpidSend(reporter,config, config.createConnection(),dest);
- sender.setUp();
- sender.send();
- if (config.getSendEOS() > 0)
- {
- sender.sendEndMessage();
- }
- if (config.isReportTotal())
- {
- reporter.report();
- }
- sender.tearDown();
- }
+ private Connection con;
+ private Session session;
+ private Destination dest;
+ private MessageProducer producer;
+ private MessageType msgType;
+ private Message msg;
+ private Object payload;
+ private List<Object> payloads;
+ private boolean cacheMsg = false;
+ private boolean randomMsgSize = false;
+ private boolean durable = false;
+ private Random random;
+ private int msgSizeRange = 1024;
+ private int totalMsgCount = 0;
+ private boolean rateLimitProducer = false;
+ private boolean transacted = false;
+ private int txSize = 0;
+
+ private static final Logger _logger = LoggerFactory.getLogger(QpidSend.class);
+ Reporter report;
+ TestConfiguration config;
+
+ public QpidSend(Reporter report, TestConfiguration config, Connection con, Destination dest)
+ {
+ this(report,config, con, dest, UUID.randomUUID().toString());
+ }
+
+ public QpidSend(Reporter report, TestConfiguration config, Connection con, Destination dest, String prefix)
+ {
+ //System.out.println("Producer ID : " + id);
+ this.report = report;
+ this.config = config;
+ this.con = con;
+ this.dest = dest;
+ }
+
+ public void setUp() throws Exception
+ {
+ con.start();
+ if (config.isTransacted())
+ {
+ session = con.createSession(true, Session.SESSION_TRANSACTED);
+ }
+ else
+ {
+ session = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ }
+
+ durable = config.isDurable();
+ rateLimitProducer = config.getSendRate() > 0 ? true : false;
+ if (_logger.isDebugEnabled() && rateLimitProducer)
+ {
+ _logger.debug("The test will attempt to limit the producer to " + config.getSendRate() + " msg/sec");
+ }
+
+ transacted = config.isTransacted();
+ txSize = config.getTransactionSize();
+
+ msgType = MessageType.getType(config.getMessageType());
+ // if message caching is enabled we pre create the message
+ // else we pre create the payload
+ if (config.isCacheMessage())
+ {
+ cacheMsg = true;
+ msg = createMessage(createPayload(config.getMsgSize()));
+ msg.setJMSDeliveryMode(durable?
+ DeliveryMode.PERSISTENT :
+ DeliveryMode.NON_PERSISTENT
+ );
+ }
+ else if (config.isRandomMsgSize())
+ {
+ random = new Random(20080921);
+ randomMsgSize = true;
+ msgSizeRange = config.getMsgSize();
+ payloads = new ArrayList<Object>(msgSizeRange);
+
+ for (int i=0; i < msgSizeRange; i++)
+ {
+ payloads.add(createPayload(i));
+ }
+ }
+ else
+ {
+ payload = createPayload(config.getMsgSize());
+ }
+
+ producer = session.createProducer(dest);
+ if (_logger.isDebugEnabled())
+ {
+ _logger.debug("Producer: " + /*id +*/ " Sending messages to: " + ((AMQDestination)dest).getAddressName());
+ }
+ producer.setDisableMessageID(config.isDisableMessageID());
+ //we add a separate timestamp to allow interoperability with other clients.
+ producer.setDisableMessageTimestamp(true);
+ if (config.getTTL() > 0)
+ {
+ producer.setTimeToLive(config.getTTL());
+ }
+ if (config.getPriority() > 0)
+ {
+ producer.setPriority(config.getPriority());
+ }
+ }
+
+ Object createPayload(int size)
+ {
+ if (msgType == MessageType.TEXT)
+ {
+ return MessageFactory.createMessagePayload(size);
+ }
+ else
+ {
+ return MessageFactory.createMessagePayload(size).getBytes();
+ }
+ }
+
+ Message createMessage(Object payload) throws Exception
+ {
+ if (msgType == MessageType.TEXT)
+ {
+ return session.createTextMessage((String)payload);
+ }
+ else
+ {
+ BytesMessage m = session.createBytesMessage();
+ m.writeBytes((byte[])payload);
+ return m;
+ }
+ }
+
+ protected Message getNextMessage() throws Exception
+ {
+ if (cacheMsg)
+ {
+ return msg;
+ }
+ else
+ {
+ Message m;
+
+ if (!randomMsgSize)
+ {
+ m = createMessage(payload);
+ }
+ else
+ {
+ m = createMessage(payloads.get(random.nextInt(msgSizeRange)));
+ }
+ m.setJMSDeliveryMode(durable?
+ DeliveryMode.PERSISTENT :
+ DeliveryMode.NON_PERSISTENT
+ );
+ return m;
+ }
+ }
+
+ public void commit() throws Exception
+ {
+ session.commit();
+ }
+
+ public void send() throws Exception
+ {
+ send(config.getMsgCount());
+ }
+
+ public void send(int count) throws Exception
+ {
+ int sendRate = config.getSendRate();
+ if (rateLimitProducer)
+ {
+ int iterations = count/sendRate;
+ int remainder = count%sendRate;
+ for (int i=0; i < iterations; i++)
+ {
+ long iterationStart = System.currentTimeMillis();
+ sendMessages(sendRate);
+ long elapsed = System.currentTimeMillis() - iterationStart;
+ long diff = Clock.SEC - elapsed;
+ if (diff > 0)
+ {
+ // We have sent more messages in a sec than specified by the rate.
+ Thread.sleep(diff);
+ }
+ }
+ sendMessages(remainder);
+ }
+ else
+ {
+ sendMessages(count);
+ }
+ }
+
+ private void sendMessages(int count) throws Exception
+ {
+ boolean isTimestamp = !config.isDisableTimestamp();
+ long s = System.currentTimeMillis();
+ for(int i=0; i < count; i++ )
+ {
+ Message msg = getNextMessage();
+ if (isTimestamp)
+ {
+ msg.setLongProperty(TestConfiguration.TIMESTAMP, System.currentTimeMillis());
+ }
+ producer.send(msg);
+ //report.message(msg);
+ totalMsgCount++;
+
+ if ( transacted && ((totalMsgCount) % txSize == 0))
+ {
+ session.commit();
+ }
+ }
+ long e = System.currentTimeMillis() - s;
+ //System.out.println("Rate : " + totalMsgCount/e);
+ }
+
+ public void resetCounters()
+ {
+ totalMsgCount = 0;
+ report.clear();
+ }
+
+ public void sendEndMessage() throws Exception
+ {
+ Message msg = session.createTextMessage(TestConfiguration.EOS);
+ producer.send(msg);
+ }
+
+ public void tearDown() throws Exception
+ {
+ session.close();
+ }
+
+ public static void main(String[] args) throws Exception
+ {
+ TestConfiguration config = new JVMArgConfiguration();
+ Reporter reporter = new BasicReporter(Throughput.class,
+ System.out,
+ config.reportEvery(),
+ config.isReportHeader()
+ );
+ Destination dest = AMQDestination.createDestination(config.getAddress());
+ QpidSend sender = new QpidSend(reporter,config, config.createConnection(),dest);
+ sender.setUp();
+ sender.send();
+ if (config.getSendEOS() > 0)
+ {
+ sender.sendEndMessage();
+ }
+ if (config.isReportTotal())
+ {
+ reporter.report();
+ }
+ sender.tearDown();
+ }
}
Modified: qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/TestConfiguration.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/TestConfiguration.java?rev=1337045&r1=1337044&r2=1337045&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/TestConfiguration.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/TestConfiguration.java Fri May 11 07:42:48 2012
@@ -26,20 +26,20 @@ import javax.jms.Connection;
public interface TestConfiguration
{
- enum MessageType {
- BYTES, TEXT, MAP, OBJECT;
+ enum MessageType {
+ BYTES, TEXT, MAP, OBJECT;
- public static MessageType getType(String s) throws Exception
- {
- if ("text".equalsIgnoreCase(s))
- {
- return TEXT;
- }
- else if ("bytes".equalsIgnoreCase(s))
- {
- return BYTES;
- }
- /*else if ("map".equalsIgnoreCase(s))
+ public static MessageType getType(String s) throws Exception
+ {
+ if ("text".equalsIgnoreCase(s))
+ {
+ return TEXT;
+ }
+ else if ("bytes".equalsIgnoreCase(s))
+ {
+ return BYTES;
+ }
+ /*else if ("map".equalsIgnoreCase(s))
{
return MAP;
}
@@ -47,80 +47,88 @@ public interface TestConfiguration
{
return OBJECT;
}*/
- else
- {
- throw new Exception("Unsupported message type");
- }
- }
- };
+ else
+ {
+ throw new Exception("Unsupported message type");
+ }
+ }
+ };
+
+ public final static String TIMESTAMP = "ts";
+
+ public final static String EOS = "eos";
+
+ public final static String SEQUENCE_NUMBER = "sn";
+
+ public String getUrl();
- public final static String TIMESTAMP = "ts";
+ public String getHost();
- public final static String EOS = "eos";
+ public int getPort();
- public final static String SEQUENCE_NUMBER = "sn";
+ public String getAddress();
- public String getUrl();
+ public long getTimeout();
- public String getHost();
+ public int getAckMode();
- public int getPort();
+ public int getMsgCount();
- public String getAddress();
+ public int getMsgSize();
- public int getAckMode();
+ public int getRandomMsgSizeStartFrom();
- public int getMsgCount();
+ public boolean isDurable();
- public int getMsgSize();
+ public boolean isTransacted();
- public int getRandomMsgSizeStartFrom();
+ public int getTransactionSize();
- public boolean isDurable();
+ public int getWarmupCount();
- public boolean isTransacted();
+ public boolean isCacheMessage();
- public int getTransactionSize();
+ public boolean isDisableMessageID();
- public int getWarmupCount();
+ public boolean isDisableTimestamp();
- public boolean isCacheMessage();
+ public boolean isRandomMsgSize();
- public boolean isDisableMessageID();
+ public String getMessageType();
- public boolean isDisableTimestamp();
+ public boolean isPrintStdDev();
- public boolean isRandomMsgSize();
+ public int getSendRate();
- public String getMessageType();
+ public boolean isExternalController();
- public boolean isPrintStdDev();
+ public boolean isUseUniqueDests();
- public int getSendRate();
+ public int getAckFrequency();
- public boolean isExternalController();
+ public Connection createConnection() throws Exception;
- public boolean isUseUniqueDests();
+ public DecimalFormat getDecimalFormat();
- public int getAckFrequency();
+ public int reportEvery();
- public Connection createConnection() throws Exception;
+ public boolean isReportTotal();
- public DecimalFormat getDecimalFormat();
+ public boolean isReportHeader();
- public int reportEvery();
+ public int getSendEOS();
- public boolean isReportTotal();
+ public int getConnectionCount();
- public boolean isReportHeader();
+ public int getRollbackFrequency();
- public boolean isReportLatency();
+ public boolean isPrintHeaders();
- public int getSendEOS();
+ public boolean isPrintContent();
- public int getConnectionCount();
+ public long getTTL();
- public int getRollbackFrequency();
+ public int getPriority();
- public boolean isPrintHeaders();
+ public String getReadyAddress();
}
\ No newline at end of file
Modified: qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/report/Statistics.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/report/Statistics.java?rev=1337045&r1=1337044&r2=1337045&view=diff
==============================================================================
--- qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/report/Statistics.java (original)
+++ qpid/branches/java-config-and-management/qpid/java/tools/src/main/java/org/apache/qpid/tools/report/Statistics.java Fri May 11 07:42:48 2012
@@ -25,115 +25,121 @@ import java.text.DecimalFormat;
import javax.jms.Message;
+import org.apache.qpid.tools.TestConfiguration;
+
public interface Statistics
{
- public void message(Message msg);
- public void report(PrintStream out);
- public void header(PrintStream out);
- public void clear();
-
- static class Throughput implements Statistics
- {
- DecimalFormat df = new DecimalFormat("###.##");
- int messages = 0;
- long start = 0;
- boolean started = false;
-
- @Override
- public void message(Message msg)
- {
- ++messages;
- if (!started)
- {
- start = System.currentTimeMillis();
- started = true;
- }
- }
-
- @Override
- public void report(PrintStream out)
- {
- long elapsed = System.currentTimeMillis() - start;
- out.print(df.format((double)messages/(double)elapsed));
- }
-
- @Override
- public void header(PrintStream out)
- {
- out.print("tp(m/s)");
- }
-
- public void clear()
- {
- messages = 0;
- start = 0;
- started = false;
- }
- }
-
- static class ThroughputAndLatency extends Throughput
- {
- long minLatency = Long.MAX_VALUE;
- long maxLatency = Long.MIN_VALUE;
- double totalLatency = 0;
- int sampleCount = 0;
-
- @Override
- public void message(Message msg)
- {
- super.message(msg);
- try
- {
- long ts = msg.getLongProperty("ts");
- long latency = System.currentTimeMillis() - ts;
- minLatency = Math.min(latency, minLatency);
- maxLatency = Math.min(latency, maxLatency);
- totalLatency = totalLatency + latency;
- sampleCount++;
- }
- catch(Exception e)
- {
- System.out.println("Error calculating latency");
- }
- }
-
- @Override
- public void report(PrintStream out)
- {
- super.report(out);
- double avgLatency = totalLatency/(double)sampleCount;
- out.append('\t')
- .append(String.valueOf(minLatency))
- .append('\t')
- .append(String.valueOf(maxLatency))
- .append('\t')
- .append(df.format(avgLatency));
-
- out.flush();
- }
-
- @Override
- public void header(PrintStream out)
- {
- super.header(out);
- out.append('\t')
- .append("l-min")
- .append('\t')
- .append("l-max")
- .append('\t')
- .append("l-avg");
-
- out.flush();
- }
-
- public void clear()
- {
- super.clear();
- minLatency = 0;
- maxLatency = 0;
- totalLatency = 0;
- sampleCount = 0;
- }
- }
+ public void message(Message msg);
+ public void report(PrintStream out);
+ public void header(PrintStream out);
+ public void clear();
+
+ static class Throughput implements Statistics
+ {
+ DecimalFormat df = new DecimalFormat("###");
+ int messages = 0;
+ long start = 0;
+ boolean started = false;
+
+ @Override
+ public void message(Message msg)
+ {
+ ++messages;
+ if (!started)
+ {
+ start = System.currentTimeMillis();
+ started = true;
+ }
+ }
+
+ @Override
+ public void report(PrintStream out)
+ {
+ long elapsed = System.currentTimeMillis() - start;
+ out.println(df.format((double)messages/(double)elapsed));
+ }
+
+ @Override
+ public void header(PrintStream out)
+ {
+ out.println("tp(m/s)");
+ }
+
+ public void clear()
+ {
+ messages = 0;
+ start = 0;
+ started = false;
+ }
+ }
+
+ static class ThroughputAndLatency extends Throughput
+ {
+ long minLatency = Long.MAX_VALUE;
+ long maxLatency = Long.MIN_VALUE;
+ double totalLatency = 0;
+ int sampleCount = 0;
+
+ @Override
+ public void message(Message msg)
+ {
+ super.message(msg);
+ try
+ {
+ long ts = msg.getLongProperty(TestConfiguration.TIMESTAMP);
+ long latency = System.currentTimeMillis() - ts;
+ minLatency = Math.min(latency, minLatency);
+ maxLatency = Math.max(latency, maxLatency);
+ totalLatency = totalLatency + latency;
+ sampleCount++;
+ }
+ catch(Exception e)
+ {
+ System.out.println("Error calculating latency " + e);
+ }
+ }
+
+ @Override
+ public void report(PrintStream out)
+ {
+ long elapsed = System.currentTimeMillis() - start;
+ double rate = (double)messages/(double)elapsed;
+ double avgLatency = totalLatency/(double)sampleCount;
+ out.append("\n")
+ .append(df.format(rate))
+ .append('\t')
+ .append(String.valueOf(minLatency))
+ .append('\t')
+ .append(String.valueOf(maxLatency))
+ .append('\t')
+ .append(df.format(avgLatency))
+ .append("\n");
+
+ out.flush();
+ }
+
+ @Override
+ public void header(PrintStream out)
+ {
+ out.append("tp(m/s)")
+ .append('\t')
+ .append("l-min")
+ .append('\t')
+ .append("l-max")
+ .append('\t')
+ .append("l-avg");
+
+ out.flush();
+ }
+
+ public void clear()
+ {
+ super.clear();
+ minLatency = 0;
+ maxLatency = 0;
+ totalLatency = 0;
+ sampleCount = 0;
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org