From 1d6bec9c7ea9f4ea71a919e59010fd0744581ac9 Mon Sep 17 00:00:00 2001 From: Martin Ritchie Date: Mon, 7 May 2007 09:40:58 +0000 Subject: Merged revisions 534897-534902,534904-535253,535255-535809 via svnmerge from https://svn.apache.org/repos/asf/incubator/qpid/branches/M2 ........ r534897 | bhupendrab | 2007-05-03 15:53:20 +0100 (Thu, 03 May 2007) | 1 line Attribute details background made same as other displays. ........ r535309 | ritchiem | 2007-05-04 17:12:59 +0100 (Fri, 04 May 2007) | 2 lines QPID-466 Changes to FieldTable along with corresponding PropertyValueTest to limit the Java client to only AMQP 0-8 compliant values. ........ r535809 | ritchiem | 2007-05-07 10:28:15 +0100 (Mon, 07 May 2007) | 5 lines QPID-466 Updated FieldTable to ensure no Decimal value is set that is larger than can be transmitted over AMQP. That is a max scale value of Byte.MAX_VALUE and value of up to Integer.MAX_VALUE. Additional tests to ensure this is the case. ........ git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@535819 13f79535-47bb-0310-9956-ffa450edef68 --- .../apache/qpid/requestreply/PingPongProducer.java | 182 ++++++++++++--------- .../org/apache/qpid/ping/PingLatencyTestPerf.java | 72 ++++---- 2 files changed, 148 insertions(+), 106 deletions(-) (limited to 'qpid/java/perftests/src') diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java b/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java index 5dec2125ee..642f3077fd 100644 --- a/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java +++ b/qpid/java/perftests/src/main/java/org/apache/qpid/requestreply/PingPongProducer.java @@ -37,6 +37,7 @@ import org.apache.log4j.Logger; import org.apache.qpid.AMQException; import org.apache.qpid.client.*; import org.apache.qpid.client.message.TestMessageFactory; +import org.apache.qpid.client.message.AMQMessage; import org.apache.qpid.exchange.ExchangeDefaults; import org.apache.qpid.framing.AMQShortString; import org.apache.qpid.jms.MessageProducer; @@ -374,7 +375,7 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis * ping producers on the same JVM. */ private static Map perCorrelationIds = - Collections.synchronizedMap(new HashMap()); + Collections.synchronizedMap(new HashMap()); /** A convenient formatter to use when time stamping output. */ protected static final DateFormat timestampFormatter = new SimpleDateFormat("hh:mm:ss:SS"); @@ -549,13 +550,13 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis * Starts a ping-pong loop running from the command line. The bounce back client {@link PingPongBouncer} also needs * to be started to bounce the pings back again. * - * @param args The command line arguments. + * @param args The command line arguments. */ public static void main(String[] args) { try { - Properties options = CommandLineParser.processCommandLine(args, new CommandLineParser(new String[][] {})); + Properties options = CommandLineParser.processCommandLine(args, new CommandLineParser(new String[][]{})); // Create a ping producer overriding its defaults with all options passed on the command line. PingPongProducer pingProducer = new PingPongProducer(options); @@ -597,7 +598,8 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis Thread.sleep(sleepTime); } catch (InterruptedException ie) - { } + { + } } } @@ -648,11 +650,11 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis * @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 + boolean durable) throws JMSException, AMQException { log.debug("public void createPingDestinations(int noOfDestinations = " + noOfDestinations + ", String selector = " - + selector + ", String rootName = " + rootName + ", boolean unique = " + unique + ", boolean durable = " - + durable + "): called"); + + selector + ", String rootName = " + rootName + ", boolean unique = " + unique + ", boolean durable = " + + durable + "): called"); _pingDestinations = new ArrayList(); @@ -688,8 +690,8 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis else { destination = - AMQTopic.createDurableTopic(new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME, rootName + id), - _clientID, (AMQConnection) _connection); + AMQTopic.createDurableTopic(new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME, rootName + id), + _clientID, (AMQConnection) _connection); log.debug("Created durable topic " + destination); } } @@ -698,11 +700,11 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis { AMQShortString destinationName = new AMQShortString(rootName + id); destination = - new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, destinationName, destinationName, false, false, - _isDurable); + 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); + ExchangeDefaults.DIRECT_EXCHANGE_NAME); log.debug("Created queue " + destination); } @@ -715,15 +717,15 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis /** * Creates consumers for the specified destinations and registers this pinger to listen to their messages. * - * @param destinations The destinations to listen to. - * @param selector A selector to filter the messages with. + * @param destinations The destinations to listen to. + * @param selector A selector to filter the messages with. * * @throws javax.jms.JMSException Any JMSExceptions are allowed to fall through. */ public void createReplyConsumers(Collection destinations, String selector) throws JMSException { log.debug("public void createReplyConsumers(Collection destinations = " + destinations - + ", String selector = " + selector + "): called"); + + ", String selector = " + selector + "): called"); log.debug("Creating " + destinations.size() + " reply consumers."); @@ -731,8 +733,8 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis { // Create a consumer for the destination and set this pinger to listen to its messages. _consumer = - _consumerSession.createConsumer(destination, PREFETCH_DEFAULT, NO_LOCAL_DEFAULT, EXCLUSIVE_DEFAULT, - selector); + _consumerSession.createConsumer(destination, PREFETCH_DEFAULT, NO_LOCAL_DEFAULT, EXCLUSIVE_DEFAULT, + selector); _consumer.setMessageListener(this); log.debug("Set this to listen to replies sent to destination: " + destination); @@ -740,8 +742,9 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis } /** - * Stores the received message in the replies map, then resets the boolean latch that a thread waiting for a correlating - * reply may be waiting on. This is only done if the reply has a correlation id that is expected in the replies map. + * Stores the received message in the replies map, then resets the boolean latch that a thread waiting for a + * correlating reply may be waiting on. This is only done if the reply has a correlation id that is expected in the + * replies map. * * @param message The received message. */ @@ -830,26 +833,26 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis } /** - * Sends the specified number of ping message and then waits for all correlating replies. If the wait times out before a - * reply arrives, then a null reply is returned from this method. This method allows the caller to specify the - * correlation id. + * Sends the specified number of ping message and then waits for all correlating replies. If the wait times out + * before a reply arrives, then a null reply is returned from this method. This method allows the caller to specify + * the correlation id. * - * @param message The message to send. 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. + * @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. + * @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 = " - + timeout + ", String messageCorrelationId = " + messageCorrelationId + "): called"); + + timeout + ", String messageCorrelationId = " + messageCorrelationId + "): called"); // Generate a unique correlation id to put on the messages before sending them, if one was not specified. if (messageCorrelationId == null) @@ -929,16 +932,16 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis /** * Sends the specified number of ping messages and does not wait for correlating replies. * - * @param message The message to send. - * @param numPings The number of pings to send. - * @param messageCorrelationId A correlation id to place on all messages sent. + * @param message The message to send. + * @param numPings The number of pings to send. + * @param messageCorrelationId A correlation id to place on all messages sent. * * @throws JMSException All underlying JMSExceptions are allowed to fall through. */ public void pingNoWaitForReply(Message message, int numPings, String messageCorrelationId) throws JMSException { log.debug("public void pingNoWaitForReply(Message message, int numPings = " + numPings - + ", String messageCorrelationId = " + messageCorrelationId + "): called"); + + ", String messageCorrelationId = " + messageCorrelationId + "): called"); if (message == null) { @@ -1040,9 +1043,9 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis } /** - * 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. + * 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() { @@ -1050,7 +1053,7 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis { // Generate a sample message and time stamp it. Message msg = getTestMessage(_replyDestination, _messageSize, _persistent); - msg.setLongProperty(MESSAGE_TIMESTAMP_PROPNAME, System.nanoTime()); + setTimestamp(msg); // Send the message and wait for a reply. pingAndWaitForReply(msg, TX_BATCH_SIZE_DEFAULT, TIMEOUT_DEFAULT, null); @@ -1068,7 +1071,8 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis } /** - * Sets a chained message listener. The message listener on this pinger, chains all its messages to the one set here. + * Sets a chained message listener. The message listener on this pinger, chains all its messages to the one set + * here. * * @param messageListener The chained message listener. */ @@ -1077,9 +1081,7 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis _chainedMessageListener = messageListener; } - /** - * Removes any chained message listeners from this pinger. - */ + /** Removes any chained message listeners from this pinger. */ public void removeChainedMessageListener() { _chainedMessageListener = null; @@ -1088,9 +1090,9 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis /** * 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 true if the message should use persistent delivery, false otherwise. + * @param replyQueue The reply-to destination for the message. + * @param messageSize The desired size of the message in bytes. + * @param persistent true if the message should use persistent delivery, false otherwise. * * @return A freshly generated test message. * @@ -1101,23 +1103,50 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis ObjectMessage msg = TestMessageFactory.newObjectMessage(_producerSession, replyQueue, messageSize, persistent); // Timestamp the message in nanoseconds. - msg.setLongProperty(MESSAGE_TIMESTAMP_PROPNAME, System.nanoTime()); + + setTimestamp(msg); return msg; } + 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()); + } + } + + 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 before it notices that this - * flag has been cleared. + * Stops the ping loop by clearing the publish flag. The current loop will complete before it notices that this flag + * has been cleared. */ public void stop() { _publish = false; } - /** - * 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. @@ -1128,8 +1157,8 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis } /** - * Callback method, implementing ExceptionListener. This should be registered to listen for exceptions on the connection, - * this clears the publish flag which in turn will halt the ping loop. + * Callback method, implementing ExceptionListener. This should be registered to listen for exceptions on the + * connection, this clears the publish flag which in turn will halt the ping loop. * * @param e The exception that triggered this callback method. */ @@ -1140,20 +1169,20 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis } /** - * Gets a shutdown hook that will cleanly shut this down when it is running the ping loop. This can be registered with - * the runtime system as a shutdown hook. + * Gets a shutdown hook that will cleanly shut this down when it is running the ping loop. This can be registered + * with the runtime system as a shutdown hook. * * @return A shutdown hook for the ping loop. */ public Thread getShutdownHook() { return new Thread(new Runnable() - { - public void run() - { - stop(); - } - }); + { + public void run() + { + stop(); + } + }); } /** @@ -1202,19 +1231,18 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis *

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. * - *

