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 2007/03/23 13:44:15 UTC
svn commit: r521705 [1/2] - in /incubator/qpid/branches/M2/java: ./
perftests/ perftests/src/main/java/org/apache/qpid/requestreply/
perftests/src/test/java/org/apache/qpid/ping/
perftests/src/test/java/org/apache/qpid/requestreply/
Author: rgreig
Date: Fri Mar 23 05:44:14 2007
New Revision: 521705
URL: http://svn.apache.org/viewvc?view=rev&rev=521705
Log:
Updates to performance tests.
Modified:
incubator/qpid/branches/M2/java/perftests/pom.xml
incubator/qpid/branches/M2/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
incubator/qpid/branches/M2/java/perftests/src/test/java/org/apache/qpid/ping/PingAsyncTestPerf.java
incubator/qpid/branches/M2/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java
incubator/qpid/branches/M2/java/perftests/src/test/java/org/apache/qpid/requestreply/PingPongTestPerf.java
incubator/qpid/branches/M2/java/pom.xml
Modified: incubator/qpid/branches/M2/java/perftests/pom.xml
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/M2/java/perftests/pom.xml?view=diff&rev=521705&r1=521704&r2=521705
==============================================================================
--- incubator/qpid/branches/M2/java/perftests/pom.xml (original)
+++ incubator/qpid/branches/M2/java/perftests/pom.xml Fri Mar 23 05:44:14 2007
@@ -158,32 +158,32 @@
<!-- More example Tests. These are examples to exercise all the features of the test harness. Can scale up with option overrides. -->
<Ping-Tx>-n Ping-Tx -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf transacted=true</Ping-Tx>
- <Ping-Size>-n Ping-Size -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf messagesize=512</Ping-Size>
+ <Ping-Size>-n Ping-Size -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf messageSize=512</Ping-Size>
<Ping-Concurrent>-n Ping-Concurrent -s [100] -c [4] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf</Ping-Concurrent>
- <Ping-Many-Queues>-n Ping-Many-Queues -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf destinationscount=4</Ping-Many-Queues>
+ <Ping-Many-Queues>-n Ping-Many-Queues -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf destinationCount=4</Ping-Many-Queues>
<Ping-Duration>-n Ping-Duration -s [100] -d10S -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf</Ping-Duration>
<Ping-Rate>-n Ping-Rate -s [100] -d10S -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf rate=500</Ping-Rate>
<Ping-PubSub>-n Ping-PubSub -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf pubsub=true</Ping-PubSub>
<Ping-Many-Topics>
- -n Ping-Many-Topics -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf pubsub=true destinationscount=4
+ -n Ping-Many-Topics -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf pubsub=true destinationCount=4
</Ping-Many-Topics>
<Ping-Persistent>
-n Ping-Persistent -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf persistent=true
</Ping-Persistent>
<Ping-Batch-Logging>
- -n Ping-Batch-Logging -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf BatchSize=10
+ -n Ping-Batch-Logging -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf batchSize=10
</Ping-Batch-Logging>
<Ping-Failover-Before-Send>
- -n Ping-Failover-Before-Send -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf CommitBatchSize=10 FailBeforeSend=true
+ -n Ping-Failover-Before-Send -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf commitBatchSize=10 failBeforeSend=true
</Ping-Failover-Before-Send>
<Ping-Failover-After-Send>
- -n Ping-Failover-After-Send -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf CommitBatchSize=10 FailAfterSend=true
+ -n Ping-Failover-After-Send -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf commitBatchSize=10 failAfterSend=true
</Ping-Failover-After-Send>
<Ping-Failover-Before-Commit>
- -n Ping-Failover-Before-Commit -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf CommitBatchSize=10 FailBeforeCommit=true
+ -n Ping-Failover-Before-Commit -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf commitBatchSize=10 failBeforeCommit=true
</Ping-Failover-Before-Commit>
<Ping-Failover-After-Commit>
- -n Ping-Failover-After-Commit -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf CommitBatchSize=10 FailAfterCommit=true
+ -n Ping-Failover-After-Commit -s [100] -o . -t testAsyncPingOk org.apache.qpid.ping.PingAsyncTestPerf commitBatchSize=10 failAfterCommit=true
</Ping-Failover-After-Commit>
</commands>
Modified: incubator/qpid/branches/M2/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/M2/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java?view=diff&rev=521705&r1=521704&r2=521705
==============================================================================
--- incubator/qpid/branches/M2/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java (original)
+++ incubator/qpid/branches/M2/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java Fri Mar 23 05:44:14 2007
@@ -40,279 +40,249 @@
import org.apache.qpid.client.AMQQueue;
import org.apache.qpid.client.AMQTopic;
import org.apache.qpid.client.message.TestMessageFactory;
+import org.apache.qpid.exchange.ExchangeDefaults;
import org.apache.qpid.jms.MessageProducer;
import org.apache.qpid.jms.Session;
import org.apache.qpid.topic.Config;
-import org.apache.qpid.exchange.ExchangeDefaults;
import uk.co.thebadgerset.junit.extensions.BatchedThrottle;
import uk.co.thebadgerset.junit.extensions.Throttle;
/**
- * PingPongProducer is a client that sends pings to a queue and waits for pongs to be bounced back by a bounce back
- * client (see {@link PingPongBouncer} for the bounce back client).
- * <p/>
- * <p/>The pings are sent with a reply-to field set to a single temporary queue, which is the same for all pings.
- * This means that this class has to do some work to correlate pings with pongs; it expectes the original message
- * correlation id in the ping to be bounced back in the reply correlation id.
- * <p/>
- * <p/>This ping tool accepts a vast number of configuration options, all of which are passed in to the constructor.
- * It can ping topics or queues; ping multiple destinations; do persistent pings; send messages of any size; do pings
- * within transactions; control the number of pings to send in each transaction; limit its sending rate; and perform
- * failover testing.
- * <p/>
+ * PingPongProducer is a client that sends test messages, and waits for replies to these messages. The replies may
+ * either be generated by another client (see {@link PingPongBouncer}, or an extension of it may be used that listens
+ * to its own messages and does not send replies (see {@link org.apache.qpid.ping.PingClient}). The intention of ping
+ * pong producer is that it is a swiss-army knife test client that makes almost every aspect of its behaviour
+ * configurable.
+ *
+ * <p/>The pings are sent with a reply-to field set to a single temporary queue, which is the same for all pings. This
+ * means that this class has to do some work to correlate pings with pongs; it expectes the original message correlation
+ * id in the ping to be bounced back in the reply correlation id.
+ *
+ * <p/>This ping tool accepts a vast number of configuration options, all of which are passed in to the constructor. It
+ * can ping topics or queues; ping multiple destinations; do persistent pings; send messages of any size; do pings within
+ * transactions; control the number of pings to send in each transaction; limit its sending rate; and perform failover
+ * testing. A complete list of accepted parameters, default values and comments on their usage is provided here:
+ *
+ * <p/><table><caption>CRC Card</caption>
+ * <tr><th> Parameter <th> Default <th> Comments
+ * <tr><td> messageSize <td> 0 <td> Message size in bytes. Not including any headers.
+ * <tr><td> destinationName <td> ping <td> The root name to use to generate destination names to ping.
+ * <tr><td> persistent <td> false <td> Determines whether peristent delivery is used.
+ * <tr><td> transacted <td> false <td> Determines whether messages are sent/received in transactions.
+ * <tr><td> broker <td> tcp://localhost:5672 <td> Determines the broker to connect to.
+ * <tr><td> virtualHost <td> test <td> Determines the virtual host to send all ping over.
+ * <tr><td> rate <td> 0 <td> The maximum rate (in hertz) to send messages at. 0 means no limit.
+ * <tr><td> verbose <td> false <td> The verbose flag for debugging. Prints to console on every message.
+ * <tr><td> pubsub <td> false <td> Whether to ping topics or queues. Uses p2p by default.
+ * <tr><td> failAfterCommit <td> false <td> Whether to prompt user to kill broker after a commit batch.
+ * <tr><td> failBeforeCommit <td> false <td> Whether to prompt user to kill broker before a commit batch.
+ * <tr><td> failAfterSend <td> false <td> Whether to prompt user to kill broker after a send.
+ * <tr><td> failBeforeSend <td> false <td> Whether to prompt user to kill broker before a send.
+ * <tr><td> failOnce <td> true <td> Whether to prompt for failover only once.
+ * <tr><td> username <td> guest <td> The username to access the broker with.
+ * <tr><td> password <td> guest <td> The password to access the broker with.
+ * <tr><td> selector <td> null <td> Not used. Defines a message selector to filter pings with.
+ * <tr><td> destinationCount <td> 1 <td> The number of receivers listening to the pings.
+ * <tr><td> timeout <td> 30000 <td> In milliseconds. The timeout to stop waiting for replies.
+ * <tr><td> commitBatchSize <td> 1 <td> The number of messages per transaction in transactional mode.
+ * <tr><td> uniqueDests <td> true <td> Whether each receiver only listens to one ping destination or all.
+ * <tr><td> ackMode <td> NO_ACK <td> The message acknowledgement mode.
+ * <tr><td> pauseBatch <td> 0 <td> In milliseconds. A pause to insert between transaction batches.
+ * </table>
+ *
* <p/>This implements the Runnable interface with a run method that implements an infinite ping loop. The ping loop
- * does all its work through helper methods, so that code wishing to run a ping-pong cycle is not forced to do so
- * by starting a new thread. The command line invocation does take advantage of this ping loop. A shutdown hook is
- * also registered to terminate the ping-pong loop cleanly.
- * <p/>
+ * does all its work through helper methods, so that code wishing to run a ping-pong cycle is not forced to do so by
+ * starting a new thread. The command line invocation does take advantage of this ping loop. A shutdown hook is also
+ * registered to terminate the ping-pong loop cleanly.
+ *
* <p/><table id="crc"><caption>CRC Card</caption>
* <tr><th> Responsibilities <th> Collaborations
* <tr><td> Provide a ping and wait for all responses cycle.
* <tr><td> Provide command line invocation to loop the ping cycle on a configurable broker url.
* </table>
*
- * @todo The use of a ping rate {@link #DEFAULT_RATE} and waits between pings {@link #DEFAULT_SLEEP_TIME} are overlapping.
- * Use the rate and throttling only. Ideally, optionally pass the rate throttle into the ping method, throttle to
- * be created and configured by the test runner from the -f command line option and made available through
- * the timing controller on timing aware tests or by throttling rate of calling tests methods on non-timing aware
- * tests.
- * @todo Make acknowledege mode a test option.
- * @todo Make the message listener a static for all replies to be sent to? It won't be any more of a bottle neck than
- * having one per PingPongProducer, as will synchronize on message correlation id, allowing threads to process
- * messages concurrently for different ids. Needs to be static so that when using a chained message listener and
- * shared destinations between multiple PPPs, it gets notified about all replies, not just those that happen to
- * be picked up by the PPP that it is atteched to.
- * @todo Use read/write lock in the onmessage, not for reading writing but to make use of a shared and exlcusive lock
- * pair. Obtian read lock on all messages, before decrementing the message count. At the end of the on message
- * method add a block that obtains the write lock for the very last message, releases any waiting producer. Means
- * that the last message waits until all other messages have been handled before releasing producers but allows
- * messages to be processed concurrently, unlike the current synchronized block.
- * @todo Need to multiply up the number of expected messages for pubsub tests as each can be received by many consumers?
+ * @todo Make the message listener a static for all replies to be sent to? It won't be any more of a bottle neck than having
+ * one per PingPongProducer, as will synchronize on message correlation id, allowing threads to process messages
+ * concurrently for different ids. Needs to be static so that when using a chained message listener and shared
+ * destinations between multiple PPPs, it gets notified about all replies, not just those that happen to be picked up
+ * by the PPP that it is atteched to.
+ *
+ * @todo Use read/write lock in the onmessage, not for reading writing but to make use of a shared and exlcusive lock pair.
+ * Obtian read lock on all messages, before decrementing the message count. At the end of the on message method add a
+ * block that obtains the write lock for the very last message, releases any waiting producer. Means that the last
+ * message waits until all other messages have been handled before releasing producers but allows messages to be
+ * processed concurrently, unlike the current synchronized block.
+ *
+ * @todo Get rid of pauses between batches, it will impact the timing statistics, and generate meanigless timings.
+ * Instead make mina use a bounded blocking buffer, or other form of back pressure, to stop data being written
+ * faster than it can be sent.
*/
public class PingPongProducer implements Runnable, MessageListener, ExceptionListener
{
private static final Logger _logger = Logger.getLogger(PingPongProducer.class);
- /**
- * Holds the name of the property to get the test message size from.
- */
- public static final String MESSAGE_SIZE_PROPNAME = "messagesize";
+ /** Holds the name of the property to get the test message size from. */
+ public static final String MESSAGE_SIZE_PROPNAME = "messageSize";
- /**
- * Holds the name of the property to get the ping queue name from.
- */
- public static final String PING_QUEUE_NAME_PROPNAME = "destinationname";
+ /** Used to set up a default message size. */
+ public static final int MESSAGE_SIZE_DEAFULT = 0;
- /**
- * Holds the name of the property to get the test delivery mode from.
- */
- public static final String PERSISTENT_MODE_PROPNAME = "persistent";
+ /** Holds the name of the property to get the ping queue name from. */
+ public static final String PING_QUEUE_NAME_PROPNAME = "destinationName";
- /**
- * Holds the name of the property to get the test transactional mode from.
- */
- public static final String TRANSACTED_PROPNAME = "transacted";
+ /** Holds the name of the default destination to send pings on. */
+ public static final String PING_QUEUE_NAME_DEFAULT = "ping";
- /**
- * Holds the name of the property to get the test broker url from.
- */
- public static final String BROKER_PROPNAME = "broker";
+ /** Holds the name of the property to get the test delivery mode from. */
+ public static final String PERSISTENT_MODE_PROPNAME = "persistent";
- /**
- * Holds the name of the property to get the test broker virtual path.
- */
- public static final String VIRTUAL_PATH_PROPNAME = "virtualPath";
+ /** Holds the message delivery mode to use for the test. */
+ public static final boolean PERSISTENT_MODE_DEFAULT = false;
- /**
- * Holds the name of the property to get the message rate from.
- */
- public static final String RATE_PROPNAME = "rate";
+ /** Holds the name of the property to get the test transactional mode from. */
+ public static final String TRANSACTED_PROPNAME = "transacted";
- public static final String VERBOSE_OUTPUT_PROPNAME = "verbose";
+ /** Holds the transactional mode to use for the test. */
+ public static final boolean TRANSACTED_DEFAULT = false;
- /**
- * Holds the true or false depending on wether it is P2P test or PubSub
- */
- public static final String IS_PUBSUB_PROPNAME = "pubsub";
+ /** Holds the name of the property to get the test broker url from. */
+ public static final String BROKER_PROPNAME = "broker";
- public static final String FAIL_AFTER_COMMIT_PROPNAME = "FailAfterCommit";
+ /** Holds the default broker url for the test. */
+ public static final String BROKER_DEFAULT = "tcp://localhost:5672";
- public static final String FAIL_BEFORE_COMMIT_PROPNAME = "FailBeforeCommit";
+ /** Holds the name of the property to get the test broker virtual path. */
+ public static final String VIRTUAL_HOST_PROPNAME = "virtualHost";
- public static final String FAIL_AFTER_SEND_PROPNAME = "FailAfterSend";
+ /** Holds the default virtual path for the test. */
+ public static final String VIRTUAL_HOST_DEFAULT = "";
- public static final String FAIL_BEFORE_SEND_PROPNAME = "FailBeforeSend";
+ /** Holds the name of the property to get the message rate from. */
+ public static final String RATE_PROPNAME = "rate";
- public static final String FAIL_ONCE_PROPNAME = "FailOnce";
+ /** Defines the default rate (in pings per second) to send pings at. 0 means as fast as possible, no restriction. */
+ public static final int RATE_DEFAULT = 0;
- public static final String USERNAME_PROPNAME = "username";
+ /** Holds the name of the property to get the verbose mode proeprty from. */
+ public static final String VERBOSE_PROPNAME = "verbose";
- public static final String PASSWORD_PROPNAME = "password";
+ /** Holds the default verbose mode. */
+ public static final boolean VERBOSE_DEFAULT = false;
- public static final String SELECTOR_PROPNAME = "selector";
+ /** Holds the name of the property to get the p2p or pub/sub messaging mode from. */
+ public static final String PUBSUB_PROPNAME = "pubsub";
- public static final String PING_DESTINATION_COUNT_PROPNAME = "destinationscount";
+ /** Holds the pub/sub mode default, true means ping a topic, false means ping a queue. */
+ public static final boolean PUBSUB_DEFAULT = false;
- /**
- * Holds the name of the property to get the waiting timeout for response messages.
- */
- public static final String TIMEOUT_PROPNAME = "timeout";
+ /** Holds the name of the property to get the fail after commit flag from. */
+ public static final String FAIL_AFTER_COMMIT_PROPNAME = "failAfterCommit";
- public static final String COMMIT_BATCH_SIZE_PROPNAME = "CommitBatchSize";
+ /** Holds the default failover after commit test flag. */
+ public static final String FAIL_AFTER_COMMIT_DEFAULT = "false";
- public static final String UNIQUE_PROPNAME = "uniqueDests";
+ /** Holds the name of the proeprty to get the fail before commit flag from. */
+ public static final String FAIL_BEFORE_COMMIT_PROPNAME = "failBeforeCommit";
- public static final String ACK_MODE_PROPNAME = "ackMode";
+ /** Holds the default failover before commit test flag. */
+ public static final String FAIL_BEFORE_COMMIT_DEFAULT = "false";
- public static final String PAUSE_AFTER_BATCH_PROPNAME = "pausetimeAfterEachBatch";
-
- /**
- * Used to set up a default message size.
- */
- public static final int DEFAULT_MESSAGE_SIZE = 0;
+ /** Holds the name of the proeprty to get the fail after send flag from. */
+ public static final String FAIL_AFTER_SEND_PROPNAME = "failAfterSend";
- /**
- * Holds the name of the default destination to send pings on.
- */
- public static final String DEFAULT_PING_DESTINATION_NAME = "ping";
+ /** Holds the default failover after send test flag. */
+ public static final String FAIL_AFTER_SEND_DEFAULT = "false";
- /**
- * Defines the default number of destinations to ping.
- */
- public static final int DEFAULT_DESTINATION_COUNT = 1;
+ /** Holds the name of the property to get the fail before send flag from. */
+ public static final String FAIL_BEFORE_SEND_PROPNAME = "failBeforeSend";
- /**
- * Defines the default rate (in pings per second) to send pings at. 0 means as fast as possible, no restriction.
- */
- public static final int DEFAULT_RATE = 0;
+ /** Holds the default failover before send test flag. */
+ public static final String FAIL_BEFORE_SEND_DEFAULT = "false";
- /**
- * Defines the default wait between pings.
- */
- public static final long DEFAULT_SLEEP_TIME = 250;
+ /** Holds the name of the property to get the fail once flag from. */
+ public static final String FAIL_ONCE_PROPNAME = "failOnce";
- /**
- * Default time to wait before assuming that a ping has timed out.
- */
- public static final long DEFAULT_TIMEOUT = 30000;
+ /** The default failover once flag, true means only do one failover, false means failover on every commit cycle. */
+ public static final String FAIL_ONCE_DEFAULT = "true";
- /**
- * Defines the default number of pings to send in each transaction when running transactionally.
- */
- public static final int DEFAULT_TX_BATCH_SIZE = 100;
+ /** Holds the name of the property to get the broker access username from. */
+ public static final String USERNAME_PROPNAME = "username";
- /**
- * Defines the default prefetch size to use when consuming messages.
- */
- public static final int DEFAULT_PREFETCH = 100;
+ /** Holds the default broker log on username. */
+ public static final String USERNAME_DEFAULT = "guest";
- /**
- * Defines the default value of the no local flag to use when consuming messages.
- */
- public static final boolean DEFAULT_NO_LOCAL = false;
+ /** Holds the name of the property to get the broker access password from. */
+ public static final String PASSWORD_PROPNAME = "password";
- /**
- * Defines the default value of the exclusive flag to use when consuming messages.
- */
- public static final boolean DEFAULT_EXCLUSIVE = false;
+ /** Holds the default broker log on password. */
+ public static final String PASSWORD_DEFAULT = "guest";
- /**
- * Holds the message delivery mode to use for the test.
- */
- public static final boolean DEFAULT_PERSISTENT_MODE = false;
+ /** Holds the name of the proeprty to get the. */
+ public static final String SELECTOR_PROPNAME = "selector";
- /**
- * Holds the transactional mode to use for the test.
- */
- public static final boolean DEFAULT_TRANSACTED = false;
+ /** Holds the default message selector. */
+ public static final String SELECTOR_DEFAULT = null;
- /**
- * Holds the default broker url for the test.
- */
- public static final String DEFAULT_BROKER = "tcp://localhost:5672";
+ /** Holds the name of the proeprty to get the destination count from. */
+ public static final String DESTINATION_COUNT_PROPNAME = "destinationCount";
- /**
- * Holds the default virtual path for the test.
- */
- public static final String DEFAULT_VIRTUAL_PATH = "test";
+ /** Defines the default number of destinations to ping. */
+ public static final int DESTINATION_COUNT_DEFAULT = 1;
- /**
- * Holds the pub/sub mode default, true means ping a topic, false means ping a queue.
- */
- public static final boolean DEFAULT_PUBSUB = false;
+ /** Holds the name of the property to get the waiting timeout for response messages. */
+ public static final String TIMEOUT_PROPNAME = "timeout";
- /**
- * Holds the default broker log on username.
- */
- public static final String DEFAULT_USERNAME = "guest";
+ /** Default time to wait before assuming that a ping has timed out. */
+ public static final long TIMEOUT_DEFAULT = 30000;
- /**
- * Holds the default broker log on password.
- */
- public static final String DEFAULT_PASSWORD = "guest";
+ /** Holds the name of the property to get the commit batch size from. */
+ public static final String TX_BATCH_SIZE_PROPNAME = "commitBatchSize";
- /**
- * Holds the default message selector.
- */
- public static final String DEFAULT_SELECTOR = null;
+ /** Defines the default number of pings to send in each transaction when running transactionally. */
+ public static final int TX_BATCH_SIZE_DEFAULT = 1;
- /**
- * Holds the default failover after commit test flag.
- */
- public static final String DEFAULT_FAIL_AFTER_COMMIT = "false";
+ /** Holds the name of the property to get the unique destinations flag from. */
+ public static final String UNIQUE_DESTS_PROPNAME = "uniqueDests";
- /**
- * Holds the default failover before commit test flag.
- */
- public static final String DEFAULT_FAIL_BEFORE_COMMIT = "false";
+ /** Defines the default value for the unique destinations property. */
+ public static final boolean UNIQUE_DESTS_DEFAULT = true;
- /**
- * Holds the default failover after send test flag.
- */
- public static final String DEFAULT_FAIL_AFTER_SEND = "false";
+ /** Holds the name of the proeprty to get the message acknowledgement mode from. */
+ public static final String ACK_MODE_PROPNAME = "ackMode";
- /**
- * Holds the default failover before send test flag.
- */
- public static final String DEFAULT_FAIL_BEFORE_SEND = "false";
+ /** Defines the default message acknowledgement mode. */
+ public static final int ACK_MODE_DEFAULT = Session.NO_ACKNOWLEDGE;
- /**
- * Holds the default failover only once flag, true means only do one failover, false means failover on every commit cycle.
- */
- public static final String DEFAULT_FAIL_ONCE = "true";
+ /** Holds the name of the property to get the pause between batches property from. */
+ public static final String PAUSE_AFTER_BATCH_PROPNAME = "pauseBatch";
- /**
- * Holds the default verbose mode.
- */
- public static final boolean DEFAULT_VERBOSE = false;
+ /** Defines the default time in milliseconds to wait between commit batches. */
+ public static final long PAUSE_AFTER_BATCH_DEFAULT = 0L;
- public static final boolean DEFAULT_UNIQUE = true;
+ /** Defines the default prefetch size to use when consuming messages. */
+ public static final int PREFETCH_DEFAULT = 100;
- public static final int DEFAULT_ACK_MODE = Session.NO_ACKNOWLEDGE;
+ /** Defines the default value of the no local flag to use when consuming messages. */
+ public static final boolean NO_LOCAL_DEFAULT = false;
- /**
- * Holds the name of the property to store nanosecond timestamps in ping messages with.
- */
+ /** Defines the default value of the exclusive flag to use when consuming messages. */
+ public static final boolean EXCLUSIVE_DEFAULT = false;
+
+ /** Holds the name of the property to store nanosecond timestamps in ping messages with. */
public static final String MESSAGE_TIMESTAMP_PROPNAME = "timestamp";
- /**
- * A source for providing sequential unique correlation ids. These will be unique within the same JVM.
- */
+ /** A source for providing sequential unique correlation ids. These will be unique within the same JVM. */
private static AtomicLong _correlationIdGenerator = new AtomicLong(0L);
/**
- * Holds a map from message ids to latches on which threads wait for replies. This map is shared accross
- * multiple ping producers on the same JVM.
+ * Holds a map from message ids to latches on which threads wait for replies. This map is shared accross multiple
+ * ping producers on the same JVM.
*/
- /*private static Map<String, CountDownLatch> trafficLights =
- Collections.synchronizedMap(new HashMap<String, CountDownLatch>());*/
private static Map<String, PerCorrelationId> perCorrelationIds =
- Collections.synchronizedMap(new HashMap<String, PerCorrelationId>());
+ Collections.synchronizedMap(new HashMap<String, PerCorrelationId>());
- /**
- * A convenient formatter to use when time stamping output.
- */
+ /** A convenient formatter to use when time stamping output. */
protected static final DateFormat timestampFormatter = new SimpleDateFormat("hh:mm:ss:SS");
/**
@@ -321,52 +291,35 @@
*/
protected static AtomicInteger _queueJVMSequenceID = new AtomicInteger();
- /**
- * Destination where the response messages will arrive.
- */
+ /** Holds the destination where the response messages will arrive. */
private Destination _replyDestination;
- /**
- * Determines whether this producer sends persistent messages.
- */
+ /** Determines whether this producer sends persistent messages. */
protected boolean _persistent;
+ /** Holds the acknowledgement mode used for sending and receiving messages. */
private int _ackMode = Session.NO_ACKNOWLEDGE;
- /**
- * Determines what size of messages this producer sends.
- */
+ /** Determines what size of messages this producer sends. */
protected int _messageSize;
- /**
- * Used to indicate that the ping loop should print out whenever it pings.
- */
- protected boolean _verbose = false;
+ /** Used to indicate that the ping loop should print out whenever it pings. */
+ protected boolean _verbose = VERBOSE_DEFAULT;
- /**
- * Holds the session on which ping replies are received.
- */
+ /** Holds the session on which ping replies are received. */
protected Session _consumerSession;
- /**
- * Used to restrict the sending rate to a specified limit.
- */
+ /** Used to restrict the sending rate to a specified limit. */
private Throttle _rateLimiter = null;
- /**
- * Holds a message listener that this message listener chains all its messages to.
- */
+ /** Holds a message listener that this message listener chains all its messages to. */
private ChainedMessageListener _chainedMessageListener = null;
- /**
- * Flag used to indicate if this is a point to point or pub/sub ping client.
- */
- protected boolean _isPubSub = false;
+ /** Flag used to indicate if this is a point to point or pub/sub ping client. */
+ protected boolean _isPubSub = PUBSUB_DEFAULT;
- /**
- * Flag used to indicate if the destinations should be unique client.
- */
- protected static boolean _isUnique = false;
+ /** Flag used to indicate if the destinations should be unique client. */
+ protected static boolean _isUnique = UNIQUE_DESTS_DEFAULT;
/**
* This id generator is used to generates ids that are only unique within this pinger. Creating multiple pingers
@@ -374,96 +327,79 @@
*/
protected AtomicInteger _queueSharedId = new AtomicInteger();
- /**
- * Used to tell the ping loop when to terminate, it only runs while this is true.
- */
+ /** Used to tell the ping loop when to terminate, it only runs while this is true. */
protected boolean _publish = true;
- /**
- * Holds the connection to the broker.
- */
+ /** Holds the connection to the broker. */
private Connection _connection;
- /**
- * Holds the producer session, needed to create ping messages.
- */
+ /** Holds the producer session, needed to create ping messages. */
private Session _producerSession;
- /**
- * Holds the set of destiniations that this ping producer pings.
- */
+ /** Holds the set of destinations that this ping producer pings. */
protected List<Destination> _pingDestinations = new ArrayList<Destination>();
- /**
- * Holds the message producer to send the pings through.
- */
+ /** Holds the message producer to send the pings through. */
protected MessageProducer _producer;
- /**
- * Flag used to indicate that the user should be prompted to terminate a broker, to test failover before a commit.
- */
+ /** Flag used to indicate that the user should be prompted to terminate a broker, to test failover before a commit. */
protected boolean _failBeforeCommit = false;
- /**
- * Flag used to indicate that the user should be prompted to terminate a broker, to test failover after a commit.
- */
+ /** Flag used to indicate that the user should be prompted to terminate a broker, to test failover after a commit. */
protected boolean _failAfterCommit = false;
- /**
- * Flag used to indicate that the user should be prompted to terminate a broker, to test failover before a send.
- */
+ /** Flag used to indicate that the user should be prompted to terminate a broker, to test failover before a send. */
protected boolean _failBeforeSend = false;
- /**
- * Flag used to indicate that the user should be prompted to terminate a broker, to test failover after a send.
- */
+ /** Flag used to indicate that the user should be prompted to terminate a broker, to test failover after a send. */
protected boolean _failAfterSend = false;
- /**
- * Flag used to indicate that failover prompting should only be done on the first commit, not on every commit.
- */
+ /** Flag used to indicate that failover prompting should only be done on the first commit, not on every commit. */
protected boolean _failOnce = true;
- /**
- * Holds the number of sends that should be performed in every transaction when using transactions.
- */
- protected int _txBatchSize = 1;
+ /** Holds the number of sends that should be performed in every transaction when using transactions. */
+ protected int _txBatchSize = TX_BATCH_SIZE_DEFAULT;
- private static long _pausetimeAfterEachBatch = 0;
+ /** Holds the wait time to insert between every batch of messages committed. */
+ private static long _pauseBatch = PAUSE_AFTER_BATCH_DEFAULT;
/**
- * Holds the number of consumers that will be attached to each topic.
- * Each pings will result in a reply from each of the attached clients
+ * Holds the number of consumers that will be attached to each topic. Each pings will result in a reply from each of the
+ * attached clients
*/
static int _consumersPerTopic = 1;
/**
- * Creates a ping producer with the specified parameters, of which there are many. See their individual comments
- * for details. This constructor creates a connection to the broker and creates producer and consumer sessions on it,
- * to send and recieve its pings and replies on. The other options are kept, and control how this pinger behaves.
- *
- * @param brokerDetails The URL of the broker to send pings to.
- * @param username The username to log onto the broker with.
- * @param password The password to log onto the broker with.
- * @param virtualpath The virtual host name to use on the broker.
- * @param destinationName The name (or root where multiple destinations are used) of the desitination to send
- * pings to.
- * @param selector The selector to filter replies with.
- * @param transacted Indicates whether or not pings are sent and received in transactions.
- * @param persistent Indicates whether pings are sent using peristent delivery.
- * @param messageSize Specifies the size of ping messages to send.
- * @param verbose Indicates that information should be printed to the console on every ping.
- * @param afterCommit Indicates that the user should be promted to terminate a broker after commits to test failover.
- * @param beforeCommit Indicates that the user should be promted to terminate a broker before commits to test failover.
- * @param afterSend Indicates that the user should be promted to terminate a broker after sends to test failover.
- * @param beforeSend Indicates that the user should be promted to terminate a broker before sends to test failover.
- * @param failOnce Indicates that the failover testing behaviour should only happen on the first commit, not all.
- * @param txBatchSize Specifies the number of pings to send in each transaction.
- * @param noOfDestinations The number of destinations to ping. Must be 1 or more.
- * @param rate Specified the number of pings per second to send. Setting this to 0 means send as fast as
- * possible, with no rate restriction.
- * @param pubsub True to ping topics, false to ping queues.
- * @param unique True to use unique destinations for each ping pong producer, false to share.
+ * Creates a ping producer with the specified parameters, of which there are many. See their individual comments for
+ * details. This constructor creates a connection to the broker and creates producer and consumer sessions on it, to send
+ * and recieve its pings and replies on. The other options are kept, and control how this pinger behaves.
+ *
+ * @param brokerDetails The URL of the broker to send pings to.
+ * @param username The username to log onto the broker with.
+ * @param password The password to log onto the broker with.
+ * @param virtualpath The virtual host name to use on the broker.
+ * @param destinationName The name (or root where multiple destinations are used) of the desitination to send pings to.
+ * @param selector The selector to filter replies with.
+ * @param transacted Indicates whether or not pings are sent and received in transactions.
+ * @param persistent Indicates whether pings are sent using peristent delivery.
+ * @param messageSize Specifies the size of ping messages to send.
+ * @param verbose Indicates that information should be printed to the console on every ping.
+ * @param afterCommit Indicates that the user should be promted to terminate a broker after commits to test
+ * failover.
+ * @param beforeCommit Indicates that the user should be promted to terminate a broker before commits to test
+ * failover.
+ * @param afterSend Indicates that the user should be promted to terminate a broker after sends to test failover.
+ * @param beforeSend Indicates that the user should be promted to terminate a broker before sends to test
+ * failover.
+ * @param failOnce Indicates that the failover testing behaviour should only happen on the first commit, not
+ * all.
+ * @param txBatchSize Specifies the number of pings to send in each transaction.
+ * @param noOfDestinations The number of destinations to ping. Must be 1 or more.
+ * @param rate Specified the number of pings per second to send. Setting this to 0 means send as fast as
+ * possible, with no rate restriction.
+ * @param pubsub True to ping topics, false to ping queues.
+ * @param unique True to use unique destinations for each ping pong producer, false to share.
+ *
* @throws Exception Any exceptions are allowed to fall through.
*/
public PingPongProducer(String brokerDetails, String username, String password, String virtualpath,
@@ -480,8 +416,8 @@
+ afterCommit + ", boolean beforeCommit = " + beforeCommit + ", boolean afterSend = " + afterSend
+ ", boolean beforeSend = " + beforeSend + ", boolean failOnce = " + failOnce + ", int txBatchSize = "
+ txBatchSize + ", int noOfDestinations = " + noOfDestinations + ", int rate = " + rate
- + ", boolean pubsub = " + pubsub + ", boolean unique = " + unique
- + ", ackMode = " + ackMode + "): called");
+ + ", boolean pubsub = " + pubsub + ", boolean unique = " + unique + ", ackMode = " + ackMode
+ + "): called");
// Keep all the relevant options.
_persistent = persistent;
@@ -495,12 +431,13 @@
_txBatchSize = txBatchSize;
_isPubSub = pubsub;
_isUnique = unique;
- _pausetimeAfterEachBatch = pause;
+ _pauseBatch = pause;
+
if (ackMode != 0)
{
_ackMode = ackMode;
}
-
+
// Check that one or more destinations were specified.
if (noOfDestinations < 1)
{
@@ -534,10 +471,11 @@
}
/**
- * Starts a ping-pong loop running from the command line. The bounce back client {@link PingPongBouncer} also needs
- * to be started to bounce the pings back again.
+ * Starts a ping-pong loop running from the command line. The bounce back client {@link PingPongBouncer} also needs to be
+ * started to bounce the pings back again.
+ *
+ * @param args The command line arguments.
*
- * @param args The command line arguments.
* @throws Exception When something went wrong with the test
*/
public static void main(String[] args) throws Exception
@@ -548,27 +486,27 @@
if (args.length == 0)
{
_logger.info("Running test with default values...");
- //usage();
- //System.exit(0);
+ // usage();
+ // System.exit(0);
}
String brokerDetails = config.getHost() + ":" + config.getPort();
- String virtualpath = DEFAULT_VIRTUAL_PATH;
- String selector = (config.getSelector() == null) ? DEFAULT_SELECTOR : config.getSelector();
+ String virtualpath = VIRTUAL_HOST_DEFAULT;
+ String selector = (config.getSelector() == null) ? SELECTOR_DEFAULT : config.getSelector();
boolean verbose = true;
boolean transacted = config.isTransacted();
boolean persistent = config.usePersistentMessages();
- int messageSize = (config.getPayload() != 0) ? config.getPayload() : DEFAULT_MESSAGE_SIZE;
- //int messageCount = config.getMessages();
- int destCount = (config.getDestinationsCount() != 0) ? config.getDestinationsCount() : DEFAULT_DESTINATION_COUNT;
- int batchSize = (config.getBatchSize() != 0) ? config.getBatchSize() : DEFAULT_TX_BATCH_SIZE;
- int rate = (config.getRate() != 0) ? config.getRate() : DEFAULT_RATE;
+ int messageSize = (config.getPayload() != 0) ? config.getPayload() : MESSAGE_SIZE_DEAFULT;
+ // int messageCount = config.getMessages();
+ int destCount = (config.getDestinationsCount() != 0) ? config.getDestinationsCount() : DESTINATION_COUNT_DEFAULT;
+ int batchSize = (config.getBatchSize() != 0) ? config.getBatchSize() : TX_BATCH_SIZE_DEFAULT;
+ int rate = (config.getRate() != 0) ? config.getRate() : RATE_DEFAULT;
boolean pubsub = config.isPubSub();
String destName = config.getDestination();
if (destName == null)
{
- destName = DEFAULT_PING_DESTINATION_NAME;
+ destName = PING_QUEUE_NAME_DEFAULT;
}
boolean afterCommit = false;
@@ -581,7 +519,7 @@
{
if (arg.startsWith("failover:"))
{
- //failover:<before|after>:<send:commit> | failover:once
+ // failover:<before|after>:<send:commit> | failover:once
String[] parts = arg.split(":");
if (parts.length == 3)
{
@@ -611,9 +549,9 @@
// Create a ping producer to handle the request/wait/reply cycle.
PingPongProducer pingProducer =
- new PingPongProducer(brokerDetails, DEFAULT_USERNAME, DEFAULT_PASSWORD, virtualpath, destName, selector,
- transacted, persistent, messageSize, verbose, afterCommit, beforeCommit, afterSend,
- beforeSend, failOnce, batchSize, destCount, rate, pubsub, false, 0, 0);
+ new PingPongProducer(brokerDetails, USERNAME_DEFAULT, PASSWORD_DEFAULT, virtualpath, destName, selector,
+ transacted, persistent, messageSize, verbose, afterCommit, beforeCommit, afterSend,
+ beforeSend, failOnce, batchSize, destCount, rate, pubsub, false, 0, 0);
pingProducer.getConnection().start();
@@ -644,14 +582,14 @@
}
catch (InterruptedException ie)
{
- //ignore
+ // ignore
}
}
}
/**
- * Gets all the reply destinations (to listen for replies on). In this case this will just be the single reply
- * to destination of this pinger.
+ * Gets all the reply destinations (to listen for replies on). In this case this will just be the single reply to
+ * destination of this pinger.
*
* @return The single reply to destination of this pinger, wrapped in a list.
*/
@@ -666,8 +604,8 @@
}
/**
- * Creates the producer to send the pings on. This is created without a default destination. Its persistent delivery
- * flag is set accoring the ping producer creation options.
+ * Creates the producer to send the pings on. This is created without a default destination. Its persistent delivery flag
+ * is set accoring the ping producer creation options.
*
* @throws JMSException Any JMSExceptions are allowed to fall through.
*/
@@ -676,23 +614,24 @@
_logger.debug("public void createProducer(): called");
_producer = (MessageProducer) _producerSession.createProducer(null);
- //_producer.setDisableMessageTimestamp(true);
+ // _producer.setDisableMessageTimestamp(true);
_producer.setDeliveryMode(_persistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);
}
/**
- * Creates consumers for the specified number of destinations. The destinations themselves are also created by
- * this method.
+ * Creates consumers for the specified number of destinations. The destinations themselves are also created by this
+ * method.
+ *
+ * @param noOfDestinations The number of destinations to create consumers for.
+ * @param selector The message selector to filter the consumers with.
+ * @param rootName The root of the name, or actual name if only one is being created.
+ * @param unique <tt>true</tt> to make the destinations unique to this pinger, <tt>false</tt> to share the
+ * numbering with all pingers on the same JVM.
*
- * @param noOfDestinations The number of destinations to create consumers for.
- * @param selector The message selector to filter the consumers with.
- * @param rootName The root of the name, or actual name if only one is being created.
- * @param unique <tt>true</tt> to make the destinations unique to this pinger, <tt>false</tt> to share
- * the numbering with all pingers on the same JVM.
* @throws JMSException Any JMSExceptions are allowed to fall through.
*/
public void createPingDestinations(int noOfDestinations, String selector, String rootName, boolean unique)
- throws JMSException
+ throws JMSException
{
_logger.debug("public void createPingDestinations(int noOfDestinations = " + noOfDestinations
+ ", String selector = " + selector + ", String rootName = " + rootName + ", boolean unique = "
@@ -736,9 +675,8 @@
}
/**
- * Stores the received message in the replies map, then resets the boolean latch that a thread waiting for a
- * correlating reply may be waiting on. This is only done if the reply has a correlation id that is expected
- * in the replies map.
+ * Stores the received message in the replies map, then resets the boolean latch that a thread waiting for a correlating
+ * reply may be waiting on. This is only done if the reply has a correlation id that is expected in the replies map.
*
* @param message The received message.
*/
@@ -827,15 +765,17 @@
}
/**
- * Sends the specified number of ping message and then waits for all correlating replies. If the wait times out
- * before a reply arrives, then a null reply is returned from this method. This method generates a new unqiue
- * correlation id for the messages.
- *
- * @param message The message to send.
- * @param numPings The number of ping messages to send.
- * @param timeout The timeout in milliseconds.
- * @return The number of replies received. This may be less than the number sent if the timeout terminated the
- * wait for all prematurely.
+ * Sends the specified number of ping message and then waits for all correlating replies. If the wait times out before a
+ * reply arrives, then a null reply is returned from this method. This method generates a new unqiue correlation id for
+ * the messages.
+ *
+ * @param message The message to send.
+ * @param numPings The number of ping messages to send.
+ * @param timeout The timeout in milliseconds.
+ *
+ * @return The number of replies received. This may be less than the number sent if the timeout terminated the wait for
+ * all prematurely.
+ *
* @throws JMSException All underlying JMSExceptions are allowed to fall through.
* @throws InterruptedException When interrupted by a timeout.
*/
@@ -851,27 +791,29 @@
}
public int pingAndWaitForReply(int numPings, long timeout, String messageCorrelationId)
- throws JMSException, InterruptedException
+ throws JMSException, InterruptedException
{
- return pingAndWaitForReply(null, numPings, timeout, messageCorrelationId);
+ return pingAndWaitForReply(null, numPings, timeout, messageCorrelationId);
}
/**
- * Sends the specified number of ping message and then waits for all correlating replies. If the wait times out
- * before a reply arrives, then a null reply is returned from this method. This method allows the caller to specify
- * the correlation id.
+ * Sends the specified number of ping message and then waits for all correlating replies. If the wait times out before a
+ * reply arrives, then a null reply is returned from this method. This method allows the caller to specify the
+ * correlation id.
+ *
+ * @param message The message to send.
+ * @param numPings The number of ping messages to send.
+ * @param timeout The timeout in milliseconds.
+ * @param messageCorrelationId The message correlation id.
+ *
+ * @return The number of replies received. This may be less than the number sent if the timeout terminated the wait for
+ * all prematurely.
*
- * @param message The message to send.
- * @param numPings The number of ping messages to send.
- * @param timeout The timeout in milliseconds.
- * @param messageCorrelationId The message correlation id.
- * @return The number of replies received. This may be less than the number sent if the timeout terminated the
- * wait for all prematurely.
* @throws JMSException All underlying JMSExceptions are allowed to fall through.
* @throws InterruptedException When interrupted by a timeout
*/
public int pingAndWaitForReply(Message message, int numPings, long timeout, String messageCorrelationId)
- throws JMSException, InterruptedException
+ throws JMSException, InterruptedException
{
_logger.debug("public int pingAndWaitForReply(Message message, int numPings = " + numPings + ", long timeout = "
+ timeout + ", String messageCorrelationId = " + messageCorrelationId + "): called");
@@ -948,9 +890,10 @@
/**
* Sends the specified number of ping messages and does not wait for correlating replies.
*
- * @param message The message to send.
- * @param numPings The number of pings to send.
- * @param messageCorrelationId A correlation id to place on all messages sent.
+ * @param message The message to send.
+ * @param numPings The number of pings to send.
+ * @param messageCorrelationId A correlation id to place on all messages sent.
+ *
* @throws JMSException All underlying JMSExceptions are allowed to fall through.
*/
public void pingNoWaitForReply(Message message, int numPings, String messageCorrelationId) throws JMSException
@@ -980,7 +923,7 @@
message.setLongProperty(MESSAGE_TIMESTAMP_PROPNAME, System.nanoTime());
// Round robin the destinations as the messages are sent.
- //return _destinationCount;
+ // return _destinationCount;
sendMessage(_pingDestinations.get(i % _pingDestinations.size()), message);
// Apply message rate throttling if a rate limit has been set up.
@@ -994,11 +937,12 @@
{
commitTx(_producerSession);
committed = true;
+
/* This pause is required for some cases. eg in load testing when sessions are non-transacted the
Mina IO layer can't clear the cache in time. So this pause gives enough time for mina to clear
the cache (without this mina throws OutOfMemoryError). pause() will check if time is != 0
- */
- pause(_pausetimeAfterEachBatch);
+ */
+ pause(_pauseBatch);
}
// Spew out per message timings on every message sonly in verbose mode.
@@ -1028,10 +972,7 @@
msg.setLongProperty(MESSAGE_TIMESTAMP_PROPNAME, System.nanoTime());
// Send the message and wait for a reply.
- pingAndWaitForReply(msg, DEFAULT_TX_BATCH_SIZE, DEFAULT_TIMEOUT);
-
- // Introduce a short pause if desired.
- pause(DEFAULT_SLEEP_TIME);
+ pingAndWaitForReply(msg, TX_BATCH_SIZE_DEFAULT, TIMEOUT_DEFAULT);
}
catch (JMSException e)
{
@@ -1051,8 +992,7 @@
}
/**
- * Sets a chained message listener. The message listener on this pinger, chains all its messages to the one set
- * here.
+ * Sets a chained message listener. The message listener on this pinger, chains all its messages to the one set here.
*
* @param messageListener The chained message listener.
*/
@@ -1072,10 +1012,12 @@
/**
* Generates a test message of the specified size, with the specified reply-to destination and persistence flag.
*
- * @param replyQueue The reply-to destination for the message.
- * @param messageSize The desired size of the message in bytes.
- * @param persistent <tt>true</tt> if the message should use persistent delivery, <tt>false</tt> otherwise.
+ * @param replyQueue The reply-to destination for the message.
+ * @param messageSize The desired size of the message in bytes.
+ * @param persistent <tt>true</tt> if the message should use persistent delivery, <tt>false</tt> otherwise.
+ *
* @return A freshly generated test message.
+ *
* @throws javax.jms.JMSException All underlying JMSException are allowed to fall through.
*/
public ObjectMessage getTestMessage(Destination replyQueue, int messageSize, boolean persistent) throws JMSException
@@ -1089,8 +1031,8 @@
}
/**
- * Stops the ping loop by clearing the publish flag. The current loop will complete before it notices that this
- * flag has been cleared.
+ * Stops the ping loop by clearing the publish flag. The current loop will complete before it notices that this flag has
+ * been cleared.
*/
public void stop()
{
@@ -1110,8 +1052,8 @@
}
/**
- * Callback method, implementing ExceptionListener. This should be registered to listen for exceptions on the
- * connection, this clears the publish flag which in turn will halt the ping loop.
+ * Callback method, implementing ExceptionListener. This should be registered to listen for exceptions on the connection,
+ * this clears the publish flag which in turn will halt the ping loop.
*
* @param e The exception that triggered this callback method.
*/
@@ -1122,20 +1064,20 @@
}
/**
- * Gets a shutdown hook that will cleanly shut this down when it is running the ping loop. This can be registered
- * with the runtime system as a shutdown hook.
+ * Gets a shutdown hook that will cleanly shut this down when it is running the ping loop. This can be registered with
+ * the runtime system as a shutdown hook.
*
* @return A shutdown hook for the ping loop.
*/
public Thread getShutdownHook()
{
return new Thread(new Runnable()
- {
- public void run()
{
- stop();
- }
- });
+ public void run()
+ {
+ stop();
+ }
+ });
}
/**
@@ -1151,8 +1093,9 @@
/**
* Creates consumers for the specified destinations and registers this pinger to listen to their messages.
*
- * @param destinations The destinations to listen to.
- * @param selector A selector to filter the messages with.
+ * @param destinations The destinations to listen to.
+ * @param selector A selector to filter the messages with.
+ *
* @throws javax.jms.JMSException Any JMSExceptions are allowed to fall through.
*/
public void createReplyConsumers(Collection<Destination> destinations, String selector) throws JMSException
@@ -1164,8 +1107,8 @@
{
// Create a consumer for the destination and set this pinger to listen to its messages.
MessageConsumer consumer =
- _consumerSession.createConsumer(destination, DEFAULT_PREFETCH, DEFAULT_NO_LOCAL, DEFAULT_EXCLUSIVE,
- selector);
+ _consumerSession.createConsumer(destination, PREFETCH_DEFAULT, NO_LOCAL_DEFAULT, EXCLUSIVE_DEFAULT,
+ selector);
consumer.setMessageListener(this);
}
}
@@ -1186,22 +1129,23 @@
}
/**
- * Convenience method to commit the transaction on the specified session. If the session to commit on is not
- * a transactional session, this method does nothing (unless the failover after send flag is set).
- * <p/>
- * <p/>If the {@link #_failAfterSend} flag is set, this will prompt the user to kill the broker before the commit
- * is applied. This flag applies whether the pinger is transactional or not.
- * <p/>
- * <p/>If the {@link #_failBeforeCommit} flag is set, this will prompt the user to kill the broker before the
- * commit is applied. If the {@link #_failAfterCommit} flag is set, this will prompt the user to kill the broker
- * after the commit is applied. These flags will only apply if using a transactional pinger.
+ * Convenience method to commit the transaction on the specified session. If the session to commit on is not a
+ * transactional session, this method does nothing (unless the failover after send flag is set).
+ *
+ * <p/>If the {@link #_failAfterSend} flag is set, this will prompt the user to kill the broker before the commit is
+ * applied. This flag applies whether the pinger is transactional or not.
+ *
+ * <p/>If the {@link #_failBeforeCommit} flag is set, this will prompt the user to kill the broker before the commit is
+ * applied. If the {@link #_failAfterCommit} flag is set, this will prompt the user to kill the broker after the commit
+ * is applied. These flags will only apply if using a transactional pinger.
+ *
+ * @param session The session to commit
*
- * @param session The session to commit
* @throws javax.jms.JMSException If the commit fails and then the rollback fails.
- * <p/>
- * //todo @todo Consider moving the fail after send logic into the send method. It is confusing to have it in this commit
- * method, because commits only apply to transactional pingers, but fail after send applied to transactional
- * and non-transactional alike.
+ *
+ * @todo Consider moving the fail after send logic into the send method. It is confusing to have it in this commit
+ * method, because commits only apply to transactional pingers, but fail after send applied to transactional and
+ * non-transactional alike.
*/
protected void commitTx(Session session) throws JMSException
{
@@ -1237,7 +1181,7 @@
long l = System.currentTimeMillis();
session.commit();
- _logger.debug("Time taken to commit :" + (System.currentTimeMillis() - l) + " ms" );
+ _logger.debug("Time taken to commit :" + (System.currentTimeMillis() - l) + " ms");
if (_failAfterCommit)
{
@@ -1279,11 +1223,12 @@
}
/**
- * Sends the message to the specified destination. If the destination is null, it gets sent to the default destination
- * of the ping producer. If an explicit destination is set, this overrides the default.
+ * Sends the message to the specified destination. If the destination is null, it gets sent to the default destination of
+ * the ping producer. If an explicit destination is set, this overrides the default.
+ *
+ * @param destination The destination to send to.
+ * @param message The message to send.
*
- * @param destination The destination to send to.
- * @param message The message to send.
* @throws javax.jms.JMSException All underlying JMSExceptions are allowed to fall through.
*/
protected void sendMessage(Destination destination, Message message) throws JMSException
@@ -1310,8 +1255,8 @@
}
/**
- * Prompts the user to terminate the broker, in order to test failover functionality. This method will block
- * until the user supplied some input on the terminal.
+ * Prompts the user to terminate the broker, in order to test failover functionality. This method will block until the
+ * user supplied some input on the terminal.
*/
protected void doFailover()
{
@@ -1322,7 +1267,7 @@
}
catch (IOException e)
{
- //ignore
+ // ignore
}
System.out.println("Continuing.");
@@ -1343,16 +1288,15 @@
return numpings * getConsumersPerTopic();
}
-
/**
- * Defines a chained message listener interface that can be attached to this pinger. Whenever this pinger's
- * {@link PingPongProducer#onMessage} method is called, the chained listener set through the
- * {@link PingPongProducer#setChainedMessageListener} method is passed the message, and the remaining expected
- * count of messages with that correlation id.
- * <p/>
- * Provided only one pinger is producing messages with that correlation id, the chained listener will always be
- * given unique message counts. It will always be called while the producer waiting for all messages to arrive is
- * still blocked.
+ * Defines a chained message listener interface that can be attached to this pinger. Whenever this pinger's {@link
+ * PingPongProducer#onMessage} method is called, the chained listener set through the {@link
+ * PingPongProducer#setChainedMessageListener} method is passed the message, and the remaining expected count of messages
+ * with that correlation id.
+ *
+ * <p/>Provided only one pinger is producing messages with that correlation id, the chained listener will always be given
+ * unique message counts. It will always be called while the producer waiting for all messages to arrive is still
+ * blocked.
*/
public static interface ChainedMessageListener
{
@@ -1360,19 +1304,15 @@
}
/**
- * Holds information on each correlation id. The countdown latch, the current timeout timer... More stuff to be
- * added to this: read/write lock to make onMessage more concurrent as described in class header comment.
+ * Holds information on each correlation id. The countdown latch, the current timeout timer... More stuff to be added to
+ * this: read/write lock to make onMessage more concurrent as described in class header comment.
*/
protected static class PerCorrelationId
{
- /**
- * Holds a countdown on number of expected messages.
- */
+ /** Holds a countdown on number of expected messages. */
CountDownLatch trafficLight;
- /**
- * Holds the last timestamp that the timeout was reset to.
- */
+ /** Holds the last timestamp that the timeout was reset to. */
Long timeOutStart;
}
}
Modified: incubator/qpid/branches/M2/java/perftests/src/test/java/org/apache/qpid/ping/PingAsyncTestPerf.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/M2/java/perftests/src/test/java/org/apache/qpid/ping/PingAsyncTestPerf.java?view=diff&rev=521705&r1=521704&r2=521705
==============================================================================
--- incubator/qpid/branches/M2/java/perftests/src/test/java/org/apache/qpid/ping/PingAsyncTestPerf.java (original)
+++ incubator/qpid/branches/M2/java/perftests/src/test/java/org/apache/qpid/ping/PingAsyncTestPerf.java Fri Mar 23 05:44:14 2007
@@ -57,10 +57,10 @@
private static Logger _logger = Logger.getLogger(PingAsyncTestPerf.class);
/** Holds the name of the property to get the test results logging batch size. */
- public static final String TEST_RESULTS_BATCH_SIZE_PROPNAME = "BatchSize";
+ public static final String TEST_RESULTS_BATCH_SIZE_PROPNAME = "batchSize";
/** Holds the default test results logging batch size. */
- public static final int DEFAULT_TEST_RESULTS_BATCH_SIZE = 1000;
+ public static final int TEST_RESULTS_BATCH_SIZE_DEFAULT = 1000;
/** Used to hold the timing controller passed from the test runner. */
private TimingController _timingController;
@@ -70,7 +70,7 @@
/** Holds test specifics by correlation id. This consists of the expected number of messages and the timing controler. */
private Map<String, PerCorrelationId> perCorrelationIds =
- Collections.synchronizedMap(new HashMap<String, PerCorrelationId>());
+ Collections.synchronizedMap(new HashMap<String, PerCorrelationId>());
/** Holds the batched results listener, that does logging on batch boundaries. */
private BatchedResultsListener batchedResultsListener = null;
@@ -86,12 +86,12 @@
// Sets up the test parameters with defaults.
testParameters.setPropertyIfNull(TEST_RESULTS_BATCH_SIZE_PROPNAME,
- Integer.toString(DEFAULT_TEST_RESULTS_BATCH_SIZE));
+ Integer.toString(TEST_RESULTS_BATCH_SIZE_DEFAULT));
}
/**
* Compile all the tests into a test suite.
- * @return The test suite to run. Should only contain testAsyncPingOk method.
+ * @return The test suite to run. Should only contain testAsyncPingOk method.
*/
public static Test suite()
{
@@ -129,7 +129,7 @@
* all replies have been received or a time out occurs before exiting this method.
*
* @param numPings The number of pings to send.
- * @throws Exception pass all errors out to the test harness
+ * @throws Exception pass all errors out to the test harness
*/
public void testAsyncPingOk(int numPings) throws Exception
{
@@ -246,9 +246,9 @@
// Extract the correlation id from the message.
String correlationId = message.getJMSCorrelationID();
- _logger.debug("public void onMessage(Message message, int remainingCount = " + remainingCount +
- "): called on batch boundary for message id: " + correlationId +
- " with thread id: " + Thread.currentThread().getId());
+ _logger.debug("public void onMessage(Message message, int remainingCount = " + remainingCount
+ + "): called on batch boundary for message id: " + correlationId + " with thread id: "
+ + Thread.currentThread().getId());
// Get the details for the correlation id and check that they are not null. They can become null
// if a test times out.
@@ -268,7 +268,6 @@
// Register a test result for the correlation id.
try
{
-
tc.completeTest(true, receivedInBatch);
}
catch (InterruptedException e)
Modified: incubator/qpid/branches/M2/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/M2/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java?view=diff&rev=521705&r1=521704&r2=521705
==============================================================================
--- incubator/qpid/branches/M2/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java (original)
+++ incubator/qpid/branches/M2/java/perftests/src/test/java/org/apache/qpid/ping/PingTestPerf.java Fri Mar 23 05:44:14 2007
@@ -66,50 +66,44 @@
ThreadLocal<PerThreadSetup> threadSetup = new ThreadLocal<PerThreadSetup>();
/** Holds a property reader to extract the test parameters from. */
- protected ParsedProperties testParameters = new ParsedProperties(System.getProperties());
+ protected ParsedProperties testParameters = new TestContextProperties(System.getProperties());
public PingTestPerf(String name)
{
super(name);
// Sets up the test parameters with defaults.
- testParameters.setPropertyIfNull(PingPongProducer.COMMIT_BATCH_SIZE_PROPNAME,
- Integer.toString(PingPongProducer.DEFAULT_TX_BATCH_SIZE));
- testParameters.setPropertyIfNull(PingPongProducer.MESSAGE_SIZE_PROPNAME,
- Integer.toString(PingPongProducer.DEFAULT_MESSAGE_SIZE));
+ testParameters.setPropertyIfNull(PingPongProducer.TX_BATCH_SIZE_PROPNAME, PingPongProducer.TX_BATCH_SIZE_DEFAULT);
+ testParameters.setPropertyIfNull(PingPongProducer.MESSAGE_SIZE_PROPNAME, PingPongProducer.MESSAGE_SIZE_DEAFULT);
testParameters.setPropertyIfNull(PingPongProducer.PING_QUEUE_NAME_PROPNAME,
- PingPongProducer.DEFAULT_PING_DESTINATION_NAME);
+ PingPongProducer.PING_QUEUE_NAME_DEFAULT);
testParameters.setPropertyIfNull(PingPongProducer.PERSISTENT_MODE_PROPNAME,
- Boolean.toString(PingPongProducer.DEFAULT_PERSISTENT_MODE));
- testParameters.setPropertyIfNull(PingPongProducer.TRANSACTED_PROPNAME,
- Boolean.toString(PingPongProducer.DEFAULT_TRANSACTED));
- testParameters.setPropertyIfNull(PingPongProducer.BROKER_PROPNAME, PingPongProducer.DEFAULT_BROKER);
- testParameters.setPropertyIfNull(PingPongProducer.USERNAME_PROPNAME, PingPongProducer.DEFAULT_USERNAME);
- testParameters.setPropertyIfNull(PingPongProducer.PASSWORD_PROPNAME, PingPongProducer.DEFAULT_PASSWORD);
- testParameters.setPropertyIfNull(PingPongProducer.VIRTUAL_PATH_PROPNAME, PingPongProducer.DEFAULT_VIRTUAL_PATH);
- testParameters.setPropertyIfNull(PingPongProducer.VERBOSE_OUTPUT_PROPNAME,
- Boolean.toString(PingPongProducer.DEFAULT_VERBOSE));
- testParameters.setPropertyIfNull(PingPongProducer.RATE_PROPNAME, Integer.toString(PingPongProducer.DEFAULT_RATE));
- testParameters.setPropertyIfNull(PingPongProducer.IS_PUBSUB_PROPNAME,
- Boolean.toString(PingPongProducer.DEFAULT_PUBSUB));
- testParameters.setPropertyIfNull(PingPongProducer.COMMIT_BATCH_SIZE_PROPNAME,
- Integer.toString(PingPongProducer.DEFAULT_TX_BATCH_SIZE));
- testParameters.setPropertyIfNull(PingPongProducer.TIMEOUT_PROPNAME, Long.toString(PingPongProducer.DEFAULT_TIMEOUT));
- testParameters.setPropertyIfNull(PingPongProducer.PING_DESTINATION_COUNT_PROPNAME,
- Integer.toString(PingPongProducer.DEFAULT_DESTINATION_COUNT));
+ PingPongProducer.PERSISTENT_MODE_DEFAULT);
+ testParameters.setPropertyIfNull(PingPongProducer.TRANSACTED_PROPNAME, PingPongProducer.TRANSACTED_DEFAULT);
+ testParameters.setPropertyIfNull(PingPongProducer.BROKER_PROPNAME, PingPongProducer.BROKER_DEFAULT);
+ testParameters.setPropertyIfNull(PingPongProducer.USERNAME_PROPNAME, PingPongProducer.USERNAME_DEFAULT);
+ testParameters.setPropertyIfNull(PingPongProducer.PASSWORD_PROPNAME, PingPongProducer.PASSWORD_DEFAULT);
+ testParameters.setPropertyIfNull(PingPongProducer.VIRTUAL_HOST_PROPNAME, PingPongProducer.VIRTUAL_HOST_DEFAULT);
+ testParameters.setPropertyIfNull(PingPongProducer.VERBOSE_PROPNAME, PingPongProducer.VERBOSE_DEFAULT);
+ testParameters.setPropertyIfNull(PingPongProducer.RATE_PROPNAME, PingPongProducer.RATE_DEFAULT);
+ testParameters.setPropertyIfNull(PingPongProducer.PUBSUB_PROPNAME, PingPongProducer.PUBSUB_DEFAULT);
+ testParameters.setPropertyIfNull(PingPongProducer.TX_BATCH_SIZE_PROPNAME, PingPongProducer.TX_BATCH_SIZE_DEFAULT);
+ testParameters.setPropertyIfNull(PingPongProducer.TIMEOUT_PROPNAME, PingPongProducer.TIMEOUT_DEFAULT);
+ testParameters.setPropertyIfNull(PingPongProducer.DESTINATION_COUNT_PROPNAME,
+ PingPongProducer.DESTINATION_COUNT_DEFAULT);
testParameters.setPropertyIfNull(PingPongProducer.FAIL_AFTER_COMMIT_PROPNAME,
- PingPongProducer.DEFAULT_FAIL_AFTER_COMMIT);
+ PingPongProducer.FAIL_AFTER_COMMIT_DEFAULT);
testParameters.setPropertyIfNull(PingPongProducer.FAIL_BEFORE_COMMIT_PROPNAME,
- PingPongProducer.DEFAULT_FAIL_BEFORE_COMMIT);
+ PingPongProducer.FAIL_BEFORE_COMMIT_DEFAULT);
testParameters.setPropertyIfNull(PingPongProducer.FAIL_AFTER_SEND_PROPNAME,
- PingPongProducer.DEFAULT_FAIL_AFTER_SEND);
+ PingPongProducer.FAIL_AFTER_SEND_DEFAULT);
testParameters.setPropertyIfNull(PingPongProducer.FAIL_BEFORE_SEND_PROPNAME,
- PingPongProducer.DEFAULT_FAIL_BEFORE_SEND);
- testParameters.setPropertyIfNull(PingPongProducer.FAIL_ONCE_PROPNAME, PingPongProducer.DEFAULT_FAIL_ONCE);
- testParameters.setPropertyIfNull(PingPongProducer.UNIQUE_PROPNAME, PingPongProducer.DEFAULT_UNIQUE);
- testParameters.setSysPropertyIfNull(PingPongProducer.ACK_MODE_PROPNAME,
- Integer.toString(PingPongProducer.DEFAULT_ACK_MODE));
- testParameters.setSysPropertyIfNull(PingPongProducer.PAUSE_AFTER_BATCH_PROPNAME, 0l);
+ PingPongProducer.FAIL_BEFORE_SEND_DEFAULT);
+ testParameters.setPropertyIfNull(PingPongProducer.FAIL_ONCE_PROPNAME, PingPongProducer.FAIL_ONCE_DEFAULT);
+ testParameters.setPropertyIfNull(PingPongProducer.UNIQUE_DESTS_PROPNAME, PingPongProducer.UNIQUE_DESTS_DEFAULT);
+ testParameters.setPropertyIfNull(PingPongProducer.ACK_MODE_PROPNAME, PingPongProducer.ACK_MODE_DEFAULT);
+ testParameters.setPropertyIfNull(PingPongProducer.PAUSE_AFTER_BATCH_PROPNAME,
+ PingPongProducer.PAUSE_AFTER_BATCH_DEFAULT);
}
/**
@@ -144,11 +138,11 @@
// Generate a sample message. This message is already time stamped and has its reply-to destination set.
ObjectMessage msg =
- perThreadSetup._pingClient.getTestMessage(perThreadSetup._pingClient.getReplyDestinations().get(0),
- testParameters.getPropertyAsInteger(
- PingPongProducer.MESSAGE_SIZE_PROPNAME),
- testParameters.getPropertyAsBoolean(
- PingPongProducer.PERSISTENT_MODE_PROPNAME));
+ perThreadSetup._pingClient.getTestMessage(perThreadSetup._pingClient.getReplyDestinations().get(0),
+ testParameters.getPropertyAsInteger(
+ PingPongProducer.MESSAGE_SIZE_PROPNAME),
+ testParameters.getPropertyAsBoolean(
+ PingPongProducer.PERSISTENT_MODE_PROPNAME));
// start the test
long timeout = Long.parseLong(testParameters.getProperty(PingPongProducer.TIMEOUT_PROPNAME));
@@ -177,28 +171,28 @@
String brokerDetails = testParameters.getProperty(PingPongProducer.BROKER_PROPNAME);
String username = testParameters.getProperty(PingPongProducer.USERNAME_PROPNAME);
String password = testParameters.getProperty(PingPongProducer.PASSWORD_PROPNAME);
- String virtualPath = testParameters.getProperty(PingPongProducer.VIRTUAL_PATH_PROPNAME);
+ String virtualPath = testParameters.getProperty(PingPongProducer.VIRTUAL_HOST_PROPNAME);
String destinationName = testParameters.getProperty(PingPongProducer.PING_QUEUE_NAME_PROPNAME);
boolean persistent = testParameters.getPropertyAsBoolean(PingPongProducer.PERSISTENT_MODE_PROPNAME);
boolean transacted = testParameters.getPropertyAsBoolean(PingPongProducer.TRANSACTED_PROPNAME);
String selector = testParameters.getProperty(PingPongProducer.SELECTOR_PROPNAME);
- boolean verbose = testParameters.getPropertyAsBoolean(PingPongProducer.VERBOSE_OUTPUT_PROPNAME);
+ boolean verbose = testParameters.getPropertyAsBoolean(PingPongProducer.VERBOSE_PROPNAME);
int messageSize = testParameters.getPropertyAsInteger(PingPongProducer.MESSAGE_SIZE_PROPNAME);
int rate = testParameters.getPropertyAsInteger(PingPongProducer.RATE_PROPNAME);
- boolean pubsub = testParameters.getPropertyAsBoolean(PingPongProducer.IS_PUBSUB_PROPNAME);
+ boolean pubsub = testParameters.getPropertyAsBoolean(PingPongProducer.PUBSUB_PROPNAME);
boolean failAfterCommit = testParameters.getPropertyAsBoolean(PingPongProducer.FAIL_AFTER_COMMIT_PROPNAME);
boolean failBeforeCommit = testParameters.getPropertyAsBoolean(PingPongProducer.FAIL_BEFORE_COMMIT_PROPNAME);
boolean failAfterSend = testParameters.getPropertyAsBoolean(PingPongProducer.FAIL_AFTER_SEND_PROPNAME);
boolean failBeforeSend = testParameters.getPropertyAsBoolean(PingPongProducer.FAIL_BEFORE_SEND_PROPNAME);
- int batchSize = testParameters.getPropertyAsInteger(PingPongProducer.COMMIT_BATCH_SIZE_PROPNAME);
+ int batchSize = testParameters.getPropertyAsInteger(PingPongProducer.TX_BATCH_SIZE_PROPNAME);
Boolean failOnce = testParameters.getPropertyAsBoolean(PingPongProducer.FAIL_ONCE_PROPNAME);
- boolean unique = testParameters.getPropertyAsBoolean(PingPongProducer.UNIQUE_PROPNAME);
+ boolean unique = testParameters.getPropertyAsBoolean(PingPongProducer.UNIQUE_DESTS_PROPNAME);
int ackMode = testParameters.getPropertyAsInteger(PingPongProducer.ACK_MODE_PROPNAME);
int pausetime = testParameters.getPropertyAsInteger(PingPongProducer.PAUSE_AFTER_BATCH_PROPNAME);
// Extract the test set up paramaeters.
int destinationscount =
- Integer.parseInt(testParameters.getProperty(PingPongProducer.PING_DESTINATION_COUNT_PROPNAME));
+ Integer.parseInt(testParameters.getProperty(PingPongProducer.DESTINATION_COUNT_PROPNAME));
// This is synchronized because there is a race condition, which causes one connection to sleep if
// all threads try to create connection concurrently.
@@ -208,8 +202,8 @@
perThreadSetup._pingClient = new PingClient(brokerDetails, username, password, virtualPath, destinationName,
selector, transacted, persistent, messageSize, verbose,
failAfterCommit, failBeforeCommit, failAfterSend, failBeforeSend,
- failOnce, batchSize, destinationscount, rate, pubsub,
- unique, ackMode, pausetime);
+ failOnce, batchSize, destinationscount, rate, pubsub, unique,
+ ackMode, pausetime);
}
// Start the client connection
perThreadSetup._pingClient.getConnection().start();