You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ar...@apache.org on 2007/08/24 17:58:09 UTC
svn commit: r569429 - in
/incubator/qpid/trunk/qpid/java/perftests/src/main/java: ./
org/apache/qpid/client/message/ org/apache/qpid/config/
org/apache/qpid/ping/ org/apache/qpid/requestreply/ org/apache/qpid/topic/
Author: arnaudsimon
Date: Fri Aug 24 08:58:08 2007
New Revision: 569429
URL: http://svn.apache.org/viewvc?rev=569429&view=rev
Log:
updated for using pure JMS
Added:
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/InitialContextHelper.java (with props)
incubator/qpid/trunk/qpid/java/perftests/src/main/java/perftests.properties
Removed:
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/config/
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongBouncer.java
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/topic/
Modified:
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/client/message/TestMessageFactory.java
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingDurableClient.java
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingLatencyTestPerf.java
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingSendOnlyClient.java
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongTestPerf.java
Modified: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/client/message/TestMessageFactory.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/client/message/TestMessageFactory.java?rev=569429&r1=569428&r2=569429&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/client/message/TestMessageFactory.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/client/message/TestMessageFactory.java Fri Aug 24 08:58:08 2007
@@ -20,16 +20,12 @@
*/
package org.apache.qpid.client.message;
-import org.apache.mina.common.ByteBuffer;
-import org.apache.mina.common.SimpleByteBufferAllocator;
-
import javax.jms.JMSException;
import javax.jms.Session;
import javax.jms.ObjectMessage;
import javax.jms.StreamMessage;
import javax.jms.BytesMessage;
import javax.jms.TextMessage;
-import javax.jms.Queue;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
@@ -42,15 +38,6 @@
return session.createTextMessage(createMessagePayload(size));
}
- public static JMSTextMessage newJMSTextMessage(int size, String encoding) throws JMSException
- {
- ByteBuffer byteBuffer = (new SimpleByteBufferAllocator()).allocate(size, true);
- JMSTextMessage message = new JMSTextMessage(byteBuffer, encoding);
- message.clearBody();
- message.setText(createMessagePayload(size));
- return message;
- }
-
public static BytesMessage newBytesMessage(Session session, int size) throws JMSException
{
BytesMessage message = session.createBytesMessage();
@@ -78,7 +65,8 @@
}
/**
- * Creates an ObjectMessage with given size and sets the JMS properties (JMSReplyTo and DeliveryMode)
+ * Creates an ObjectMessage with given size and sets the JMS properties
+ * (JMSReplyTo and DeliveryMode)
* @param session
* @param replyDestination
* @param size
Modified: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingDurableClient.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingDurableClient.java?rev=569429&r1=569428&r2=569429&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingDurableClient.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingDurableClient.java Fri Aug 24 08:58:08 2007
@@ -116,7 +116,6 @@
defaults.setProperty(PERSISTENT_MODE_PROPNAME, "true");
defaults.setProperty(TX_BATCH_SIZE_PROPNAME, "10");
defaults.setProperty(RATE_PROPNAME, "20");
- defaults.setProperty(DURABLE_DESTS_PROPNAME, "true");
defaults.setProperty(NUM_MESSAGES_TO_ACTION_PROPNAME, NUM_MESSAGES_TO_ACTION_DEFAULT);
}
Modified: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingLatencyTestPerf.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingLatencyTestPerf.java?rev=569429&r1=569428&r2=569429&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingLatencyTestPerf.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingLatencyTestPerf.java Fri Aug 24 08:58:08 2007
@@ -25,9 +25,6 @@
import org.apache.log4j.Logger;
-import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.client.message.AMQMessage;
-import org.apache.qpid.framing.AMQShortString;
import org.apache.qpid.requestreply.PingPongProducer;
import uk.co.thebadgerset.junit.extensions.TimingController;
@@ -36,7 +33,6 @@
import javax.jms.JMSException;
import javax.jms.Message;
-import javax.jms.ObjectMessage;
import java.util.Collections;
import java.util.HashMap;
@@ -246,9 +242,7 @@
public BatchedResultsListener(int batchSize)
{
_batchSize = batchSize;
- _strictAMQP =
- Boolean.parseBoolean(System.getProperties().getProperty(AMQSession.STRICT_AMQP,
- AMQSession.STRICT_AMQP_DEFAULT));
+ _strictAMQP = false;
}
/**
Modified: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingSendOnlyClient.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingSendOnlyClient.java?rev=569429&r1=569428&r2=569429&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingSendOnlyClient.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingSendOnlyClient.java Fri Aug 24 08:58:08 2007
@@ -26,7 +26,6 @@
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
-import javax.jms.ObjectMessage;
import org.apache.log4j.Logger;
Added: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/InitialContextHelper.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/InitialContextHelper.java?rev=569429&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/InitialContextHelper.java (added)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/InitialContextHelper.java Fri Aug 24 08:58:08 2007
@@ -0,0 +1,48 @@
+/* Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.qpid.requestreply;
+
+import javax.naming.Context;
+import javax.naming.InitialContext;
+import javax.naming.NamingException;
+import java.util.Properties;
+import java.io.InputStream;
+import java.io.IOException;
+
+/**
+ *
+ *
+ */
+public class InitialContextHelper
+{
+
+ public static Context getInitialContext(String propertyFile) throws IOException, NamingException
+ {
+ if ((propertyFile == null) || (propertyFile.length() == 0))
+ {
+ propertyFile = "/perftests.properties";
+ }
+
+ Properties fileProperties = new Properties();
+ ClassLoader cl = Thread.currentThread().getContextClassLoader();
+ // NB: Need to change path to reflect package if moving classes around !
+ InputStream is = cl.getResourceAsStream(propertyFile);
+ fileProperties.load(is);
+ return new InitialContext(fileProperties);
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/InitialContextHelper.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java?rev=569429&r1=569428&r2=569429&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java Fri Aug 24 08:58:08 2007
@@ -21,17 +21,8 @@
package org.apache.qpid.requestreply;
import org.apache.log4j.Logger;
-import org.apache.log4j.NDC;
-import org.apache.qpid.AMQException;
-import org.apache.qpid.client.*;
-import org.apache.qpid.client.message.AMQMessage;
import org.apache.qpid.client.message.TestMessageFactory;
-import org.apache.qpid.exchange.ExchangeDefaults;
-import org.apache.qpid.framing.AMQShortString;
-import org.apache.qpid.jms.MessageProducer;
-import org.apache.qpid.jms.Session;
-import org.apache.qpid.url.URLSyntaxException;
import org.apache.qpid.util.CommandLineParser;
import uk.co.thebadgerset.junit.extensions.BatchedThrottle;
@@ -39,6 +30,7 @@
import uk.co.thebadgerset.junit.extensions.util.ParsedProperties;
import javax.jms.*;
+import javax.naming.Context;
import java.io.IOException;
import java.net.InetAddress;
@@ -56,16 +48,16 @@
* 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/>
* <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. A complete list of accepted parameters, default values and comments on their usage is provided here:
- *
+ * <p/>
* <p/><table><caption>Parameters</caption>
* <tr><th> Parameter <th> Default <th> Comments
* <tr><td> messageSize <td> 0 <td> Message size in bytes. Not including any headers.
@@ -92,27 +84,27 @@
* <tr><td> uniqueDests <td> true <td> Whether each receivers only listens to one ping destination or all.
* <tr><td> durableDests <td> false <td> Whether or not durable destinations are used.
* <tr><td> ackMode <td> AUTO_ACK <td> The message acknowledgement mode. Possible values are:
- * 0 - SESSION_TRANSACTED
- * 1 - AUTO_ACKNOWLEDGE
- * 2 - CLIENT_ACKNOWLEDGE
- * 3 - DUPS_OK_ACKNOWLEDGE
- * 257 - NO_ACKNOWLEDGE
- * 258 - PRE_ACKNOWLEDGE
+ * 0 - SESSION_TRANSACTED
+ * 1 - AUTO_ACKNOWLEDGE
+ * 2 - CLIENT_ACKNOWLEDGE
+ * 3 - DUPS_OK_ACKNOWLEDGE
+ * 257 - NO_ACKNOWLEDGE
+ * 258 - PRE_ACKNOWLEDGE
* <tr><td> consTransacted <td> false <td> Whether or not consumers use transactions. Defaults to the same value
- * as the 'transacted' option if not seperately defined.
+ * as the 'transacted' option if not seperately defined.
* <tr><td> consAckMode <td> AUTO_ACK <td> The message acknowledgement mode for consumers. Defaults to the same
- * value as 'ackMode' if not seperately defined.
+ * value as 'ackMode' if not seperately defined.
* <tr><td> maxPending <td> 0 <td> The maximum size in bytes, of messages sent but not yet received.
- * Limits the volume of messages currently buffered on the client
- * or broker. Can help scale test clients by limiting amount of buffered
- * data to avoid out of memory errors.
+ * Limits the volume of messages currently buffered on the client
+ * or broker. Can help scale test clients by limiting amount of buffered
+ * data to avoid out of memory errors.
* </table>
- *
+ * <p/>
* <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/>
* <p/><table id="crc"><caption>CRC Card</caption>
* <tr><th> Responsibilities <th> Collaborations
* <tr><td> Provide a ping and wait for all responses cycle.
@@ -120,168 +112,249 @@
* </table>
*
* @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.
- *
+ * 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.
- * Obtain 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.
- *
+ * Obtain 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.
+ * 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
{
- /** Used for debugging. */
- private static final Logger log = Logger.getLogger(PingPongProducer.class);
+ /**
+ * Used for debugging.
+ */
+ private static final Logger _log = Logger.getLogger(PingPongProducer.class);
- /** Holds the name of the property to get the test message size from. */
+ /**
+ * Helds the factory name
+ */
+ public static final String FACTORY_NAME_PROPNAME = "factoryName";
+ public static final String FACTORY_NAME_DEAFULT = "local";
+
+ /**
+ * Helds the file properties name
+ */
+ public static final String FILE_PROPERTIES_PROPNAME = "properties";
+ public static final String FILE_PROPERTIES_DEAFULT = "/perftests.properties";
+
+ /**
+ * Holds the name of the property to get the test message size from.
+ */
public static final String MESSAGE_SIZE_PROPNAME = "messageSize";
- /** Used to set up a default message size. */
+ /**
+ * 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 ping queue name from. */
+ /**
+ * 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 default destination to send pings on. */
+ /**
+ * 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 delivery mode from. */
+ /**
+ * Holds the name of the property to get the test delivery mode from.
+ */
public static final String PERSISTENT_MODE_PROPNAME = "persistent";
- /** Holds the message delivery mode to use for the test. */
+ /**
+ * 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 test transactional mode from. */
+ /**
+ * Holds the name of the property to get the test transactional mode from.
+ */
public static final String TRANSACTED_PROPNAME = "transacted";
- /** Holds the transactional mode to use for the test. */
+ /**
+ * Holds the transactional mode to use for the test.
+ */
public static final boolean TRANSACTED_DEFAULT = false;
public static final String CONSUMER_TRANSACTED_PROPNAME = "consTransacted";
public static final boolean CONSUMER_TRANSACTED_DEFAULT = false;
- /** Holds the name of the property to get the test broker url from. */
- public static final String BROKER_PROPNAME = "broker";
-
- /** Holds the default broker url for the test. */
- public static final String BROKER_DEFAULT = "tcp://localhost:5672";
-
- /** Holds the name of the property to get the test broker virtual path. */
- public static final String VIRTUAL_HOST_PROPNAME = "virtualHost";
-
- /** Holds the default virtual path for the test. */
- public static final String VIRTUAL_HOST_DEFAULT = "";
-
- /** Holds the name of the property to get the message rate from. */
+ /**
+ * Holds the name of the property to get the message rate from.
+ */
public static final String RATE_PROPNAME = "rate";
- /** Defines the default rate (in pings per second) to send pings at. 0 means as fast as possible, no restriction. */
+ /**
+ * 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;
- /** Holds the name of the property to get the verbose mode proeprty from. */
+ /**
+ * Holds the name of the property to get the verbose mode proeprty from.
+ */
public static final String VERBOSE_PROPNAME = "verbose";
- /** Holds the default verbose mode. */
+ /**
+ * Holds the default verbose mode.
+ */
public static final boolean VERBOSE_DEFAULT = false;
- /** Holds the name of the property to get the p2p or pub/sub messaging mode from. */
+ /**
+ * Holds the name of the property to get the p2p or pub/sub messaging mode from.
+ */
public static final String PUBSUB_PROPNAME = "pubsub";
- /** Holds the pub/sub mode default, true means ping a topic, false means ping a queue. */
+ /**
+ * 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 fail after commit flag from. */
+ /**
+ * Holds the name of the property to get the fail after commit flag from.
+ */
public static final String FAIL_AFTER_COMMIT_PROPNAME = "failAfterCommit";
- /** Holds the default failover after commit test flag. */
+ /**
+ * Holds the default failover after commit test flag.
+ */
public static final boolean FAIL_AFTER_COMMIT_DEFAULT = false;
- /** Holds the name of the proeprty to get the fail before commit flag from. */
+ /**
+ * Holds the name of the proeprty to get the fail before commit flag from.
+ */
public static final String FAIL_BEFORE_COMMIT_PROPNAME = "failBeforeCommit";
- /** Holds the default failover before commit test flag. */
+ /**
+ * Holds the default failover before commit test flag.
+ */
public static final boolean FAIL_BEFORE_COMMIT_DEFAULT = false;
- /** Holds the name of the proeprty to get the fail after send flag from. */
+ /**
+ * 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 default failover after send test flag. */
+ /**
+ * Holds the default failover after send test flag.
+ */
public static final boolean FAIL_AFTER_SEND_DEFAULT = false;
- /** Holds the name of the property to get the fail before send flag from. */
+ /**
+ * Holds the name of the property to get the fail before send flag from.
+ */
public static final String FAIL_BEFORE_SEND_PROPNAME = "failBeforeSend";
- /** Holds the default failover before send test flag. */
+ /**
+ * Holds the default failover before send test flag.
+ */
public static final boolean FAIL_BEFORE_SEND_DEFAULT = false;
- /** Holds the name of the property to get the fail once flag from. */
+ /**
+ * Holds the name of the property to get the fail once flag from.
+ */
public static final String FAIL_ONCE_PROPNAME = "failOnce";
- /** The default failover once flag, true means only do one failover, false means failover on every commit cycle. */
+ /**
+ * The default failover once flag, true means only do one failover, false means failover on every commit cycle.
+ */
public static final boolean FAIL_ONCE_DEFAULT = true;
- /** Holds the name of the property to get the broker access username from. */
+ /**
+ * Holds the name of the property to get the broker access username from.
+ */
public static final String USERNAME_PROPNAME = "username";
- /** Holds the default broker log on username. */
+ /**
+ * Holds the default broker _log on username.
+ */
public static final String USERNAME_DEFAULT = "guest";
- /** Holds the name of the property to get the broker access password from. */
+ /**
+ * Holds the name of the property to get the broker access password from.
+ */
public static final String PASSWORD_PROPNAME = "password";
- /** Holds the default broker log on password. */
+ /**
+ * Holds the default broker _log on password.
+ */
public static final String PASSWORD_DEFAULT = "guest";
- /** Holds the name of the proeprty to get the. */
+ /**
+ * Holds the name of the proeprty to get the.
+ */
public static final String SELECTOR_PROPNAME = "selector";
- /** Holds the default message selector. */
+ /**
+ * Holds the default message selector.
+ */
public static final String SELECTOR_DEFAULT = "";
- /** Holds the name of the property to get the destination count from. */
+ /**
+ * Holds the name of the property to get the destination count from.
+ */
public static final String DESTINATION_COUNT_PROPNAME = "destinationCount";
- /** Defines the default number of destinations to ping. */
+ /**
+ * Defines the default number of destinations to ping.
+ */
public static final int DESTINATION_COUNT_DEFAULT = 1;
- /** Holds the name of the property to get the number of consumers per destination from. */
+ /**
+ * Holds the name of the property to get the number of consumers per destination from.
+ */
public static final String NUM_CONSUMERS_PROPNAME = "numConsumers";
- /** Defines the default number consumers per destination. */
+ /**
+ * Defines the default number consumers per destination.
+ */
public static final int NUM_CONSUMERS_DEFAULT = 1;
- /** Holds the name of the property to get the waiting timeout for response messages. */
+ /**
+ * Holds the name of the property to get the waiting timeout for response messages.
+ */
public static final String TIMEOUT_PROPNAME = "timeout";
- /** Default time to wait before assuming that a ping has timed out. */
+ /**
+ * Default time to wait before assuming that a ping has timed out.
+ */
public static final long TIMEOUT_DEFAULT = 30000;
- /** Holds the name of the property to get the commit batch size from. */
+ /**
+ * Holds the name of the property to get the commit batch size from.
+ */
public static final String TX_BATCH_SIZE_PROPNAME = "commitBatchSize";
- /** Defines the default number of pings to send in each transaction when running transactionally. */
+ /**
+ * 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 name of the property to get the unique destinations flag from. */
+ /**
+ * Holds the name of the property to get the unique destinations flag from.
+ */
public static final String UNIQUE_DESTS_PROPNAME = "uniqueDests";
- /** Defines the default value for the unique destinations property. */
+ /**
+ * Defines the default value for the unique destinations property.
+ */
public static final boolean UNIQUE_DESTS_DEFAULT = true;
- public static final String DURABLE_DESTS_PROPNAME = "durableDests";
- public static final boolean DURABLE_DESTS_DEFAULT = false;
-
- /** Holds the name of the proeprty to get the message acknowledgement mode from. */
+ /**
+ * Holds the name of the proeprty to get the message acknowledgement mode from.
+ */
public static final String ACK_MODE_PROPNAME = "ackMode";
- /** Defines the default message acknowledgement mode. */
+ /**
+ * Defines the default message acknowledgement mode.
+ */
public static final int ACK_MODE_DEFAULT = Session.AUTO_ACKNOWLEDGE;
public static final String CONSUMER_ACK_MODE_PROPNAME = "consAckMode";
@@ -290,27 +363,37 @@
public static final String MAX_PENDING_PROPNAME = "maxPending";
public static final int MAX_PENDING_DEFAULT = 0;
- /** Defines the default prefetch size to use when consuming messages. */
+ /**
+ * Defines the default prefetch size to use when consuming messages.
+ */
public static final int PREFETCH_DEFAULT = 100;
- /** Defines the default value of the no local flag to use when consuming messages. */
+ /**
+ * Defines the default value of the no local flag to use when consuming messages.
+ */
public static final boolean NO_LOCAL_DEFAULT = false;
- /** Defines the default value of the exclusive flag to use when consuming messages. */
+ /**
+ * 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. */
+ /**
+ * Holds the name of the property to store nanosecond timestamps in ping messages with.
+ */
public static final String MESSAGE_TIMESTAMP_PROPNAME = "timestamp";
- /** Holds the default configuration properties. */
+ /**
+ * Holds the default configuration properties.
+ */
public static ParsedProperties defaults = new ParsedProperties();
static
{
- defaults.setPropertyIfNull(BROKER_PROPNAME, BROKER_DEFAULT);
+ defaults.setPropertyIfNull(FILE_PROPERTIES_PROPNAME, FILE_PROPERTIES_DEAFULT);
+ defaults.setPropertyIfNull(FACTORY_NAME_PROPNAME, FACTORY_NAME_DEAFULT);
defaults.setPropertyIfNull(USERNAME_PROPNAME, USERNAME_DEFAULT);
defaults.setPropertyIfNull(PASSWORD_PROPNAME, PASSWORD_DEFAULT);
- defaults.setPropertyIfNull(VIRTUAL_HOST_PROPNAME, VIRTUAL_HOST_DEFAULT);
defaults.setPropertyIfNull(PING_QUEUE_NAME_PROPNAME, PING_QUEUE_NAME_DEFAULT);
defaults.setPropertyIfNull(SELECTOR_PROPNAME, SELECTOR_DEFAULT);
defaults.setPropertyIfNull(TRANSACTED_PROPNAME, TRANSACTED_DEFAULT);
@@ -322,7 +405,6 @@
defaults.setPropertyIfNull(VERBOSE_PROPNAME, VERBOSE_DEFAULT);
defaults.setPropertyIfNull(PUBSUB_PROPNAME, PUBSUB_DEFAULT);
defaults.setPropertyIfNull(UNIQUE_DESTS_PROPNAME, UNIQUE_DESTS_DEFAULT);
- defaults.setPropertyIfNull(DURABLE_DESTS_PROPNAME, DURABLE_DESTS_DEFAULT);
defaults.setPropertyIfNull(FAIL_BEFORE_COMMIT_PROPNAME, FAIL_BEFORE_COMMIT_DEFAULT);
defaults.setPropertyIfNull(FAIL_AFTER_COMMIT_PROPNAME, FAIL_AFTER_COMMIT_DEFAULT);
defaults.setPropertyIfNull(FAIL_BEFORE_SEND_PROPNAME, FAIL_BEFORE_SEND_DEFAULT);
@@ -336,6 +418,8 @@
defaults.setPropertyIfNull(MAX_PENDING_PROPNAME, MAX_PENDING_DEFAULT);
}
+ protected String _factoryName;
+ protected String _fileProperties;
protected String _brokerDetails;
protected String _username;
protected String _password;
@@ -345,54 +429,81 @@
protected boolean _transacted;
protected boolean _consTransacted;
- /** 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. */
+ /**
+ * Holds the acknowledgement mode used for sending and receiving messages.
+ */
protected int _ackMode;
protected int _consAckMode;
- /** 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. */
+ /**
+ * Used to indicate that the ping loop should print out whenever it pings.
+ */
protected boolean _verbose;
- /** Flag used to indicate if this is a point to point or pub/sub ping client. */
+ /**
+ * Flag used to indicate if this is a point to point or pub/sub ping client.
+ */
protected boolean _isPubSub;
- /** Flag used to indicate if the destinations should be unique client. */
+ /**
+ * Flag used to indicate if the destinations should be unique client.
+ */
protected boolean _isUnique;
- /** Flag used to indicate that durable destination should be used. */
- protected boolean _isDurable;
-
- /** 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;
- /** 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;
- /** 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;
- /** 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;
- /** 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;
- /** Holds the number of sends that should be performed in every transaction when using transactions. */
+ /**
+ * Holds the number of sends that should be performed in every transaction when using transactions.
+ */
protected int _txBatchSize;
- /** Holds the number of destinations to ping. */
+ /**
+ * Holds the number of destinations to ping.
+ */
protected int _noOfDestinations;
- /** Holds the number of consumers per destination. */
+ /**
+ * Holds the number of consumers per destination.
+ */
protected int _noOfConsumers;
- /** Holds the maximum send rate in herz. */
+ /**
+ * Holds the maximum send rate in herz.
+ */
protected int _rate;
/**
@@ -407,10 +518,14 @@
*/
protected Object _sendPauseMonitor = new Object();
- /** Keeps a count of the number of message currently sent but not received. */
+ /**
+ * Keeps a count of the number of message currently sent but not received.
+ */
protected AtomicInteger _unreceived = new AtomicInteger(0);
- /** 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);
/**
@@ -418,33 +533,51 @@
* ping producers on the same JVM.
*/
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");
- /** Holds the connection to the broker. */
+ /**
+ * Holds the connection to the broker.
+ */
protected Connection _connection;
- /** Holds the consumer connections. */
+ /**
+ * Holds the consumer connections.
+ */
protected Connection[] _consumerConnection;
- /** Holds the controlSession on which ping replies are received. */
+ /**
+ * Holds the controlSession on which ping replies are received.
+ */
protected Session[] _consumerSession;
- /** Holds the producer controlSession, needed to create ping messages. */
- protected Session _producerSession;
+ /**
+ * Holds the producer controlSession, needed to create ping messages.
+ */
+ protected Session _producerSession = (Session) _connection.createSession(_transacted, _ackMode);
- /** Holds the destination where the response messages will arrive. */
+ /**
+ * Holds the destination where the response messages will arrive.
+ */
protected Destination _replyDestination;
- /** Holds the set of destinations that this ping producer pings. */
+ /**
+ * Holds the set of destinations that this ping producer pings.
+ */
protected List<Destination> _pingDestinations;
- /** Used to restrict the sending rate to a specified limit. */
+ /**
+ * Used to restrict the sending rate to a specified limit.
+ */
protected Throttle _rateLimiter;
- /** 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.
+ */
protected ChainedMessageListener _chainedMessageListener = null;
/**
@@ -459,20 +592,29 @@
*/
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 message producer to send the pings through. */
+ /**
+ * Holds the message producer to send the pings through.
+ */
protected MessageProducer _producer;
- /** Holds the message consumer to receive the ping replies through. */
+ /**
+ * Holds the message consumer to receive the ping replies through.
+ */
protected MessageConsumer[] _consumer;
- /** The prompt to display when asking the user to kill the broker for failover testing. */
+ /**
+ * The prompt to display when asking the user to kill the broker for failover testing.
+ */
private static final String KILL_BROKER_PROMPT = "Kill broker now, then press Return.";
- private String _clientID;
- /** Keeps count of the total messages sent purely for debugging purposes. */
+ /**
+ * Keeps count of the total messages sent purely for debugging purposes.
+ */
private static AtomicInteger numSent = new AtomicInteger();
/**
@@ -480,23 +622,20 @@
* 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.
*
- * @param overrides Properties containing any desired overrides to the defaults.
- *
+ * @param overrides Properties containing any desired overrides to the defaults.
* @throws Exception Any exceptions are allowed to fall through.
*/
public PingPongProducer(Properties overrides) throws Exception
{
- // log.debug("public PingPongProducer(Properties overrides = " + overrides + "): called");
-
+ // _log.debug("public PingPongProducer(Properties overrides = " + overrides + "): called");
// Create a set of parsed properties from the defaults overriden by the passed in values.
ParsedProperties properties = new ParsedProperties(defaults);
properties.putAll(overrides);
-
// Extract the configuration properties to set the pinger up with.
- _brokerDetails = properties.getProperty(BROKER_PROPNAME);
+ _factoryName = properties.getProperty(FACTORY_NAME_PROPNAME);
+ _fileProperties = properties.getProperty(FILE_PROPERTIES_PROPNAME);
_username = properties.getProperty(USERNAME_PROPNAME);
_password = properties.getProperty(PASSWORD_PROPNAME);
- _virtualpath = properties.getProperty(VIRTUAL_HOST_PROPNAME);
_destinationName = properties.getProperty(PING_QUEUE_NAME_PROPNAME);
_selector = properties.getProperty(SELECTOR_PROPNAME);
_transacted = properties.getPropertyAsBoolean(TRANSACTED_PROPNAME);
@@ -515,26 +654,20 @@
_rate = properties.getPropertyAsInteger(RATE_PROPNAME);
_isPubSub = properties.getPropertyAsBoolean(PUBSUB_PROPNAME);
_isUnique = properties.getPropertyAsBoolean(UNIQUE_DESTS_PROPNAME);
- _isDurable = properties.getPropertyAsBoolean(DURABLE_DESTS_PROPNAME);
_ackMode = properties.getPropertyAsInteger(ACK_MODE_PROPNAME);
_consAckMode = properties.getPropertyAsInteger(CONSUMER_ACK_MODE_PROPNAME);
_maxPendingSize = properties.getPropertyAsInteger(MAX_PENDING_PROPNAME);
-
// Check that one or more destinations were specified.
if (_noOfDestinations < 1)
{
throw new IllegalArgumentException("There must be at least one destination.");
}
-
// Set up a throttle to control the send rate, if a rate > 0 is specified.
if (_rate > 0)
{
_rateLimiter = new BatchedThrottle();
_rateLimiter.setRate(_rate);
}
-
- // Create the connection and message producers/consumers.
- // establishConnection(true, true);
}
/**
@@ -543,33 +676,32 @@
*
* @param producer Flag to indicate whether or not the producer should be set up.
* @param consumer Flag to indicate whether or not the consumers should be set up.
- *
* @throws Exception Any exceptions are allowed to fall through.
*/
public void establishConnection(boolean producer, boolean consumer) throws Exception
{
- // log.debug("public void establishConnection(): called");
+ // _log.debug("public void establishConnection(): called");
// Generate a unique identifying name for this client, based on it ip address and the current time.
InetAddress address = InetAddress.getLocalHost();
- _clientID = address.getHostName() + System.currentTimeMillis();
+ String _clientID = address.getHostName() + System.currentTimeMillis();
// Create a connection to the broker.
createConnection(_clientID);
// Create transactional or non-transactional sessions, based on the command line arguments.
- _producerSession = (Session) _connection.createSession(_transacted, _ackMode);
+ _producerSession = _connection.createSession(_transacted, _ackMode);
_consumerSession = new Session[_noOfConsumers];
for (int i = 0; i < _noOfConsumers; i++)
{
- _consumerSession[i] = (Session) _consumerConnection[i].createSession(_consTransacted, _consAckMode);
+ _consumerSession[i] = _consumerConnection[i].createSession(_consTransacted, _consAckMode);
}
// Create the destinations to send pings to and receive replies from.
_replyDestination = _consumerSession[0].createTemporaryQueue();
- createPingDestinations(_noOfDestinations, _selector, _destinationName, _isUnique, _isDurable);
+ createPingDestinations(_noOfDestinations, _selector, _destinationName);
// Create the message producer only if instructed to.
if (producer)
@@ -589,25 +721,24 @@
* created with.
*
* @param clientID The clients identifier.
- *
- * @throws AMQException Any underlying exceptions are allowed to fall through.
- * @throws URLSyntaxException Any underlying exceptions are allowed to fall through.
+ * @throws Exception Any underlying exceptions are allowed to fall through.
*/
- protected void createConnection(String clientID) throws AMQException, URLSyntaxException
+ protected void createConnection(String clientID) throws Exception
{
- // log.debug("protected void createConnection(String clientID = " + clientID + "): called");
-
- // log.debug("Creating a connection for the message producer.");
+ // _log.debug("protected void createConnection(String clientID = " + clientID + "): called");
- _connection = new AMQConnection(_brokerDetails, _username, _password, clientID, _virtualpath);
+ // _log.debug("Creating a connection for the message producer.");
+ Context context = InitialContextHelper.getInitialContext(_fileProperties);
+ ConnectionFactory factory = (ConnectionFactory) context.lookup(_factoryName);
+ _connection = factory.createConnection(_username, _password);
- // log.debug("Creating " + _noOfConsumers + " connections for the consumers.");
+ // _log.debug("Creating " + _noOfConsumers + " connections for the consumers.");
_consumerConnection = new Connection[_noOfConsumers];
for (int i = 0; i < _noOfConsumers; i++)
{
- _consumerConnection[i] = new AMQConnection(_brokerDetails, _username, _password, clientID, _virtualpath);
+ _consumerConnection[i] = factory.createConnection(_username, _password);
}
}
@@ -621,8 +752,8 @@
{
try
{
- Properties options =
- CommandLineParser.processCommandLine(args, new CommandLineParser(new String[][] {}), System.getProperties());
+ Properties options = CommandLineParser
+ .processCommandLine(args, new CommandLineParser(new String[][]{}), System.getProperties());
// Create a ping producer overriding its defaults with all options passed on the command line.
PingPongProducer pingProducer = new PingPongProducer(options);
@@ -645,7 +776,7 @@
catch (Exception e)
{
System.err.println(e.getMessage());
- log.error("Top level handler caught execption.", e);
+ _log.error("Top level handler caught execption.", e);
System.exit(1);
}
}
@@ -664,7 +795,9 @@
Thread.sleep(sleepTime);
}
catch (InterruptedException ie)
- { }
+ {
+ // do nothing
+ }
}
}
@@ -676,12 +809,12 @@
*/
public List<Destination> getReplyDestinations()
{
- // log.debug("public List<Destination> getReplyDestinations(): called");
+ // _log.debug("public List<Destination> getReplyDestinations(): called");
List<Destination> replyDestinations = new ArrayList<Destination>();
replyDestinations.add(_replyDestination);
- // log.debug("replyDestinations = " + replyDestinations);
+ // _log.debug("replyDestinations = " + replyDestinations);
return replyDestinations;
}
@@ -694,84 +827,58 @@
*/
public void createProducer() throws JMSException
{
- // log.debug("public void createProducer(): called");
-
- _producer = (MessageProducer) _producerSession.createProducer(null);
+ _producer = _producerSession.createProducer(null);
_producer.setDeliveryMode(_persistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);
-
- // log.debug("Created producer for " + (_persistent ? "persistent" : "non-persistent") + " messages.");
+ if (_log.isDebugEnabled())
+ {
+ _log.debug("Created producer for " + (_persistent ? "persistent" : "non-persistent") + " messages.");
+ }
}
/**
* 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.
* @throws JMSException Any JMSExceptions are allowed to fall through.
*/
- public void createPingDestinations(int noOfDestinations, String selector, String rootName, boolean unique,
- boolean durable) throws JMSException, AMQException
+ public void createPingDestinations(int noOfDestinations, String selector, String rootName) throws JMSException
{
- /*log.debug("public void createPingDestinations(int noOfDestinations = " + noOfDestinations + ", String selector = "
- + selector + ", String rootName = " + rootName + ", boolean unique = " + unique + ", boolean durable = "
- + durable + "): called");*/
-
+ if (_log.isDebugEnabled())
+ {
+ _log.debug(
+ "public void createPingDestinations(int noOfDestinations = " + noOfDestinations + ", String selector = " + selector + ", String rootName = " + rootName);
+ }
_pingDestinations = new ArrayList<Destination>();
-
// Create the desired number of ping destinations and consumers for them.
- // log.debug("Creating " + noOfDestinations + " destinations to ping.");
-
+ if (_log.isDebugEnabled())
+ {
+ _log.debug("Creating " + noOfDestinations + " destinations to ping.");
+ }
+ String id;
for (int i = 0; i < noOfDestinations; i++)
{
- AMQDestination destination;
-
- String id;
-
- // Generate an id, unique within this pinger or to the whole JVM depending on the unique flag.
- if (unique)
- {
- // log.debug("Creating unique destinations.");
- id = "_" + _queueJVMSequenceID.incrementAndGet() + "_" + _connection.getClientID();
- }
- else
- {
- // log.debug("Creating shared destinations.");
- id = "_" + _queueSharedID.incrementAndGet();
- }
-
+ Destination destination;
+ id = "_" + _queueSharedID.incrementAndGet();
// Check if this is a pub/sub pinger, in which case create topics.
if (_isPubSub)
{
- if (!durable)
+ destination = _producerSession.createTopic(rootName + id);
+ if (_log.isDebugEnabled())
{
- destination = new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME, rootName + id);
- // log.debug("Created non-durable topic " + destination);
- }
- else
- {
- destination =
- AMQTopic.createDurableTopic(new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME, rootName + id),
- _clientID, (AMQConnection) _connection);
- // log.debug("Created durable topic " + destination);
+ _log.debug("Created topic " + rootName + id);
}
}
// Otherwise this is a p2p pinger, in which case create queues.
else
{
- AMQShortString destinationName = new AMQShortString(rootName + id);
- destination =
- new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, destinationName, destinationName, false, false,
- _isDurable);
- ((AMQSession) _producerSession).createQueue(destinationName, false, _isDurable, false);
- ((AMQSession) _producerSession).bindQueue(destinationName, destinationName, null,
- ExchangeDefaults.DIRECT_EXCHANGE_NAME);
-
- // log.debug("Created queue " + destination);
+ destination = _producerSession.createQueue(rootName + id);
+ if (_log.isDebugEnabled())
+ {
+ _log.debug("Created queue " + rootName + id);
+ }
}
// Keep the destination.
@@ -784,17 +891,16 @@
*
* @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
{
- /*log.debug("public void createReplyConsumers(Collection<Destination> destinations = " + destinations
+ /*_log.debug("public void createReplyConsumers(Collection<Destination> destinations = " + destinations
+ ", String selector = " + selector + "): called");*/
- // log.debug("There are " + destinations.size() + " destinations.");
- // log.debug("Creating " + _noOfConsumers + " consumers on each destination.");
- // log.debug("Total number of consumers is: " + (destinations.size() * _noOfConsumers));
+ // _log.debug("There are " + destinations.size() + " destinations.");
+ // _log.debug("Creating " + _noOfConsumers + " consumers on each destination.");
+ // _log.debug("Total number of consumers is: " + (destinations.size() * _noOfConsumers));
for (Destination destination : destinations)
{
@@ -803,21 +909,19 @@
for (int i = 0; i < _noOfConsumers; i++)
{
// Create a consumer for the destination and set this pinger to listen to its messages.
- _consumer[i] =
- _consumerSession[i].createConsumer(destination, PREFETCH_DEFAULT, NO_LOCAL_DEFAULT, EXCLUSIVE_DEFAULT,
- selector);
+ _consumer[i] = _consumerSession[i].createConsumer(destination, selector, NO_LOCAL_DEFAULT);
final int consumerNo = i;
_consumer[i].setMessageListener(new MessageListener()
+ {
+ public void onMessage(Message message)
{
- public void onMessage(Message message)
- {
- onMessageWithConsumerNo(message, consumerNo);
- }
- });
+ onMessageWithConsumerNo(message, consumerNo);
+ }
+ });
- // log.debug("Set consumer " + i + " to listen to replies sent to destination: " + destination);
+ // _log.debug("Set consumer " + i + " to listen to replies sent to destination: " + destination);
}
}
}
@@ -831,7 +935,7 @@
*/
public void onMessageWithConsumerNo(Message message, int consumerNo)
{
- // log.debug("public void onMessageWithConsumerNo(Message message, int consumerNo = " + consumerNo + "): called");
+ // _log.debug("public void onMessageWithConsumerNo(Message message, int consumerNo = " + consumerNo + "): called");
try
{
long now = System.nanoTime();
@@ -842,13 +946,13 @@
// Extract the messages correlation id.
String correlationID = message.getJMSCorrelationID();
- // log.debug("correlationID = " + correlationID);
+ // _log.debug("correlationID = " + correlationID);
int num = message.getIntProperty("MSG_NUM");
- // log.info("Message " + num + " received.");
+ // _log.info("Message " + num + " received.");
boolean isRedelivered = message.getJMSRedelivered();
- // log.debug("isRedelivered = " + isRedelivered);
+ // _log.debug("isRedelivered = " + isRedelivered);
if (!isRedelivered)
{
@@ -862,7 +966,7 @@
// Restart the timeout timer on every message.
perCorrelationId.timeOutStart = System.nanoTime();
- // log.debug("Reply was expected, decrementing the latch for the id, " + correlationID);
+ // _log.debug("Reply was expected, decrementing the latch for the id, " + correlationID);
// Decrement the countdown latch. Before this point, it is possible that two threads might enter this
// method simultanesouly with the same correlation id. Decrementing the latch in a synchronized block
@@ -887,7 +991,7 @@
if ((_maxPendingSize > 0) && (unreceivedSize < _maxPendingSize))
// && (_sendPauseBarrier.getNumberWaiting() == 1))
{
- // log.debug("unreceived size estimate under limit = " + unreceivedSize);
+ // _log.debug("unreceived size estimate under limit = " + unreceivedSize);
// Wait on the send pause barrier for the limit to be re-established.
/*try
@@ -908,19 +1012,19 @@
// NDC.push("/rem" + remainingCount);
- // log.debug("remainingCount = " + remainingCount);
- // log.debug("trueCount = " + trueCount);
+ // _log.debug("remainingCount = " + remainingCount);
+ // _log.debug("trueCount = " + trueCount);
// Commit on transaction batch size boundaries. At this point in time the waiting producer remains
// blocked, even on the last message.
// Commit count is divided by noOfConsumers in p2p mode, so that each consumer only commits on
// each batch boundary. For pub/sub each consumer gets every message so no division is done.
long commitCount = _isPubSub ? remainingCount : (remainingCount / _noOfConsumers);
- // log.debug("commitCount = " + commitCount);
+ // _log.debug("commitCount = " + commitCount);
if ((commitCount % _txBatchSize) == 0)
{
- // log.debug("Trying commit for consumer " + consumerNo + ".");
+ // _log.debug("Trying commit for consumer " + consumerNo + ".");
commitTx(_consumerSession[consumerNo]);
}
@@ -940,12 +1044,12 @@
}
else
{
- log.warn("Got unexpected message with correlationId: " + correlationID);
+ _log.warn("Got unexpected message with correlationId: " + correlationID);
}
}
else
{
- log.warn("Got redelivered message, ignoring.");
+ _log.warn("Got redelivered message, ignoring.");
}
// Print out ping times for every message in verbose mode only.
@@ -956,17 +1060,17 @@
if (timestamp != null)
{
long diff = System.nanoTime() - timestamp;
- //log.trace("Time for round trip (nanos): " + diff);
+ //_log.trace("Time for round trip (nanos): " + diff);
}
}*/
}
catch (JMSException e)
{
- log.warn("There was a JMSException: " + e.getMessage(), e);
+ _log.warn("There was a JMSException: " + e.getMessage(), e);
}
finally
{
- // log.debug("public void onMessageWithConsumerNo(Message message, int consumerNo): ending");
+ // _log.debug("public void onMessageWithConsumerNo(Message message, int consumerNo): ending");
// NDC.clear();
}
}
@@ -980,17 +1084,15 @@
* @param numPings The number of ping messages to send.
* @param timeout The timeout in milliseconds.
* @param messageCorrelationId The message correlation id. If this is null, one is generated.
- *
* @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
{
- /*log.debug("public int pingAndWaitForReply(Message message, int numPings = " + numPings + ", long timeout = "
+ /*_log.debug("public int pingAndWaitForReply(Message message, int numPings = " + numPings + ", long timeout = "
+ timeout + ", String messageCorrelationId = " + messageCorrelationId + "): called");*/
// Generate a unique correlation id to put on the messages before sending them, if one was not specified.
@@ -1034,31 +1136,31 @@
allMessagesReceived = numReplies == getExpectedNumPings(numPings);
- // log.debug("numReplies = " + numReplies);
- // log.debug("allMessagesReceived = " + allMessagesReceived);
+ // _log.debug("numReplies = " + numReplies);
+ // _log.debug("allMessagesReceived = " + allMessagesReceived);
// Recheck the timeout condition.
long now = System.nanoTime();
long lastMessageReceievedAt = perCorrelationId.timeOutStart;
timedOut = (now - lastMessageReceievedAt) > (timeout * 1000000);
- // log.debug("now = " + now);
- // log.debug("lastMessageReceievedAt = " + lastMessageReceievedAt);
+ // _log.debug("now = " + now);
+ // _log.debug("lastMessageReceievedAt = " + lastMessageReceievedAt);
}
while (!timedOut && !allMessagesReceived);
if ((numReplies < getExpectedNumPings(numPings)) && _verbose)
{
- log.info("Timed out (" + timeout + " ms) before all replies received on id, " + messageCorrelationId);
+ _log.info("Timed out (" + timeout + " ms) before all replies received on id, " + messageCorrelationId);
}
else if (_verbose)
{
- log.info("Got all replies on id, " + messageCorrelationId);
+ _log.info("Got all replies on id, " + messageCorrelationId);
}
// commitTx(_consumerSession);
- // log.debug("public int pingAndWaitForReply(Message message, int numPings, long timeout): ending");
+ // _log.debug("public int pingAndWaitForReply(Message message, int numPings, long timeout): ending");
return numReplies;
}
@@ -1077,12 +1179,11 @@
* @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
{
- /*log.debug("public void pingNoWaitForReply(Message message, int numPings = " + numPings
+ /*_log.debug("public void pingNoWaitForReply(Message message, int numPings = " + numPings
+ ", String messageCorrelationId = " + messageCorrelationId + "): called");*/
if (message == null)
@@ -1112,7 +1213,7 @@
// Spew out per message timings on every message sonly in verbose mode.
/*if (_verbose)
{
- log.info(timestampFormatter.format(new Date()) + ": Pinged at with correlation id, " + messageCorrelationId);
+ _log.info(timestampFormatter.format(new Date()) + ": Pinged at with correlation id, " + messageCorrelationId);
}*/
}
@@ -1131,15 +1232,13 @@
*
* @param i The count of messages sent so far in a loop of multiple calls to this send method.
* @param message The message to send.
- *
* @return <tt>true</tt> if the messages were committed, <tt>false</tt> otherwise.
- *
* @throws JMSException All underlyiung JMSExceptions are allowed to fall through.
*/
protected boolean sendMessage(int i, Message message) throws JMSException
{
- // log.debug("protected boolean sendMessage(int i = " + i + ", Message message): called");
- // log.debug("_txBatchSize = " + _txBatchSize);
+ // _log.debug("protected boolean sendMessage(int i = " + i + ", Message message): called");
+ // _log.debug("_txBatchSize = " + _txBatchSize);
// Round robin the destinations as the messages are sent.
Destination destination = _pingDestinations.get(i % _pingDestinations.size());
@@ -1152,7 +1251,7 @@
_failBeforeSend = false;
}
- // log.trace("Failing Before Send");
+ // _log.trace("Failing Before Send");
waitForUser(KILL_BROKER_PROMPT);
}
@@ -1167,7 +1266,7 @@
if (unreceivedSize > _maxPendingSize)
{
- // log.debug("unreceived size estimate over limit = " + unreceivedSize);
+ // _log.debug("unreceived size estimate over limit = " + unreceivedSize);
// Wait on the send pause barrier for the limit to be re-established.
try
@@ -1200,7 +1299,7 @@
message.setIntProperty("MSG_NUM", num);
setTimestamp(message);
_producer.send(message);
- // log.info("Message " + num + " sent.");
+ // _log.info("Message " + num + " sent.");
}
else
{
@@ -1208,7 +1307,7 @@
message.setIntProperty("MSG_NUM", num);
setTimestamp(message);
_producer.send(destination, message);
- // log.info("Message " + num + " sent.");
+ // _log.info("Message " + num + " sent.");
}
// Increase the unreceived size, this may actually happen aftern the message is recevied.
@@ -1226,7 +1325,7 @@
// Commit on every transaction batch size boundary. Here i + 1 is the count of actual messages sent.
if (((i + 1) % _txBatchSize) == 0)
{
- // log.debug("Trying commit on producer session.");
+ // _log.debug("Trying commit on producer session.");
committed = commitTx(_producerSession);
}
@@ -1252,12 +1351,12 @@
catch (JMSException e)
{
_publish = false;
- // log.debug("There was a JMSException: " + e.getMessage(), e);
+ // _log.debug("There was a JMSException: " + e.getMessage(), e);
}
catch (InterruptedException e)
{
_publish = false;
- // log.debug("There was an interruption: " + e.getMessage(), e);
+ // _log.debug("There was an interruption: " + e.getMessage(), e);
}
}
@@ -1272,7 +1371,9 @@
_chainedMessageListener = messageListener;
}
- /** Removes any chained message listeners from this pinger. */
+ /**
+ * Removes any chained message listeners from this pinger.
+ */
public void removeChainedMessageListener()
{
_chainedMessageListener = null;
@@ -1284,9 +1385,7 @@
* @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 Message getTestMessage(Destination replyQueue, int messageSize, boolean persistent) throws JMSException
@@ -1302,29 +1401,12 @@
protected void setTimestamp(Message msg) throws JMSException
{
- if (((AMQSession) _producerSession).isStrictAMQP())
- {
- ((AMQMessage) msg).setTimestampProperty(new AMQShortString(MESSAGE_TIMESTAMP_PROPNAME), System.nanoTime());
- }
- else
- {
- msg.setLongProperty(MESSAGE_TIMESTAMP_PROPNAME, System.nanoTime());
- }
+ msg.setLongProperty(MESSAGE_TIMESTAMP_PROPNAME, System.nanoTime());
}
protected long getTimestamp(Message msg) throws JMSException
{
-
- if (((AMQSession) _producerSession).isStrictAMQP())
- {
- Long value = ((AMQMessage) msg).getTimestampProperty(new AMQShortString(MESSAGE_TIMESTAMP_PROPNAME));
-
- return (value == null) ? 0L : value;
- }
- else
- {
- return msg.getLongProperty(PingPongProducer.MESSAGE_TIMESTAMP_PROPNAME);
- }
+ return msg.getLongProperty(PingPongProducer.MESSAGE_TIMESTAMP_PROPNAME);
}
/**
@@ -1346,7 +1428,9 @@
}
}
- /** Implements a ping loop that repeatedly pings until the publish flag becomes false. */
+ /**
+ * Implements a ping loop that repeatedly pings until the publish flag becomes false.
+ */
public void run()
{
// Keep running until the publish flag is cleared.
@@ -1364,7 +1448,7 @@
*/
public void onException(JMSException e)
{
- // log.debug("public void onException(JMSException e = " + e + "): called", e);
+ // _log.debug("public void onException(JMSException e = " + e + "): called", e);
_publish = false;
}
@@ -1377,12 +1461,12 @@
public Thread getShutdownHook()
{
return new Thread(new Runnable()
- {
- public void run()
- {
- stop();
- }
- });
+ {
+ public void run()
+ {
+ stop();
+ }
+ });
}
/**
@@ -1392,14 +1476,14 @@
*/
public void close() throws JMSException
{
- // log.debug("public void close(): called");
+ // _log.debug("public void close(): called");
try
{
if (_connection != null)
{
_connection.close();
- // log.debug("Close connection.");
+ // _log.debug("Close connection.");
}
for (int i = 0; i < _noOfConsumers; i++)
@@ -1407,7 +1491,7 @@
if (_consumerConnection[i] != null)
{
_consumerConnection[i].close();
- // log.debug("Closed consumer connection.");
+ // _log.debug("Closed consumer connection.");
}
}
}
@@ -1427,46 +1511,43 @@
/**
* Convenience method to commit the transaction on the specified controlSession. If the controlSession to commit on is not a
* transactional controlSession, 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.
*
* @param session The controlSession to commit
- *
* @return <tt>true</tt> if the controlSession was committed, <tt>false</tt> if it was not.
- *
* @throws javax.jms.JMSException If the commit fails and then the rollback fails.
- *
* @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 boolean commitTx(Session session) throws JMSException
{
- // log.debug("protected void commitTx(Session session): called");
+ // _log.debug("protected void commitTx(Session session): called");
boolean committed = false;
- // log.trace("Batch time reached");
+ // _log.trace("Batch time reached");
if (_failAfterSend)
{
- // log.trace("Batch size reached");
+ // _log.trace("Batch size reached");
if (_failOnce)
{
_failAfterSend = false;
}
- // log.trace("Failing After Send");
+ // _log.trace("Failing After Send");
waitForUser(KILL_BROKER_PROMPT);
}
if (session.getTransacted())
{
- // log.debug("Session is transacted.");
+ // _log.debug("Session is transacted.");
try
{
@@ -1477,14 +1558,14 @@
_failBeforeCommit = false;
}
- // log.trace("Failing Before Commit");
+ // _log.trace("Failing Before Commit");
waitForUser(KILL_BROKER_PROMPT);
}
long start = System.nanoTime();
session.commit();
committed = true;
- // log.debug("Time taken to commit :" + ((System.nanoTime() - start) / 1000000f) + " ms");
+ // _log.debug("Time taken to commit :" + ((System.nanoTime() - start) / 1000000f) + " ms");
if (_failAfterCommit)
{
@@ -1493,30 +1574,23 @@
_failAfterCommit = false;
}
- // log.trace("Failing After Commit");
+ // _log.trace("Failing After Commit");
waitForUser(KILL_BROKER_PROMPT);
}
- // log.debug("Session Commited.");
+ // _log.debug("Session Commited.");
}
catch (JMSException e)
{
- // log.debug("JMSException on commit:" + e.getMessage(), e);
-
- // Warn that the bounce back client is not available.
- if (e.getLinkedException() instanceof AMQNoConsumersException)
- {
- // log.debug("No consumers on queue.");
- }
-
+ // _log.debug("JMSException on commit:" + e.getMessage(), e);
try
{
session.rollback();
- // log.debug("Message rolled back.");
+ // _log.debug("Message rolled back.");
}
catch (JMSException jmse)
{
- // log.debug("JMSE on rollback:" + jmse.getMessage(), jmse);
+ // _log.debug("JMSE on rollback:" + jmse.getMessage(), jmse);
// Both commit and rollback failed. Throw the rollback exception.
throw jmse;
@@ -1562,14 +1636,13 @@
* Calculates how many pings are expected to be received for the given number sent.
*
* @param numpings The number of pings that will be sent.
- *
* @return The number that should be received, for the test to pass.
*/
public int getExpectedNumPings(int numpings)
{
- // log.debug("public int getExpectedNumPings(int numpings = " + numpings + "): called");
+ // _log.debug("public int getExpectedNumPings(int numpings = " + numpings + "): called");
- // log.debug("Each ping will be received by " + (_isPubSub ? getConsumersPerDestination() : 1) + " consumers.");
+ // _log.debug("Each ping will be received by " + (_isPubSub ? getConsumersPerDestination() : 1) + " consumers.");
return numpings * (_isPubSub ? getConsumersPerDestination() : 1);
}
@@ -1579,7 +1652,7 @@
* PingPongProducer#onMessageWithConsumerNo} 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/>
* <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.
@@ -1593,7 +1666,6 @@
* @param message The newly arrived message.
* @param remainingCount The number of messages left to complete the test.
* @param latency The nanosecond latency of the message.
- *
* @throws JMSException Any JMS exceptions is allowed to fall through.
*/
public void onMessage(Message message, int remainingCount, long latency) throws JMSException;
@@ -1605,10 +1677,14 @@
*/
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/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongTestPerf.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongTestPerf.java?rev=569429&r1=569428&r2=569429&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongTestPerf.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongTestPerf.java Fri Aug 24 08:58:08 2007
@@ -172,10 +172,10 @@
PerThreadSetup perThreadSetup = new PerThreadSetup();
// Extract the test set up paramaeters.
- String brokerDetails = testParameters.getProperty(PingPongProducer.BROKER_PROPNAME);
+ String fileProperties = testParameters.getProperty(PingPongProducer.FILE_PROPERTIES_PROPNAME);
+ String factoryName = testParameters.getProperty(PingPongProducer.FACTORY_NAME_PROPNAME);
String username = testParameters.getProperty(PingPongProducer.USERNAME_PROPNAME);
String password = testParameters.getProperty(PingPongProducer.PASSWORD_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);
@@ -187,7 +187,7 @@
{
// Establish a bounce back client on the ping queue to bounce back the pings.
perThreadSetup._testPingBouncer =
- new PingPongBouncer(brokerDetails, username, password, virtualPath, destinationName, persistent,
+ new PingPongBouncer(fileProperties, factoryName, username, password, destinationName, persistent,
transacted, selector, verbose, pubsub);
// Start the connections for client and producer running.
Added: incubator/qpid/trunk/qpid/java/perftests/src/main/java/perftests.properties
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/perftests.properties?rev=569429&view=auto
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/perftests.properties (added)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/perftests.properties Fri Aug 24 08:58:08 2007
@@ -0,0 +1,39 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+
+java.naming.factory.initial = org.apache.qpid.jndi.PropertiesFileInitialContextFactory
+
+# use the following property to configure the default connector
+#java.naming.provider.url - ignored.
+
+# register some connection factories
+# connectionfactory.[jndiname] = [ConnectionURL]
+connectionfactory.local = amqp://guest:guest@clientid/test?brokerlist='tcp://localhost:5672'
+
+# register some queues in JNDI using the form
+# queue.[jndiName] = [physicalName]
+queue.MyQueue = example.MyQueue
+
+# register some topics in JNDI using the form
+# topic.[jndiName] = [physicalName]
+topic.ibmStocks = stocks.nyse.ibm
+
+# Register an AMQP destination in JNDI
+# NOTE: Qpid currently only supports direct,topics and headers
+# destination.[jniName] = [BindingURL]
+destination.direct = direct://amq.direct//directQueue