/*
*
* 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 org.apache.log4j.Logger;
import org.apache.log4j.NDC;
import org.apache.qpid.test.framework.TestUtils;
import org.apache.qpid.junit.extensions.BatchedThrottle;
import org.apache.qpid.junit.extensions.Throttle;
import org.apache.qpid.junit.extensions.util.CommandLineParser;
import org.apache.qpid.junit.extensions.util.ParsedProperties;
import javax.jms.*;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;
import java.io.*;
import java.net.InetAddress;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
/**
* 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>Parameters</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 destinations to send pings to.
* <tr><td> numConsumers <td> 1 <td> The number of consumers on each destination.
* <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 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
* <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.
* <tr><td> consAckMode <td> AUTO_ACK <td> The message acknowledgement mode for consumers. Defaults to the same
* 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.
* </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/><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 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.
*/
public class PingPongProducer implements Runnable, ExceptionListener
{
/** Used for debugging. */
private static final Logger log = Logger.getLogger(PingPongProducer.class);
/** Holds the name of the property to determine whether of not client id is overridden at connection time. */
public static final String OVERRIDE_CLIENT_ID_PROPNAME = "overrideClientId";
/** Holds the default value of the override client id flag. */
public static final String OVERRIDE_CLIENT_ID_DEAFULT = "false";
/** Holds the name of the property to define the JNDI factory name with. */
public static final String FACTORY_NAME_PROPNAME = "factoryName";
/** Holds the default JNDI name of the connection factory. */
public static final String FACTORY_NAME_DEAFULT = "local";
/** Holds the name of the property to set the JNDI initial context properties with. */
public static final String FILE_PROPERTIES_PROPNAME = "properties";
/** Holds the default file name of the JNDI initial context 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. */
public static final int MESSAGE_SIZE_DEAFULT = 0;
/** 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. */
public static final String PING_QUEUE_NAME_DEFAULT = "ping";
/** Holds the name of the property to get the queue name postfix from. */
public static final String QUEUE_NAME_POSTFIX_PROPNAME = "queueNamePostfix";
/** Holds the default queue name postfix value. */
public static final String QUEUE_NAME_POSTFIX_DEFAULT = "";
/** 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. */
public static final boolean PERSISTENT_MODE_DEFAULT = false;
/** 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. */
public static final boolean TRANSACTED_DEFAULT = false;
/** Holds the name of the property to get the test consumer transacted mode from. */
public static final String CONSUMER_TRANSACTED_PROPNAME = "consTransacted";
/** Holds the consumer transactional mode default setting. */
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. */
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. */
public static final int RATE_DEFAULT = 0;
/** 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. */
public static final boolean VERBOSE_DEFAULT = false;
/** 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. */
public static final boolean PUBSUB_DEFAULT = false;
/** 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. */
public static final boolean FAIL_AFTER_COMMIT_DEFAULT = false;
/** 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. */
public static final boolean FAIL_BEFORE_COMMIT_DEFAULT = false;
/** 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. */
public static final boolean FAIL_AFTER_SEND_DEFAULT = false;
/** 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. */
public static final boolean FAIL_BEFORE_SEND_DEFAULT = false;
/** 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. */
public static final boolean FAIL_ONCE_DEFAULT = true;
/** 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. */
public static final String USERNAME_DEFAULT = "guest";
/** 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. */
public static final String PASSWORD_DEFAULT = "guest";
/** Holds the name of the proeprty to get the. */
public static final String SELECTOR_PROPNAME = "selector";
/** Holds the default message selector. */
public static final String SELECTOR_DEFAULT = "";
/** 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. */
public static final int DESTINATION_COUNT_DEFAULT = 1;
/** 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. */
public static final int NUM_CONSUMERS_DEFAULT = 1;
/** 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. */
public static final long TIMEOUT_DEFAULT = 30000;
/** 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. */
public static final int TX_BATCH_SIZE_DEFAULT = 1;
/** 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. */
public static final boolean UNIQUE_DESTS_DEFAULT = true;
/** Holds the name of the property to get the durable destinations flag from. */
public static final String DURABLE_DESTS_PROPNAME = "durableDests";
/** Defines the default value of the durable destinations flag. */
public static final boolean DURABLE_DESTS_DEFAULT = false;
/** 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. */
public static final int ACK_MODE_DEFAULT = Session.AUTO_ACKNOWLEDGE;
/** Holds the name of the property to get the consumers message acknowledgement mode from. */
public static final String CONSUMER_ACK_MODE_PROPNAME = "consAckMode";
/** Defines the default consumers message acknowledgement mode. */
public static final int CONSUMER_ACK_MODE_DEFAULT = Session.AUTO_ACKNOWLEDGE;
/** Holds the name of the property to get the maximum pending message size setting from. */
public static final String MAX_PENDING_PROPNAME = "maxPending";
/** Defines the default value for the maximum pending message size setting. 0 means no limit. */
public static final int MAX_PENDING_DEFAULT = 0;
/** 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. */
public static final boolean NO_LOCAL_DEFAULT = false;
/** 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";
/** Holds the name of the property to get the number of message to prefill the broker with before starting the main test. */
public static final String PREFILL_PROPNAME = "preFill";
/** Defines the default value for the number of messages to prefill. 0,default, no messages. */
public static final int PREFILL_DEFAULT = 0;
/** Holds the name of the property to get the delay to wait in ms before starting the main test after having prefilled. */
public static final String DELAY_BEFORE_CONSUME_PROPNAME = "delayBeforeConsume";
/** Defines the default value for delay in ms to wait before starting thet test run. 0,default, no delay. */
public static final long DELAY_BEFORE_CONSUME = 0;
/** Holds the name of the property to get when no messasges should be sent. */
public static final String CONSUME_ONLY_PROPNAME = "consumeOnly";
/** Defines the default value of the consumeOnly flag to use when publishing messages is not desired. */
public static final boolean CONSUME_ONLY_DEFAULT = false;
/** Holds the name of the property to get when no messasges should be sent. */
public static final String SEND_ONLY_PROPNAME = "sendOnly";
/** Defines the default value of the consumeOnly flag to use when publishing messages is not desired. */
public static final boolean SEND_ONLY_DEFAULT = false;
/** Holds the default configuration properties. */
public static ParsedProperties defaults = new ParsedProperties();
static
{
defaults.setPropertyIfNull(OVERRIDE_CLIENT_ID_PROPNAME, OVERRIDE_CLIENT_ID_DEAFULT);
defaults.setPropertyIfNull(FILE_PROPERTIES_PROPNAME, FILE_PROPERTIES_DEAFULT);
defaults.setPropertyIfNull(FACTORY_NAME_PROPNAME, FACTORY_NAME_DEAFULT);
defaults.setPropertyIfNull(BROKER_PROPNAME, BROKER_DEFAULT);
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(QUEUE_NAME_POSTFIX_PROPNAME, QUEUE_NAME_POSTFIX_DEFAULT);
defaults.setPropertyIfNull(SELECTOR_PROPNAME, SELECTOR_DEFAULT);
defaults.setPropertyIfNull(TRANSACTED_PROPNAME, TRANSACTED_DEFAULT);
defaults.setPropertyIfNull(CONSUMER_TRANSACTED_PROPNAME, CONSUMER_TRANSACTED_DEFAULT);
defaults.setPropertyIfNull(PERSISTENT_MODE_PROPNAME, PERSISTENT_MODE_DEFAULT);
defaults.setPropertyIfNull(ACK_MODE_PROPNAME, ACK_MODE_DEFAULT);
defaults.setPropertyIfNull(CONSUMER_ACK_MODE_PROPNAME, CONSUMER_ACK_MODE_DEFAULT);
defaults.setPropertyIfNull(MESSAGE_SIZE_PROPNAME, MESSAGE_SIZE_DEAFULT);
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);
defaults.setPropertyIfNull(FAIL_AFTER_SEND_PROPNAME, FAIL_AFTER_SEND_DEFAULT);
defaults.setPropertyIfNull(FAIL_ONCE_PROPNAME, FAIL_ONCE_DEFAULT);
defaults.setPropertyIfNull(TX_BATCH_SIZE_PROPNAME, TX_BATCH_SIZE_DEFAULT);
defaults.setPropertyIfNull(DESTINATION_COUNT_PROPNAME, DESTINATION_COUNT_DEFAULT);
defaults.setPropertyIfNull(NUM_CONSUMERS_PROPNAME, NUM_CONSUMERS_DEFAULT);
defaults.setPropertyIfNull(RATE_PROPNAME, RATE_DEFAULT);
defaults.setPropertyIfNull(TIMEOUT_PROPNAME, TIMEOUT_DEFAULT);
defaults.setPropertyIfNull(MAX_PENDING_PROPNAME, MAX_PENDING_DEFAULT);
defaults.setPropertyIfNull(PREFILL_PROPNAME, PREFILL_DEFAULT);
defaults.setPropertyIfNull(DELAY_BEFORE_CONSUME_PROPNAME, DELAY_BEFORE_CONSUME);
defaults.setPropertyIfNull(CONSUME_ONLY_PROPNAME, CONSUME_ONLY_DEFAULT);
defaults.setPropertyIfNull(SEND_ONLY_PROPNAME, SEND_ONLY_DEFAULT);
}
/** Allows setting of client ID on the connection, rather than through the connection URL. */
protected boolean _overrideClientId;
/** Holds the JNDI name of the JMS connection factory. */
protected String _factoryName;
/** Holds the name of the properties file to configure JNDI with. */
protected String _fileProperties;
/** Holds the broker url. */
protected String _brokerDetails;
/** Holds the username to access the broker with. */
protected String _username;
/** Holds the password to access the broker with. */
protected String _password;
/** Holds the virtual host on the broker to run the tests through. */
protected String _virtualpath;
/** Holds the root name from which to generate test destination names. */
protected String _destinationName;
/** Holds the default queue name postfix value. */
protected String _queueNamePostfix;
/** Holds the message selector to filter the pings with. */
protected String _selector;
/** Holds the producers transactional mode flag. */
protected boolean _transacted;
/** Holds the consumers transactional mode flag. */
protected boolean _consTransacted;
/** Determines whether this producer sends persistent messages. */
protected boolean _persistent;
/** Holds the acknowledgement mode used for the producers. */
protected int _ackMode;
/** Holds the acknowledgement mode setting for the consumers. */
protected int _consAckMode;
/** 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;
/** 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. */
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. */
protected boolean _failBeforeCommit;
/** 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. */
protected boolean _failBeforeSend;
/** 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. */
protected boolean _failOnce;
/** 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. */
protected int _noOfDestinations;
/** Holds the number of consumers per destination. */
protected int _noOfConsumers;
/** Holds the maximum send rate in herz. */
protected int _rate;
/**
* Holds the size of the maximum amount of pending data that the client should buffer, sending is suspended
* if this limit is breached.
*/
protected int _maxPendingSize;
/**
* Holds the number of messages to send during the setup phase, before the clients start consuming.
*/
private Integer _preFill;
/**
* Holds the time in ms to wait after preFilling before starting thet test.
*/
private Long _delayBeforeConsume;
/**
* Holds a boolean value of wither this test should just consume, i.e. skips
* sending messages, but still expects to receive the specified number.
*/
private boolean _consumeOnly;
/**
* Holds a boolean value of wither this test should just send, i.e. skips
* consuming messages, but still creates clients just doesn't start them.
*/
private boolean _sendOnly;
/** A source for providing sequential unique correlation ids. These will be unique within the same JVM. */
private static AtomicLong _correlationIdGenerator = new AtomicLong(0L);
/** A source for providing sequential unqiue ids for instances of this class to be identifed with. */
private static AtomicInteger _instanceIdGenerator = new AtomicInteger(0);
/** Holds this instances unique id. */
private int instanceId;
/**
* 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, PerCorrelationId> perCorrelationIds =
Collections.synchronizedMap(new HashMap<String, PerCorrelationId>());
/** A convenient formatter to use when time stamping output. */
protected static final DateFormat timestampFormatter = new SimpleDateFormat("hh:mm:ss:SS");
/** Holds the connection for the message producer. */
protected Connection _connection;
/** Holds the consumer connections. */
protected Connection[] _consumerConnection;
/** 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 destination where the response messages will arrive. */
protected Destination _replyDestination;
/** Holds the set of destinations that this ping producer pings. */
protected List<Destination> _pingDestinations;
/** 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. */
protected ChainedMessageListener _chainedMessageListener = null;
/**
* This id generator is used to generate ids to append to the queue name to ensure that queues can be unique when
* creating multiple ping producers in the same JVM.
*/
protected static AtomicInteger _queueJVMSequenceID = new AtomicInteger();
/**
* This id generator is used to generates ids that are only unique within this pinger. Creating multiple pingers
* on the same JVM using this id generator will allow them to ping on the same queues.
*/
protected AtomicInteger _queueSharedID = new AtomicInteger();
/** 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. */
protected MessageProducer _producer;
/** 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. */
private static final String KILL_BROKER_PROMPT = "Kill broker now, then press Return.";
/** Holds the name for this test client to be identified to the broker with. */
private String _clientID;
/** Keeps count of the total messages sent purely for debugging purposes. */
private static AtomicInteger numSent = new AtomicInteger();
/**
* Holds a monitor which is used to synchronize sender and receivers threads, where the sender has elected
* to wait until the number of unreceived message is reduced before continuing to send. This monitor is a
* fair SynchronousQueue becuase that provides fair scheduling, to ensure that all producer threads get an
* equal chance to produce messages.
*/
static final SynchronousQueue _sendPauseMonitor = new SynchronousQueue(true);
/** Keeps a count of the number of message currently sent but not received. */
static AtomicInteger _unreceived = new AtomicInteger(0);
/**
* Creates a ping producer with the specified parameters, of which there are many. See the class level 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.
*
* @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");
instanceId = _instanceIdGenerator.getAndIncrement();
// 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.
_overrideClientId = properties.getPropertyAsBoolean(OVERRIDE_CLIENT_ID_PROPNAME);
_factoryName = properties.getProperty(FACTORY_NAME_PROPNAME);
_fileProperties = properties.getProperty(FILE_PROPERTIES_PROPNAME);
_brokerDetails = properties.getProperty(BROKER_PROPNAME);
_username = properties.getProperty(USERNAME_PROPNAME);
_password = properties.getProperty(PASSWORD_PROPNAME);
_virtualpath = properties.getProperty(VIRTUAL_HOST_PROPNAME);
_destinationName = properties.getProperty(PING_QUEUE_NAME_PROPNAME);
_queueNamePostfix = properties.getProperty(QUEUE_NAME_POSTFIX_PROPNAME);
_selector = properties.getProperty(SELECTOR_PROPNAME);
_transacted = properties.getPropertyAsBoolean(TRANSACTED_PROPNAME);
_consTransacted = properties.getPropertyAsBoolean(CONSUMER_TRANSACTED_PROPNAME);
_persistent = properties.getPropertyAsBoolean(PERSISTENT_MODE_PROPNAME);
_messageSize = properties.getPropertyAsInteger(MESSAGE_SIZE_PROPNAME);
_verbose = properties.getPropertyAsBoolean(VERBOSE_PROPNAME);
_failAfterCommit = properties.getPropertyAsBoolean(FAIL_AFTER_COMMIT_PROPNAME);
_failBeforeCommit = properties.getPropertyAsBoolean(FAIL_BEFORE_COMMIT_PROPNAME);
_failAfterSend = properties.getPropertyAsBoolean(FAIL_AFTER_SEND_PROPNAME);
_failBeforeSend = properties.getPropertyAsBoolean(FAIL_BEFORE_SEND_PROPNAME);
_failOnce = properties.getPropertyAsBoolean(FAIL_ONCE_PROPNAME);
_txBatchSize = properties.getPropertyAsInteger(TX_BATCH_SIZE_PROPNAME);
_noOfDestinations = properties.getPropertyAsInteger(DESTINATION_COUNT_PROPNAME);
_noOfConsumers = properties.getPropertyAsInteger(NUM_CONSUMERS_PROPNAME);
_rate = properties.getPropertyAsInteger(RATE_PROPNAME);
_isPubSub = properties.getPropertyAsBoolean(PUBSUB_PROPNAME);
_isUnique = properties.getPropertyAsBoolean(UNIQUE_DESTS_PROPNAME);
_isDurable = properties.getPropertyAsBoolean(DURABLE_DESTS_PROPNAME);
_ackMode = _transacted ? 0 : properties.getPropertyAsInteger(ACK_MODE_PROPNAME);
_consAckMode = _consTransacted ? 0 : properties.getPropertyAsInteger(CONSUMER_ACK_MODE_PROPNAME);
_maxPendingSize = properties.getPropertyAsInteger(MAX_PENDING_PROPNAME);
_preFill = properties.getPropertyAsInteger(PREFILL_PROPNAME);
_delayBeforeConsume = properties.getPropertyAsLong(DELAY_BEFORE_CONSUME_PROPNAME);
_consumeOnly = properties.getPropertyAsBoolean(CONSUME_ONLY_PROPNAME);
_sendOnly = properties.getPropertyAsBoolean(SEND_ONLY_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);
}
/**
* Establishes a connection to the broker and creates message consumers and producers based on the parameters
* that this ping client was created with.
*
* @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");
// 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();
_clientID = "perftest_" + instanceId;
// Create a connection to the broker.
createConnection(_clientID);
// Create transactional or non-transactional sessions, based on the command line arguments.
_producerSession = _connection.createSession(_transacted, _ackMode);
_consumerSession = new Session[_noOfConsumers];
for (int i = 0; i < _noOfConsumers; i++)
{
_consumerSession[i] = _consumerConnection[i].createSession(_consTransacted, _consAckMode);
}
// Create the destinations to send pings to and receive replies from.
if (_noOfConsumers > 0)
{
_replyDestination = _consumerSession[0].createTemporaryQueue();
}
createPingDestinations(_noOfDestinations, _selector, _destinationName, _isUnique, _isDurable);
// Create the message producer only if instructed to.
if (producer)
{
createProducer();
}
// Create the message consumer only if instructed to.
if (consumer)
{
createReplyConsumers(getReplyDestinations(), _selector);
}
}
/**
* Establishes a connection to the broker, based on the configuration parameters that this ping client was
* created with.
*
* @param clientID The clients identifier.
*
* @throws JMSException Underlying exceptions allowed to fall through.
* @throws NamingException Underlying exceptions allowed to fall through.
* @throws IOException Underlying exceptions allowed to fall through.
*/
protected void createConnection(String clientID) throws JMSException, NamingException, IOException
{
// _log.debug("protected void createConnection(String clientID = " + clientID + "): called");
// _log.debug("Creating a connection for the message producer.");
File propsFile = new File(_fileProperties);
InputStream is = new FileInputStream(propsFile);
Properties properties = new Properties();
properties.load(is);
Context context = new InitialContext(properties);
ConnectionFactory factory = (ConnectionFactory) context.lookup(_factoryName);
_connection = factory.createConnection(_username, _password);
if (_overrideClientId)
{
_connection.setClientID(clientID);
}
// _log.debug("Creating " + _noOfConsumers + " connections for the consumers.");
_consumerConnection = new Connection[_noOfConsumers];
for (int i = 0; i < _noOfConsumers; i++)
{
_consumerConnection[i] = factory.createConnection(_username, _password);
// _consumerConnection[i].setClientID(clientID);
}
}
/**
* 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.
*/
public static void main(String[] args)
{
try
{
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);
pingProducer.establishConnection(true, true);
// Start the ping producers dispatch thread running.
pingProducer._connection.start();
// Create a shutdown hook to terminate the ping-pong producer.
Runtime.getRuntime().addShutdownHook(pingProducer.getShutdownHook());
// Ensure that the ping pong producer is registered to listen for exceptions on the connection too.
pingProducer._connection.setExceptionListener(pingProducer);
// Create the ping loop thread and run it until it is terminated by the shutdown hook or exception.
Thread pingThread = new Thread(pingProducer);
pingThread.run();
pingThread.join();
}
catch (Exception e)
{
System.err.println(e.getMessage());
log.error("Top level handler caught execption.", e);
System.exit(1);
}
}
/**
* Convenience method for a short pause.
*
* @param sleepTime The time in milliseconds to pause for.
*/
public static void pause(long sleepTime)
{
if (sleepTime > 0)
{
try
{
Thread.sleep(sleepTime);
}
catch (InterruptedException ie)
{ }
}
}
/**
* 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.
*/
public List<Destination> getReplyDestinations()
{
// log.debug("public List<Destination> getReplyDestinations(): called");
List<Destination> replyDestinations = new ArrayList<Destination>();
replyDestinations.add(_replyDestination);
// log.debug("replyDestinations = " + replyDestinations);
return replyDestinations;
}
/**
* 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.
*/
public void createProducer() throws JMSException
{
// log.debug("public void createProducer(): called");
_producer = (MessageProducer) _producerSession.createProducer(null);
_producer.setDeliveryMode(_persistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT);
// 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 durable If the destinations are durable topics.
*
* @throws JMSException Any JMSExceptions are allowed to fall through.
*/
public void createPingDestinations(int noOfDestinations, String selector, String rootName, boolean unique,
boolean durable) throws JMSException
{
/*log.debug("public void createPingDestinations(int noOfDestinations = " + noOfDestinations + ", String selector = "
+ selector + ", String rootName = " + rootName + ", boolean unique = " + unique + ", boolean durable = "
+ durable + "): called");*/
_pingDestinations = new ArrayList<Destination>();
// Create the desired number of ping destinations and consumers for them.
// log.debug("Creating " + noOfDestinations + " destinations to ping.");
for (int i = 0; i < noOfDestinations; i++)
{
Destination 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();
}
// Check if this is a pub/sub pinger, in which case create topics.
if (_isPubSub)
{
destination = _producerSession.createTopic(rootName + id);
// log.debug("Created non-durable topic " + destination);
if (durable)
{
_producerSession.createDurableSubscriber((Topic) destination, _connection.getClientID());
}
}
// Otherwise this is a p2p pinger, in which case create queues.
else
{
destination = _producerSession.createQueue(rootName + id + _queueNamePostfix);
// log.debug("Created queue " + destination);
}
// Keep the destination.
_pingDestinations.add(destination);
}
}
/**
* 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.
*
* @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
+ ", 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));
for (Destination destination : destinations)
{
_consumer = new MessageConsumer[_noOfConsumers];
// If we don't have consumers then ensure we have created the
// destination.
if (_noOfConsumers == 0)
{
_producerSession.createConsumer(destination, selector,
NO_LOCAL_DEFAULT).close();
}
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, selector, NO_LOCAL_DEFAULT);
final int consumerNo = i;
_consumer[i].setMessageListener(new MessageListener()
{
public void onMessage(Message message)
{
onMessageWithConsumerNo(message, consumerNo);
}
});
log.debug("Set consumer " + i + " to listen to replies sent to destination: " + destination);
}
}
}
/**
* 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.
* @param consumerNo The consumer number within this test pinger instance.
*/
public void onMessageWithConsumerNo(Message message, int consumerNo)
{
// log.debug("public void onMessageWithConsumerNo(Message message, int consumerNo = " + consumerNo + "): called");
try
{
long now = System.nanoTime();
long timestamp = getTimestamp(message);
long pingTime = now - timestamp;
// NDC.push("id" + instanceId + "/cons" + consumerNo);
// Extract the messages correlation id.
String correlationID = message.getJMSCorrelationID();
// log.debug("correlationID = " + correlationID);
// int num = message.getIntProperty("MSG_NUM");
// log.info("Message " + num + " received.");
boolean isRedelivered = message.getJMSRedelivered();
// log.debug("isRedelivered = " + isRedelivered);
if (!isRedelivered)
{
// Countdown on the traffic light if there is one for the matching correlation id.
PerCorrelationId perCorrelationId = perCorrelationIds.get(correlationID);
if (perCorrelationId != null)
{
CountDownLatch trafficLight = perCorrelationId.trafficLight;
// Restart the timeout timer on every message.
perCorrelationId.timeOutStart = System.nanoTime();
// log.debug("Reply was expected, decrementing the latch for the id, " + correlationID);
// Release waiting senders if there are some and using maxPending limit.
if ((_maxPendingSize > 0))
{
// Decrement the count of sent but not yet received messages.
int unreceived = _unreceived.decrementAndGet();
int unreceivedSize =
(unreceived * ((_messageSize == 0) ? 1 : _messageSize))
/ (_isPubSub ? getConsumersPerDestination() : 1);
// log.debug("unreceived = " + unreceived);
// log.debug("unreceivedSize = " + unreceivedSize);
// synchronized (_sendPauseMonitor)
// {
if (unreceivedSize < _maxPendingSize)
{
_sendPauseMonitor.poll();
}
// }
}
// 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
// ensures that each thread will get a unique value for the remaining messages.
long trueCount;
long remainingCount;
synchronized (trafficLight)
{
trafficLight.countDown();
trueCount = trafficLight.getCount();
remainingCount = trueCount - 1;
// NDC.push("/rem" + remainingCount);
// 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.
// When running in client ack mode, an ack is done instead of a commit, on the commit batch
// size boundaries.
long commitCount = _isPubSub ? remainingCount : (remainingCount / _noOfConsumers);
// _noOfConsumers can be set to 0 on the command line but we will not get here to
// divide by 0 as this is executed by the onMessage code when a message is recevied.
// no consumers means no message reception.
// log.debug("commitCount = " + commitCount);
if ((commitCount % _txBatchSize) == 0)
{
if (_consAckMode == 2)
{
// log.debug("Doing client ack for consumer " + consumerNo + ".");
message.acknowledge();
}
else
{
// log.debug("Trying commit for consumer " + consumerNo + ".");
commitTx(_consumerSession[consumerNo]);
// log.info("Tx committed on consumer " + consumerNo);
}
}
// Forward the message and remaining count to any interested chained message listener.
if (_chainedMessageListener != null)
{
_chainedMessageListener.onMessage(message, (int) remainingCount, pingTime);
}
// Check if this is the last message, in which case release any waiting producers. This is done
// after the transaction has been committed and any listeners notified.
if (trueCount == 1)
{
trafficLight.countDown();
}
}
}
else
{
log.warn(consumerNo + " Got unexpected message with correlationId: " + correlationID);
log.warn(consumerNo + " Map contains:" + perCorrelationIds.entrySet());
}
}
else
{
log.warn("Got redelivered message, ignoring.");
}
}
catch (JMSException e)
{
log.warn("There was a JMSException: " + e.getMessage(), e);
}
finally
{
// log.debug("public void onMessageWithConsumerNo(Message message, int consumerNo): ending");
// NDC.clear();
}
}
public void setupCorrelationID(String correlationId, int expectedCount)
{
PerCorrelationId perCorrelationId = new PerCorrelationId();
// Create a count down latch to count the number of replies with. This is created before the messages are
// sent so that the replies cannot be received before the count down is created.
// One is added to this, so that the last reply becomes a special case. The special case is that the
// chained message listener must be called before this sender can be unblocked, but that decrementing the
// countdown needs to be done before the chained listener can be called.
perCorrelationId.trafficLight = new CountDownLatch(expectedCount + 1);
perCorrelationIds.put(correlationId, perCorrelationId);
}
/**
* 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.
*
* Can be augmented through a pre-fill property (PingPongProducer.PREFILL_PROPNAME) that will populate the destination
* with a set number of messages so the total pings sent and therefore expected will be PREFILL + numPings.
*
* If pre-fill is specified then the consumers will start paused to allow the prefilling to occur.
*
* @param message The message to send. If this is null, one is generated.
* @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. If we are running in noConsumer=0 so send only mode then it will return the no msgs sent.
*
* @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
{
return pingAndWaitForReply(message, numPings, 0, timeout, messageCorrelationId);
}
public int pingAndWaitForReply(Message message, int numPings, int preFill, long timeout, String messageCorrelationId)
throws JMSException, InterruptedException
{
/*log.debug("public int pingAndWaitForReply(Message message, int numPings = " + numPings + ", long timeout = "
+ timeout + ", String messageCorrelationId = " + messageCorrelationId + "): called");*/
int totalPingsRequested = numPings + preFill;
// Generate a unique correlation id to put on the messages before sending them, if one was not specified.
if (messageCorrelationId == null)
{
messageCorrelationId = Long.toString(_correlationIdGenerator.incrementAndGet());
setupCorrelationID(messageCorrelationId, getExpectedNumPings(totalPingsRequested));
}
try
{
// NDC.push("prod");
PerCorrelationId perCorrelationId = perCorrelationIds.get(messageCorrelationId);
// Set up the current time as the start time for pinging on the correlation id. This is used to determine
// timeouts.
perCorrelationId.timeOutStart = System.nanoTime();
// Send the specifed number of messages for this test
pingNoWaitForReply(message, numPings, messageCorrelationId);
boolean timedOut;
boolean allMessagesReceived;
int numReplies;
// We don't have a consumer so don't try and wait for the messages.
// this does mean that if the producerSession is !TXed then we may
// get to exit before all msgs have been received.
//
// Return the number of requested messages, this will let the test
// report a pass.
if (_noOfConsumers == 0 || _sendOnly)
{
return getExpectedNumPings(totalPingsRequested);
}
do
{
// Block the current thread until replies to all the messages are received, or it times out.
perCorrelationId.trafficLight.await(timeout, TimeUnit.MILLISECONDS);
// Work out how many replies were receieved.
numReplies = getExpectedNumPings(totalPingsRequested) - (int) perCorrelationId.trafficLight.getCount();
allMessagesReceived = numReplies == getExpectedNumPings(totalPingsRequested);
// 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);
}
while (!timedOut && !allMessagesReceived);
if ((numReplies < getExpectedNumPings(totalPingsRequested)) && _verbose)
{
log.info("Timed out (" + timeout + " ms) before all replies received on id, " + messageCorrelationId);
}
else if (_verbose)
{
log.info("Got all replies on id, " + messageCorrelationId);
}
// commitTx(_consumerSession);
// log.debug("public int pingAndWaitForReply(Message message, int numPings, long timeout): ending");
return numReplies;
}
// Ensure that the message countdown latch is always removed from the reply map. The reply map is long lived,
// so will be a memory leak if this is not done.
finally
{
// NDC.pop();
perCorrelationIds.remove(messageCorrelationId);
}
}
/**
* 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.
*
* @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
+ ", String messageCorrelationId = " + messageCorrelationId + "): called");*/
// If we are runnning a consumeOnly test then don't send any messages
if (_consumeOnly)
{
return;
}
if (message == null)
{
message = getTestMessage(getReplyDestinations().get(0), _messageSize, _persistent);
}
message.setJMSCorrelationID(messageCorrelationId);
// Set up a committed flag to detect uncommitted messages at the end of the send loop. This may occurr if the
// transaction batch size is not a factor of the number of pings. In which case an extra commit at the end is
// needed.
boolean committed = false;
// Send all of the ping messages.
for (int i = 0; i < numPings; i++)
{
// Re-timestamp the message.
// message.setLongProperty(MESSAGE_TIMESTAMP_PROPNAME, System.nanoTime());
// Send the message, passing in the message count.
committed = sendMessage(i, message);
// 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);
}*/
}
// Call commit if the send loop finished before reaching a batch size boundary so there may still be uncommitted messages.
if (!committed)
{
commitTx(_producerSession);
}
}
/**
* Sends the sepcified message, applies rate limiting and possibly commits the current transaction. The count of
* messages sent so far must be specified and is used to round robin the ping destinations (where there are more
* than one), and to determine if the transaction batch size has been reached and the sent messages should be
* committed.
*
* @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
{
try
{
NDC.push("id" + instanceId + "/prod");
// 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());
// Prompt the user to kill the broker when doing failover testing.
_failBeforeSend = waitForUserToPromptOnFailure(_failBeforeSend);
// Get the test setup for the correlation id.
String correlationID = message.getJMSCorrelationID();
PerCorrelationId perCorrelationId = perCorrelationIds.get(correlationID);
// If necessary, wait until the max pending message size comes within its limit.
if (_maxPendingSize > 0)
{
synchronized (_sendPauseMonitor)
{
// Used to keep track of the number of times that send has to wait.
int numWaits = 0;
// The maximum number of waits before the test gives up and fails. This has been chosen to correspond with
// the test timeout.
int waitLimit = (int) (TIMEOUT_DEFAULT / 10000);
while (true)
{
// Get the size estimate of sent but not yet received messages.
int unreceived = _unreceived.get();
int unreceivedSize =
(unreceived * ((_messageSize == 0) ? 1 : _messageSize))
/ (_isPubSub ? getConsumersPerDestination() : 1);
// log.debug("unreceived = " + unreceived);
// log.debug("unreceivedSize = " + unreceivedSize);
// log.debug("_maxPendingSize = " + _maxPendingSize);
if (unreceivedSize > _maxPendingSize)
{
// log.debug("unreceived size estimate over limit = " + unreceivedSize);
// Fail the test if the send has had to wait more than the maximum allowed number of times.
if (numWaits > waitLimit)
{
String errorMessage =
"Send has had to wait for the unreceivedSize (" + unreceivedSize
+ ") to come below the maxPendingSize (" + _maxPendingSize + ") more that " + waitLimit
+ " times.";
log.warn(errorMessage);
throw new RuntimeException(errorMessage);
}
// Wait on the send pause barrier for the limit to be re-established.
try
{
long start = System.nanoTime();
// _sendPauseMonitor.wait(10000);
_sendPauseMonitor.offer(new Object(), 10000, TimeUnit.MILLISECONDS);
long end = System.nanoTime();
// Count the wait only if it was for > 99% of the requested wait time.
if (((float) (end - start) / (float) (10000 * 1000000L)) > 0.99)
{
numWaits++;
}
}
catch (InterruptedException e)
{
// Restore the interrupted status
Thread.currentThread().interrupt();
throw new RuntimeException(e);
}
}
else
{
break;
}
}
}
}
// Send the message either to its round robin destination, or its default destination.
// int num = numSent.incrementAndGet();
// message.setIntProperty("MSG_NUM", num);
setTimestamp(message);
if (destination == null)
{
_producer.send(message);
}
else
{
_producer.send(destination, message);
}
// Increase the unreceived size, this may actually happen after the message is received.
// The unreceived size is incremented by the number of consumers that will get a copy of the message,
// in pub/sub mode.
if (_maxPendingSize > 0)
{
int newUnreceivedCount = _unreceived.addAndGet(_isPubSub ? getConsumersPerDestination() : 1);
// log.debug("newUnreceivedCount = " + newUnreceivedCount);
}
// Apply message rate throttling if a rate limit has been set up.
if (_rateLimiter != null)
{
_rateLimiter.throttle();
}
// Call commit every time the commit batch size is reached.
boolean committed = false;
// 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.");
committed = commitTx(_producerSession);
}
return committed;
}
finally
{
NDC.clear();
}
}
/**
* If the specified fail flag is set, this method waits for the user to cause a failure and then indicate to the
* test that the failure has occurred, before the method returns.
*
* @param failFlag The fail flag to test.
*
* @return The new value for the fail flag. If the {@link #_failOnce} flag is set, then each fail flag is only
* used once, then reset.
*/
private boolean waitForUserToPromptOnFailure(boolean failFlag)
{
if (failFlag)
{
if (_failOnce)
{
failFlag = false;
}
// log.debug("Failing Before Send");
waitForUser(KILL_BROKER_PROMPT);
}
return failFlag;
}
/**
* Implements a single iteration of the ping loop. This sends the number of pings specified by the transaction batch
* size property, and waits for replies to all of them. Any errors cause the publish flag to be cleared, which will
* terminate the pinger.
*/
public void pingLoop()
{
try
{
// Generate a sample message and time stamp it.
Message msg = getTestMessage(_replyDestination, _messageSize, _persistent);
// setTimestamp(msg);
// Send the message and wait for a reply.
pingAndWaitForReply(msg, TX_BATCH_SIZE_DEFAULT, TIMEOUT_DEFAULT, null);
}
catch (JMSException e)
{
_publish = false;
// log.debug("There was a JMSException: " + e.getMessage(), e);
}
catch (InterruptedException e)
{
_publish = false;
// log.debug("There was an interruption: " + e.getMessage(), e);
}
}
/**
* 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.
*/
public void setChainedMessageListener(ChainedMessageListener messageListener)
{
_chainedMessageListener = messageListener;
}
/** Removes any chained message listeners from this pinger. */
public void removeChainedMessageListener()
{
_chainedMessageListener = null;
}
/**
* 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.
*
* @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
{
// return TestMessageFactory.newObjectMessage(_producerSession, replyQueue, messageSize, persistent);
return TestUtils.createTestMessageOfSize(_producerSession, messageSize);
}
/**
* Sets the current time in nanoseconds as the timestamp on the message.
*
* @param msg The message to timestamp.
*
* @throws JMSException Any JMSExceptions are allowed to fall through.
*/
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());
// }
}
/**
* Extracts the nanosecond timestamp from a message.
*
* @param msg The message to extract the time stamp from.
*
* @return The timestamp in nanos.
*
* @throws JMSException Any JMSExceptions are allowed to fall through.
*/
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);
// }
}
/**
* Stops the ping loop by clearing the publish flag. The current loop will complete when it notices that this flag
* has been cleared.
*/
public void stop()
{
_publish = false;
}
/**
* Starts the producer and consumer connections.
*
* @throws JMSException Any JMSExceptions are allowed to fall through.
*/
public void start() throws JMSException
{
// log.debug("public void start(): called");
_connection.start();
// log.debug("Producer started.");
for (int i = 0; i < _noOfConsumers; i++)
{
_consumerConnection[i].start();
// log.debug("Consumer " + i + " started.");
}
}
/** Implements a ping loop that repeatedly pings until the publish flag becomes false. */
public void run()
{
// Keep running until the publish flag is cleared.
while (_publish)
{
pingLoop();
}
}
/**
* 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.
*/
public void onException(JMSException e)
{
// log.debug("public void onException(JMSException e = " + e + "): called", e);
_publish = false;
}
/**
* 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();
}
});
}
/**
* Closes all of the producer and consumer connections.
*
* @throws JMSException All JMSException are allowed to fall through.
*/
public void close() throws JMSException
{
// log.debug("public void close(): called");
try
{
if (_connection != null)
{
// log.debug("Before close producer connection.");
_connection.close();
// log.debug("Closed producer connection.");
}
for (int i = 0; i < _noOfConsumers; i++)
{
if (_consumerConnection[i] != null)
{
// log.debug("Before close consumer connection " + i + ".");
_consumerConnection[i].close();
// log.debug("Closed consumer connection " + i + ".");
}
}
}
finally
{
_connection = null;
_producerSession = null;
_consumerSession = null;
_consumerConnection = null;
_producer = null;
_consumer = null;
_pingDestinations = null;
_replyDestination = null;
}
}
/**
* 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/>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 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");
boolean committed = false;
_failAfterSend = waitForUserToPromptOnFailure(_failAfterSend);
if (session.getTransacted())
{
// log.debug("Session is transacted.");
try
{
_failBeforeCommit = waitForUserToPromptOnFailure(_failBeforeCommit);
long start = System.nanoTime();
session.commit();
committed = true;
// log.debug("Time taken to commit :" + ((System.nanoTime() - start) / 1000000f) + " ms");
_failAfterCommit = waitForUserToPromptOnFailure(_failAfterCommit);
// log.debug("Session Commited.");
}
catch (JMSException e)
{
// log.debug("JMSException on commit:" + e.getMessage(), e);
try
{
session.rollback();
// log.debug("Message rolled back.");
}
catch (JMSException jmse)
{
// log.debug("JMSE on rollback:" + jmse.getMessage(), jmse);
// Both commit and rollback failed. Throw the rollback exception.
throw jmse;
}
}
}
return committed;
}
/**
* Outputs a prompt to the console and waits for the user to press return.
*
* @param prompt The prompt to display on the console.
*/
public void waitForUser(String prompt)
{
System.out.println(prompt);
try
{
System.in.read();
}
catch (IOException e)
{
// Ignored.
}
System.out.println("Continuing.");
}
/**
* Gets the number of consumers that are listening to each destination in the test.
*
* @return int The number of consumers subscribing to each topic.
*/
public int getConsumersPerDestination()
{
return _noOfConsumers;
}
/**
* Calculates how many pings are expected to be received for the given number sent.
*
* Note : that if you have set noConsumers to 0 then this will also return 0
* in the case of PubSub testing. This is correct as without consumers there
* will be no-one to receive the sent messages so they will be unable to respond.
*
* @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)
{
// Wow, I'm freaking sorry about this return here...
return ((_failAfterSend || _failBeforeCommit) ? numpings - 1: numpings) *
(_isPubSub ? getConsumersPerDestination() : 1);
}
/**
* Defines a chained message listener interface that can be attached to this pinger. Whenever this pinger's {@link
* 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/>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
{
/**
* Notifies interested listeners about message arrival and important test stats, the number of messages
* remaining in the test, and the messages send timestamp.
*
* @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;
}
/**
* 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. */
CountDownLatch trafficLight;
/** Holds the last timestamp that the timeout was reset to. */
Long timeOutStart;
}
}