If the {@link #_failBeforeCommit} flag is set, this will prompt the user to kill the broker before the commit is - * applied. If the {@link #_failAfterCommit} flag is set, this will prompt the user to kill the broker after the commit - * is applied. These flags will only apply if using a transactional pinger. - * - * @param session The session to commit + *

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. * - * @throws javax.jms.JMSException If the commit fails and then the rollback fails. + * @param session The session to commit * * @return true if the session was committed, false if it was not. * - * @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. + * @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 { @@ -1335,12 +1363,12 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis /** * Defines a chained message listener interface that can be attached to this pinger. Whenever this pinger's {@link * PingPongProducer#onMessage} method is called, the chained listener set through the {@link - * PingPongProducer#setChainedMessageListener} method is passed the message, and the remaining expected count of messages - * with that correlation id. + * PingPongProducer#setChainedMessageListener} method is passed the message, and the remaining expected count of + * messages with that correlation id. * - *

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. + *

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 { @@ -1348,8 +1376,8 @@ public class PingPongProducer implements Runnable, MessageListener, ExceptionLis } /** - * Holds information on each correlation id. The countdown latch, the current timeout timer... More stuff to be added to - * this: read/write lock to make onMessage more concurrent as described in class header comment. + * Holds information on each correlation id. The countdown latch, the current timeout timer... More stuff to be + * added to this: read/write lock to make onMessage more concurrent as described in class header comment. */ protected static class PerCorrelationId { diff --git a/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingLatencyTestPerf.java b/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingLatencyTestPerf.java index c822964152..a6d12d7c42 100644 --- a/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingLatencyTestPerf.java +++ b/qpid/java/perftests/src/test/java/org/apache/qpid/ping/PingLatencyTestPerf.java @@ -35,6 +35,9 @@ import junit.framework.TestSuite; import org.apache.log4j.Logger; import org.apache.qpid.requestreply.PingPongProducer; +import org.apache.qpid.client.AMQSession; +import org.apache.qpid.client.message.AMQMessage; +import org.apache.qpid.framing.AMQShortString; import uk.co.thebadgerset.junit.extensions.TimingController; import uk.co.thebadgerset.junit.extensions.TimingControllerAware; @@ -48,18 +51,16 @@ import uk.co.thebadgerset.junit.extensions.util.ParsedProperties; * waiting until all expected replies are received. * *

This test does not output timings for every single ping message, as when running at high volume, writing the test - * log for a vast number of messages would slow the testing down. Instead samples ping latency occasionally. The frequency - * of ping sampling is set using the {@link #TEST_RESULTS_BATCH_SIZE_PROPNAME} property, to override the default of every - * {@link #DEFAULT_TEST_RESULTS_BATCH_SIZE}. + * log for a vast number of messages would slow the testing down. Instead samples ping latency occasionally. The + * frequency of ping sampling is set using the {@link #TEST_RESULTS_BATCH_SIZE_PROPNAME} property, to override the + * default of every {@link #DEFAULT_TEST_RESULTS_BATCH_SIZE}. * - *

The size parameter logged for each individual ping is set to the size of the batch of messages that the individual - * timed ping was taken from, rather than 1 for a single message. This is so that the total throughput (messages / time) - * can be calculated in order to examine the relationship between throughput and latency. + *

The size parameter logged for each individual ping is set to the size of the batch of messages that the + * individual timed ping was taken from, rather than 1 for a single message. This is so that the total throughput + * (messages / time) can be calculated in order to examine the relationship between throughput and latency. * - *

- *
CRC Card
Responsibilities Collaborations - *
Send many ping messages and output timings for sampled individual pings. - *
+ *

CRC Card
Responsibilities Collaborations
Send many ping + * messages and output timings for sampled individual pings.
*/ public class PingLatencyTestPerf extends PingTestPerf implements TimingControllerAware { @@ -77,9 +78,12 @@ public class PingLatencyTestPerf extends PingTestPerf implements TimingControlle /** Used to generate unique correlation ids for each test run. */ private AtomicLong corellationIdGenerator = new AtomicLong(); - /** Holds test specifics by correlation id. This consists of the expected number of messages and the timing controler. */ + /** + * Holds test specifics by correlation id. This consists of the expected number of messages and the timing + * controler. + */ private Map perCorrelationIds = - Collections.synchronizedMap(new HashMap()); + Collections.synchronizedMap(new HashMap()); /** Holds the batched results listener, that does logging on batch boundaries. */ private BatchedResultsListener batchedResultsListener = null; @@ -98,9 +102,7 @@ public class PingLatencyTestPerf extends PingTestPerf implements TimingControlle Integer.toString(DEFAULT_TEST_RESULTS_BATCH_SIZE)); } - /** - * Compile all the tests into a test suite. - */ + /** Compile all the tests into a test suite. */ public static Test suite() { // Build a new test suite @@ -133,8 +135,8 @@ public class PingLatencyTestPerf extends PingTestPerf implements TimingControlle } /** - * Sends the specified number of pings, asynchronously outputs timings on every batch boundary, and waits until - * all replies have been received or a time out occurs before exiting this method. + * Sends the specified number of pings, asynchronously outputs timings on every batch boundary, and waits until all + * replies have been received or a time out occurs before exiting this method. * * @param numPings The number of pings to send. */ @@ -169,9 +171,9 @@ public class PingLatencyTestPerf extends PingTestPerf implements TimingControlle // Generate a sample message of the specified size. Message msg = - pingClient.getTestMessage(perThreadSetup._pingClient.getReplyDestinations().get(0), - testParameters.getPropertyAsInteger(PingPongProducer.MESSAGE_SIZE_PROPNAME), - testParameters.getPropertyAsBoolean(PingPongProducer.PERSISTENT_MODE_PROPNAME)); + pingClient.getTestMessage(perThreadSetup._pingClient.getReplyDestinations().get(0), + testParameters.getPropertyAsInteger(PingPongProducer.MESSAGE_SIZE_PROPNAME), + testParameters.getPropertyAsBoolean(PingPongProducer.PERSISTENT_MODE_PROPNAME)); // Send the requested number of messages, and wait until they have all been received. long timeout = Long.parseLong(testParameters.getProperty(PingPongProducer.TIMEOUT_PROPNAME)); @@ -190,9 +192,7 @@ public class PingLatencyTestPerf extends PingTestPerf implements TimingControlle perCorrelationIds.remove(messageCorrelationId); } - /** - * Performs test fixture creation on a per thread basis. This will only be called once for each test thread. - */ + /** Performs test fixture creation on a per thread basis. This will only be called once for each test thread. */ public void threadSetUp() { _logger.debug("public void threadSetUp(): called"); @@ -228,14 +228,15 @@ public class PingLatencyTestPerf extends PingTestPerf implements TimingControlle /** * BatchedResultsListener is a {@link org.apache.qpid.requestreply.PingPongProducer.ChainedMessageListener} that can - * be attached to the pinger, in order to receive notifications about every message received and the number remaining - * to be received. Whenever the number remaining crosses a batch size boundary this results listener outputs a test - * timing for the actual number of messages received in the current batch. + * be attached to the pinger, in order to receive notifications about every message received and the number + * remaining to be received. Whenever the number remaining crosses a batch size boundary this results listener + * outputs a test timing for the actual number of messages received in the current batch. */ private class BatchedResultsListener implements PingPongProducer.ChainedMessageListener { /** The test results logging batch size. */ int _batchSize; + private boolean _strictAMQP; /** * Creates a results listener on the specified batch size. @@ -245,6 +246,7 @@ public class PingLatencyTestPerf extends PingTestPerf implements TimingControlle public BatchedResultsListener(int batchSize) { _batchSize = batchSize; + _strictAMQP = Boolean.parseBoolean(System.getProperties().getProperty(AMQSession.STRICT_AMQP, AMQSession.STRICT_AMQP_DEFAULT)); } /** @@ -278,7 +280,19 @@ public class PingLatencyTestPerf extends PingTestPerf implements TimingControlle // Extract the send time from the message and work out from the current time, what the ping latency was. // The ping producer time stamps messages in nanoseconds. - long startTime = message.getLongProperty(PingPongProducer.MESSAGE_TIMESTAMP_PROPNAME); + long startTime; + + if (_strictAMQP) + { + Long value = ((AMQMessage) message).getTimestampProperty(new AMQShortString(PingPongProducer.MESSAGE_TIMESTAMP_PROPNAME)); + + startTime = (value == null ? 0L : value); + } + else + { + startTime = message.getLongProperty(PingPongProducer.MESSAGE_TIMESTAMP_PROPNAME); + } + long now = System.nanoTime(); long pingTime = now - startTime; @@ -306,8 +320,8 @@ public class PingLatencyTestPerf extends PingTestPerf implements TimingControlle } /** - * Holds state specific to each correlation id, needed to output test results. This consists of the count of - * the total expected number of messages, and the timing controller for the thread sending those message ids. + * Holds state specific to each correlation id, needed to output test results. This consists of the count of the + * total expected number of messages, and the timing controller for the thread sending those message ids. */ private static class PerCorrelationId { -- cgit v1.2.1