You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2008/04/24 19:51:14 UTC
svn commit: r651325 [29/37] - in /incubator/qpid/trunk/qpid/java:
broker/src/main/java/org/apache/qpid/server/exchange/
broker/src/main/java/org/apache/qpid/server/handler/
broker/src/main/java/org/apache/qpid/server/output/
broker/src/main/java/org/ap...
Modified: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingDurableClient.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingDurableClient.java?rev=651325&r1=651324&r2=651325&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingDurableClient.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingDurableClient.java Thu Apr 24 10:49:03 2008
@@ -1,452 +1,452 @@
-/*
- *
- * 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.ping;
-
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.requestreply.PingPongProducer;
-import org.apache.qpid.util.CommandLineParser;
-
-import org.apache.qpid.junit.extensions.util.MathUtils;
-import org.apache.qpid.junit.extensions.util.ParsedProperties;
-
-import javax.jms.*;
-
-import java.io.BufferedReader;
-import java.io.IOException;
-import java.io.InputStreamReader;
-import java.util.List;
-import java.util.Properties;
-import java.util.concurrent.atomic.AtomicInteger;
-
-/**
- * PingDurableClient is a variation of the {@link PingPongProducer} ping tool. Instead of sending its pings and
- * receiving replies to them at the same time, this tool sends pings until it is signalled by some 'event' to stop
- * sending. It then waits for another signal before it re-opens a fresh connection and attempts to receive all of the
- * pings that it has succesfully sent. It is intended to be an interactive test that lets a user experiment with
- * failure conditions when using durable messaging.
- *
- * <p/>The events that can stop it from sending are input from the user on the console, failure of its connection to
- * the broker, completion of sending a specified number of messages, or expiry of a specified duration. In all cases
- * it will do its best to clean up and close the connection before opening a fresh connection to receive the pings
- * with.
- *
- * <p/>The event to re-connect and attempt to recieve the pings is input from the user on the console.
- *
- * <p/>This ping client inherits the configuration properties of its parent class ({@link PingPongProducer}) and
- * additionally accepts the following parameters:
- *
- * <p/><table><caption>Parameters</caption>
- * <tr><th> Parameter <th> Default <th> Comments
- * <tr><td> numMessages <td> 100 <td> The total number of messages to send.
- * <tr><td> numMessagesToAction <td> -1 <td> The number of messages to send before taking a custom 'action'.
- * <tr><td> duration <td> 30S <td> The length of time to ping for. (Format dDhHmMsS, for d days, h hours,
- * m minutes and s seconds).
- * </table>
- *
- * <p/>This ping client also overrides some of the defaults of its parent class, to provide a reasonable set up
- * when no parameters are specified.
- *
- * <p/><table><caption>Parameters</caption>
- * <tr><th> Parameter <th> Default <th> Comments
- * <tr><td> uniqueDests <td> false <td> Prevents destination names being timestamped.
- * <tr><td> transacted <td> true <td> Only makes sense to test with transactions.
- * <tr><td> persistent <td> true <td> Only makes sense to test persistent.
- * <tr><td> durableDests <td> true <td> Should use durable queues with persistent messages.
- * <tr><td> commitBatchSize <td> 10
- * <tr><td> rate <td> 20 <td> Total default test time is 5 seconds.
- * </table>
- *
- * <p/>When a number of messages or duration is specified, this ping client will ping until the first of those limits
- * is reached. Reaching the limit will be interpreted as the first signal to stop sending, and the ping client will
- * wait for the second signal before receiving its pings.
- *
- * <p/>This class provides a mechanism for extensions to add arbitrary actions, after a particular number of messages
- * have been sent. When the number of messages equal the value set in the 'numMessagesToAction' property is method,
- * the {@link #takeAction} method is called. By default this does nothing, but extensions of this class can provide
- * custom behaviour with alternative implementations of this method (for example taking a backup).
- *
- * <p><table id="crc"><caption>CRC Card</caption>
- * <tr><th> Responsibilities <th> Collaborations
- * <tr><td> Send and receive pings.
- * <tr><td> Accept user input to signal stop sending.
- * <tr><td> Accept user input to signal start receiving.
- * <tr><td> Provide feedback on pings sent versus pings received.
- * <tr><td> Provide extension point for arbitrary action on a particular message count.
- * </table>
- */
-public class PingDurableClient extends PingPongProducer implements ExceptionListener
-{
- private static final Logger log = Logger.getLogger(PingDurableClient.class);
-
- public static final String NUM_MESSAGES_PROPNAME = "numMessages";
- public static final String NUM_MESSAGES_DEFAULT = "100";
- public static final String DURATION_PROPNAME = "duration";
- public static final String DURATION_DEFAULT = "30S";
- public static final String NUM_MESSAGES_TO_ACTION_PROPNAME = "numMessagesToAction";
- public static final String NUM_MESSAGES_TO_ACTION_DEFAULT = "-1";
-
- /** The maximum length of time to wait whilst receiving pings before assuming that no more are coming. */
- private static final long TIME_OUT = 3000;
-
- static
- {
- defaults.setProperty(NUM_MESSAGES_PROPNAME, NUM_MESSAGES_DEFAULT);
- defaults.setProperty(DURATION_PROPNAME, DURATION_DEFAULT);
- defaults.setProperty(UNIQUE_DESTS_PROPNAME, "false");
- defaults.setProperty(TRANSACTED_PROPNAME, "true");
- defaults.setProperty(PERSISTENT_MODE_PROPNAME, "true");
- defaults.setProperty(TX_BATCH_SIZE_PROPNAME, "10");
- defaults.setProperty(RATE_PROPNAME, "20");
- defaults.setProperty(DURABLE_DESTS_PROPNAME, "true");
- defaults.setProperty(NUM_MESSAGES_TO_ACTION_PROPNAME, NUM_MESSAGES_TO_ACTION_DEFAULT);
- }
-
- /** Specifies the number of pings to send, if larger than 0. 0 means send until told to stop. */
- private int numMessages;
-
- /** Holds the number of messages to send before taking triggering the action. */
- private int numMessagesToAction;
-
- /** Sepcifies how long to ping for, if larger than 0. 0 means send until told to stop. */
- private long duration;
-
- /** Used to indciate that this application should terminate. Set by the shutdown hook. */
- private boolean terminate = false;
-
- /**
- * @throws Exception Any exceptions are allowed to fall through.
- */
- public PingDurableClient(Properties overrides) throws Exception
- {
- super(overrides);
- log.debug("public PingDurableClient(Properties overrides = " + overrides + "): called");
-
- // Extract the additional configuration parameters.
- ParsedProperties properties = new ParsedProperties(defaults);
- properties.putAll(overrides);
-
- numMessages = properties.getPropertyAsInteger(NUM_MESSAGES_PROPNAME);
- String durationSpec = properties.getProperty(DURATION_PROPNAME);
- numMessagesToAction = properties.getPropertyAsInteger(NUM_MESSAGES_TO_ACTION_PROPNAME);
-
- if (durationSpec != null)
- {
- duration = MathUtils.parseDuration(durationSpec) * 1000000;
- }
- }
-
- /**
- * Starts the ping/wait/receive process.
- *
- * @param args The command line arguments.
- */
- public static void main(String[] args)
- {
- try
- {
- // Create a ping producer overriding its defaults with all options passed on the command line.
- Properties options =
- CommandLineParser.processCommandLine(args, new CommandLineParser(new String[][] {}), System.getProperties());
- PingDurableClient pingProducer = new PingDurableClient(options);
-
- // 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.getConnection().setExceptionListener(pingProducer);
-
- // Run the test procedure.
- int sent = pingProducer.send();
- pingProducer.closeConnection();
- pingProducer.waitForUser("Press return to begin receiving the pings.");
- pingProducer.receive(sent);
-
- System.exit(0);
- }
- catch (Exception e)
- {
- System.err.println(e.getMessage());
- log.error("Top level handler caught execption.", e);
- System.exit(1);
- }
- }
-
- /**
- * Performs the main test procedure implemented by this ping client. See the class level comment for details.
- */
- protected int send() throws Exception
- {
- log.debug("public void sendWaitReceive(): called");
-
- log.debug("duration = " + duration);
- log.debug("numMessages = " + numMessages);
-
- if (duration > 0)
- {
- System.out.println("Sending for up to " + (duration / 1000000000f) + " seconds.");
- }
-
- if (_rate > 0)
- {
- System.out.println("Sending at " + _rate + " messages per second.");
- }
-
- if (numMessages > 0)
- {
- System.out.println("Sending up to " + numMessages + " messages.");
- }
-
- // Establish the connection and the message producer.
- establishConnection(true, false);
- _connection.start();
-
- Message message = getTestMessage(getReplyDestinations().get(0), _messageSize, _persistent);
-
- // Send pings until a terminating condition is received.
- boolean endCondition = false;
- int messagesSent = 0;
- int messagesCommitted = 0;
- int messagesNotCommitted = 0;
- long start = System.nanoTime();
-
- // Clear console in.
- clearConsole();
-
- while (!endCondition)
- {
- boolean committed = false;
-
- try
- {
- committed = sendMessage(messagesSent, message) && _transacted;
-
- messagesSent++;
- messagesNotCommitted++;
-
- // Keep count of the number of messsages currently committed and pending commit.
- if (committed)
- {
- log.debug("Adding " + messagesNotCommitted + " messages to the committed count.");
- messagesCommitted += messagesNotCommitted;
- messagesNotCommitted = 0;
-
- System.out.println("Commited: " + messagesCommitted);
- }
- }
- catch (JMSException e)
- {
- log.debug("Got JMSException whilst sending.");
- _publish = false;
- }
-
- // Perform the arbitrary action if the number of messages sent has reached the right number.
- if (messagesSent == numMessagesToAction)
- {
- System.out.println("At action point, Messages sent = " + messagesSent + ", Messages Committed = "
- + messagesCommitted + ", Messages not Committed = " + messagesNotCommitted);
- takeAction();
- }
-
- // Determine if the end condition has been met, based on the number of messages, time passed, errors on
- // the connection or user input.
- long now = System.nanoTime();
-
- if ((duration != 0) && ((now - start) > duration))
- {
- System.out.println("Send halted because duration expired.");
- endCondition = true;
- }
- else if ((numMessages != 0) && (messagesSent >= numMessages))
- {
- System.out.println("Send halted because # messages completed.");
- endCondition = true;
- }
- else if (System.in.available() > 0)
- {
- System.out.println("Send halted by user input.");
- endCondition = true;
-
- clearConsole();
- }
- else if (!_publish)
- {
- System.out.println("Send halted by error on the connection.");
- endCondition = true;
- }
- }
-
- log.debug("messagesSent = " + messagesSent);
- log.debug("messagesCommitted = " + messagesCommitted);
- log.debug("messagesNotCommitted = " + messagesNotCommitted);
-
- System.out.println("Messages sent: " + messagesSent + ", Messages Committed = " + messagesCommitted
- + ", Messages not Committed = " + messagesNotCommitted);
-
- return messagesSent;
- }
-
- protected void closeConnection()
- {
- // Clean up the connection.
- try
- {
- close();
- }
- catch (JMSException e)
- {
- log.debug("There was an error whilst closing the connection: " + e, e);
- System.out.println("There was an error whilst closing the connection.");
-
- // Ignore as did best could manage to clean up.
- }
- }
-
- protected void receive(int messagesSent) throws Exception
- {
- // Re-establish the connection and the message consumer.
- _queueJVMSequenceID = new AtomicInteger();
- _queueSharedID = new AtomicInteger();
-
- establishConnection(false, true);
- _consumer[0].setMessageListener(null);
- _consumerConnection[0].start();
-
- // Try to receive all of the pings that were successfully sent.
- int messagesReceived = 0;
- boolean endCondition = false;
-
- while (!endCondition)
- {
- // Message received = _consumer.receiveNoWait();
- Message received = _consumer[0].receive(TIME_OUT);
- log.debug("received = " + received);
-
- if (received != null)
- {
- messagesReceived++;
- }
-
- // Determine if the end condition has been met, based on the number of messages and time passed since last
- // receiving a message.
- if (received == null)
- {
- System.out.println("Timed out.");
- endCondition = true;
- }
- else if (messagesReceived >= messagesSent)
- {
- System.out.println("Got all messages.");
- endCondition = true;
- }
- }
-
- // Ensure messages received are committed.
- if (_consTransacted)
- {
- try
- {
- _consumerSession[0].commit();
- System.out.println("Committed for all messages received.");
- }
- catch (JMSException e)
- {
- log.debug("Error during commit: " + e, e);
- System.out.println("Error during commit.");
- try
- {
- _consumerSession[0].rollback();
- System.out.println("Rolled back on all messages received.");
- }
- catch (JMSException e2)
- {
- log.debug("Error during rollback: " + e, e);
- System.out.println("Error on roll back of all messages received.");
- }
-
- }
- }
-
- log.debug("messagesReceived = " + messagesReceived);
-
- System.out.println("Messages received: " + messagesReceived);
-
- // Clean up the connection.
- close();
- }
-
- /**
- * Clears any pending input from the console.
- */
- private void clearConsole()
- {
- try
- {
- BufferedReader bis = new BufferedReader(new InputStreamReader(System.in));
-
- // System.in.skip(System.in.available());
- while (bis.ready())
- {
- bis.readLine();
- }
- }
- catch (IOException e)
- { }
- }
-
- /**
- * Returns the ping destinations themselves as the reply destinations for this pinger to listen to. This has the
- * effect of making this pinger listen to its own pings.
- *
- * @return The ping destinations.
- */
- public List<Destination> getReplyDestinations()
- {
- return _pingDestinations;
- }
-
- /**
- * 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. This shutdown hook sets an additional terminate flag, compared with the
- * shutdown hook in {@link PingPongProducer}, because the publish flag is used to indicate that sending or receiving
- * message should stop, not that the application should termiante.
- *
- * @return A shutdown hook for the ping loop.
- */
- public Thread getShutdownHook()
- {
- return new Thread(new Runnable()
- {
- public void run()
- {
- stop();
- terminate = true;
- }
- });
- }
-
- /**
- * Performs an aribtrary action once the 'numMesagesToAction' count is reached on sending messages. This default
- * implementation does nothing.
- */
- public void takeAction()
- { }
-}
+/*
+ *
+ * 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.ping;
+
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.requestreply.PingPongProducer;
+import org.apache.qpid.util.CommandLineParser;
+
+import org.apache.qpid.junit.extensions.util.MathUtils;
+import org.apache.qpid.junit.extensions.util.ParsedProperties;
+
+import javax.jms.*;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * PingDurableClient is a variation of the {@link PingPongProducer} ping tool. Instead of sending its pings and
+ * receiving replies to them at the same time, this tool sends pings until it is signalled by some 'event' to stop
+ * sending. It then waits for another signal before it re-opens a fresh connection and attempts to receive all of the
+ * pings that it has succesfully sent. It is intended to be an interactive test that lets a user experiment with
+ * failure conditions when using durable messaging.
+ *
+ * <p/>The events that can stop it from sending are input from the user on the console, failure of its connection to
+ * the broker, completion of sending a specified number of messages, or expiry of a specified duration. In all cases
+ * it will do its best to clean up and close the connection before opening a fresh connection to receive the pings
+ * with.
+ *
+ * <p/>The event to re-connect and attempt to recieve the pings is input from the user on the console.
+ *
+ * <p/>This ping client inherits the configuration properties of its parent class ({@link PingPongProducer}) and
+ * additionally accepts the following parameters:
+ *
+ * <p/><table><caption>Parameters</caption>
+ * <tr><th> Parameter <th> Default <th> Comments
+ * <tr><td> numMessages <td> 100 <td> The total number of messages to send.
+ * <tr><td> numMessagesToAction <td> -1 <td> The number of messages to send before taking a custom 'action'.
+ * <tr><td> duration <td> 30S <td> The length of time to ping for. (Format dDhHmMsS, for d days, h hours,
+ * m minutes and s seconds).
+ * </table>
+ *
+ * <p/>This ping client also overrides some of the defaults of its parent class, to provide a reasonable set up
+ * when no parameters are specified.
+ *
+ * <p/><table><caption>Parameters</caption>
+ * <tr><th> Parameter <th> Default <th> Comments
+ * <tr><td> uniqueDests <td> false <td> Prevents destination names being timestamped.
+ * <tr><td> transacted <td> true <td> Only makes sense to test with transactions.
+ * <tr><td> persistent <td> true <td> Only makes sense to test persistent.
+ * <tr><td> durableDests <td> true <td> Should use durable queues with persistent messages.
+ * <tr><td> commitBatchSize <td> 10
+ * <tr><td> rate <td> 20 <td> Total default test time is 5 seconds.
+ * </table>
+ *
+ * <p/>When a number of messages or duration is specified, this ping client will ping until the first of those limits
+ * is reached. Reaching the limit will be interpreted as the first signal to stop sending, and the ping client will
+ * wait for the second signal before receiving its pings.
+ *
+ * <p/>This class provides a mechanism for extensions to add arbitrary actions, after a particular number of messages
+ * have been sent. When the number of messages equal the value set in the 'numMessagesToAction' property is method,
+ * the {@link #takeAction} method is called. By default this does nothing, but extensions of this class can provide
+ * custom behaviour with alternative implementations of this method (for example taking a backup).
+ *
+ * <p><table id="crc"><caption>CRC Card</caption>
+ * <tr><th> Responsibilities <th> Collaborations
+ * <tr><td> Send and receive pings.
+ * <tr><td> Accept user input to signal stop sending.
+ * <tr><td> Accept user input to signal start receiving.
+ * <tr><td> Provide feedback on pings sent versus pings received.
+ * <tr><td> Provide extension point for arbitrary action on a particular message count.
+ * </table>
+ */
+public class PingDurableClient extends PingPongProducer implements ExceptionListener
+{
+ private static final Logger log = Logger.getLogger(PingDurableClient.class);
+
+ public static final String NUM_MESSAGES_PROPNAME = "numMessages";
+ public static final String NUM_MESSAGES_DEFAULT = "100";
+ public static final String DURATION_PROPNAME = "duration";
+ public static final String DURATION_DEFAULT = "30S";
+ public static final String NUM_MESSAGES_TO_ACTION_PROPNAME = "numMessagesToAction";
+ public static final String NUM_MESSAGES_TO_ACTION_DEFAULT = "-1";
+
+ /** The maximum length of time to wait whilst receiving pings before assuming that no more are coming. */
+ private static final long TIME_OUT = 3000;
+
+ static
+ {
+ defaults.setProperty(NUM_MESSAGES_PROPNAME, NUM_MESSAGES_DEFAULT);
+ defaults.setProperty(DURATION_PROPNAME, DURATION_DEFAULT);
+ defaults.setProperty(UNIQUE_DESTS_PROPNAME, "false");
+ defaults.setProperty(TRANSACTED_PROPNAME, "true");
+ defaults.setProperty(PERSISTENT_MODE_PROPNAME, "true");
+ defaults.setProperty(TX_BATCH_SIZE_PROPNAME, "10");
+ defaults.setProperty(RATE_PROPNAME, "20");
+ defaults.setProperty(DURABLE_DESTS_PROPNAME, "true");
+ defaults.setProperty(NUM_MESSAGES_TO_ACTION_PROPNAME, NUM_MESSAGES_TO_ACTION_DEFAULT);
+ }
+
+ /** Specifies the number of pings to send, if larger than 0. 0 means send until told to stop. */
+ private int numMessages;
+
+ /** Holds the number of messages to send before taking triggering the action. */
+ private int numMessagesToAction;
+
+ /** Sepcifies how long to ping for, if larger than 0. 0 means send until told to stop. */
+ private long duration;
+
+ /** Used to indciate that this application should terminate. Set by the shutdown hook. */
+ private boolean terminate = false;
+
+ /**
+ * @throws Exception Any exceptions are allowed to fall through.
+ */
+ public PingDurableClient(Properties overrides) throws Exception
+ {
+ super(overrides);
+ log.debug("public PingDurableClient(Properties overrides = " + overrides + "): called");
+
+ // Extract the additional configuration parameters.
+ ParsedProperties properties = new ParsedProperties(defaults);
+ properties.putAll(overrides);
+
+ numMessages = properties.getPropertyAsInteger(NUM_MESSAGES_PROPNAME);
+ String durationSpec = properties.getProperty(DURATION_PROPNAME);
+ numMessagesToAction = properties.getPropertyAsInteger(NUM_MESSAGES_TO_ACTION_PROPNAME);
+
+ if (durationSpec != null)
+ {
+ duration = MathUtils.parseDuration(durationSpec) * 1000000;
+ }
+ }
+
+ /**
+ * Starts the ping/wait/receive process.
+ *
+ * @param args The command line arguments.
+ */
+ public static void main(String[] args)
+ {
+ try
+ {
+ // Create a ping producer overriding its defaults with all options passed on the command line.
+ Properties options =
+ CommandLineParser.processCommandLine(args, new CommandLineParser(new String[][] {}), System.getProperties());
+ PingDurableClient pingProducer = new PingDurableClient(options);
+
+ // 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.getConnection().setExceptionListener(pingProducer);
+
+ // Run the test procedure.
+ int sent = pingProducer.send();
+ pingProducer.closeConnection();
+ pingProducer.waitForUser("Press return to begin receiving the pings.");
+ pingProducer.receive(sent);
+
+ System.exit(0);
+ }
+ catch (Exception e)
+ {
+ System.err.println(e.getMessage());
+ log.error("Top level handler caught execption.", e);
+ System.exit(1);
+ }
+ }
+
+ /**
+ * Performs the main test procedure implemented by this ping client. See the class level comment for details.
+ */
+ protected int send() throws Exception
+ {
+ log.debug("public void sendWaitReceive(): called");
+
+ log.debug("duration = " + duration);
+ log.debug("numMessages = " + numMessages);
+
+ if (duration > 0)
+ {
+ System.out.println("Sending for up to " + (duration / 1000000000f) + " seconds.");
+ }
+
+ if (_rate > 0)
+ {
+ System.out.println("Sending at " + _rate + " messages per second.");
+ }
+
+ if (numMessages > 0)
+ {
+ System.out.println("Sending up to " + numMessages + " messages.");
+ }
+
+ // Establish the connection and the message producer.
+ establishConnection(true, false);
+ _connection.start();
+
+ Message message = getTestMessage(getReplyDestinations().get(0), _messageSize, _persistent);
+
+ // Send pings until a terminating condition is received.
+ boolean endCondition = false;
+ int messagesSent = 0;
+ int messagesCommitted = 0;
+ int messagesNotCommitted = 0;
+ long start = System.nanoTime();
+
+ // Clear console in.
+ clearConsole();
+
+ while (!endCondition)
+ {
+ boolean committed = false;
+
+ try
+ {
+ committed = sendMessage(messagesSent, message) && _transacted;
+
+ messagesSent++;
+ messagesNotCommitted++;
+
+ // Keep count of the number of messsages currently committed and pending commit.
+ if (committed)
+ {
+ log.debug("Adding " + messagesNotCommitted + " messages to the committed count.");
+ messagesCommitted += messagesNotCommitted;
+ messagesNotCommitted = 0;
+
+ System.out.println("Commited: " + messagesCommitted);
+ }
+ }
+ catch (JMSException e)
+ {
+ log.debug("Got JMSException whilst sending.");
+ _publish = false;
+ }
+
+ // Perform the arbitrary action if the number of messages sent has reached the right number.
+ if (messagesSent == numMessagesToAction)
+ {
+ System.out.println("At action point, Messages sent = " + messagesSent + ", Messages Committed = "
+ + messagesCommitted + ", Messages not Committed = " + messagesNotCommitted);
+ takeAction();
+ }
+
+ // Determine if the end condition has been met, based on the number of messages, time passed, errors on
+ // the connection or user input.
+ long now = System.nanoTime();
+
+ if ((duration != 0) && ((now - start) > duration))
+ {
+ System.out.println("Send halted because duration expired.");
+ endCondition = true;
+ }
+ else if ((numMessages != 0) && (messagesSent >= numMessages))
+ {
+ System.out.println("Send halted because # messages completed.");
+ endCondition = true;
+ }
+ else if (System.in.available() > 0)
+ {
+ System.out.println("Send halted by user input.");
+ endCondition = true;
+
+ clearConsole();
+ }
+ else if (!_publish)
+ {
+ System.out.println("Send halted by error on the connection.");
+ endCondition = true;
+ }
+ }
+
+ log.debug("messagesSent = " + messagesSent);
+ log.debug("messagesCommitted = " + messagesCommitted);
+ log.debug("messagesNotCommitted = " + messagesNotCommitted);
+
+ System.out.println("Messages sent: " + messagesSent + ", Messages Committed = " + messagesCommitted
+ + ", Messages not Committed = " + messagesNotCommitted);
+
+ return messagesSent;
+ }
+
+ protected void closeConnection()
+ {
+ // Clean up the connection.
+ try
+ {
+ close();
+ }
+ catch (JMSException e)
+ {
+ log.debug("There was an error whilst closing the connection: " + e, e);
+ System.out.println("There was an error whilst closing the connection.");
+
+ // Ignore as did best could manage to clean up.
+ }
+ }
+
+ protected void receive(int messagesSent) throws Exception
+ {
+ // Re-establish the connection and the message consumer.
+ _queueJVMSequenceID = new AtomicInteger();
+ _queueSharedID = new AtomicInteger();
+
+ establishConnection(false, true);
+ _consumer[0].setMessageListener(null);
+ _consumerConnection[0].start();
+
+ // Try to receive all of the pings that were successfully sent.
+ int messagesReceived = 0;
+ boolean endCondition = false;
+
+ while (!endCondition)
+ {
+ // Message received = _consumer.receiveNoWait();
+ Message received = _consumer[0].receive(TIME_OUT);
+ log.debug("received = " + received);
+
+ if (received != null)
+ {
+ messagesReceived++;
+ }
+
+ // Determine if the end condition has been met, based on the number of messages and time passed since last
+ // receiving a message.
+ if (received == null)
+ {
+ System.out.println("Timed out.");
+ endCondition = true;
+ }
+ else if (messagesReceived >= messagesSent)
+ {
+ System.out.println("Got all messages.");
+ endCondition = true;
+ }
+ }
+
+ // Ensure messages received are committed.
+ if (_consTransacted)
+ {
+ try
+ {
+ _consumerSession[0].commit();
+ System.out.println("Committed for all messages received.");
+ }
+ catch (JMSException e)
+ {
+ log.debug("Error during commit: " + e, e);
+ System.out.println("Error during commit.");
+ try
+ {
+ _consumerSession[0].rollback();
+ System.out.println("Rolled back on all messages received.");
+ }
+ catch (JMSException e2)
+ {
+ log.debug("Error during rollback: " + e, e);
+ System.out.println("Error on roll back of all messages received.");
+ }
+
+ }
+ }
+
+ log.debug("messagesReceived = " + messagesReceived);
+
+ System.out.println("Messages received: " + messagesReceived);
+
+ // Clean up the connection.
+ close();
+ }
+
+ /**
+ * Clears any pending input from the console.
+ */
+ private void clearConsole()
+ {
+ try
+ {
+ BufferedReader bis = new BufferedReader(new InputStreamReader(System.in));
+
+ // System.in.skip(System.in.available());
+ while (bis.ready())
+ {
+ bis.readLine();
+ }
+ }
+ catch (IOException e)
+ { }
+ }
+
+ /**
+ * Returns the ping destinations themselves as the reply destinations for this pinger to listen to. This has the
+ * effect of making this pinger listen to its own pings.
+ *
+ * @return The ping destinations.
+ */
+ public List<Destination> getReplyDestinations()
+ {
+ return _pingDestinations;
+ }
+
+ /**
+ * 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. This shutdown hook sets an additional terminate flag, compared with the
+ * shutdown hook in {@link PingPongProducer}, because the publish flag is used to indicate that sending or receiving
+ * message should stop, not that the application should termiante.
+ *
+ * @return A shutdown hook for the ping loop.
+ */
+ public Thread getShutdownHook()
+ {
+ return new Thread(new Runnable()
+ {
+ public void run()
+ {
+ stop();
+ terminate = true;
+ }
+ });
+ }
+
+ /**
+ * Performs an aribtrary action once the 'numMesagesToAction' count is reached on sending messages. This default
+ * implementation does nothing.
+ */
+ public void takeAction()
+ { }
+}
Propchange: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingDurableClient.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingLatencyTestPerf.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingLatencyTestPerf.java?rev=651325&r1=651324&r2=651325&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingLatencyTestPerf.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingLatencyTestPerf.java Thu Apr 24 10:49:03 2008
@@ -1,311 +1,311 @@
-/*
- *
- * 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.ping;
-
-import junit.framework.Test;
-import junit.framework.TestSuite;
-
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.client.AMQSession;
-import org.apache.qpid.requestreply.PingPongProducer;
-
-import org.apache.qpid.junit.extensions.TimingController;
-import org.apache.qpid.junit.extensions.TimingControllerAware;
-import org.apache.qpid.junit.extensions.util.ParsedProperties;
-
-import javax.jms.JMSException;
-import javax.jms.Message;
-
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.atomic.AtomicLong;
-
-/**
- * PingLatencyTestPerf is a performance test that outputs multiple timings from its test method, using the timing
- * controller interface supplied by the test runner from a seperate listener thread. It outputs round trip timings for
- * individual ping messages rather than for how long a complete batch of messages took to process. It also differs from
- * the {@link PingTestPerf} test that it extends because it can output timings as replies are received, rather than
- * waiting until all expected replies are received.
- *
- * <p/>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}.
- *
- * <p/>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.
- *
- * <p/><table id="crc"><caption>CRC Card</caption> <tr><td> Responsibilities <th> Collaborations <tr><td> Send many ping
- * messages and output timings for sampled individual pings. </table>
- */
-public class PingLatencyTestPerf extends PingTestPerf implements TimingControllerAware
-{
- private static Logger _logger = Logger.getLogger(PingLatencyTestPerf.class);
-
- /** Holds the name of the property to get the test results logging batch size. */
- public static final String TEST_RESULTS_BATCH_SIZE_PROPNAME = "batchSize";
-
- /** Holds the default test results logging batch size. */
- public static final int DEFAULT_TEST_RESULTS_BATCH_SIZE = 1000;
-
- /** Used to hold the timing controller passed from the test runner. */
- private TimingController _timingController;
-
- /** 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.
- */
- private Map<String, PerCorrelationId> perCorrelationIds =
- Collections.synchronizedMap(new HashMap<String, PerCorrelationId>());
-
- /** Holds the batched results listener, that does logging on batch boundaries. */
- private BatchedResultsListener batchedResultsListener = null;
-
- /**
- * Creates a new asynchronous ping performance test with the specified name.
- *
- * @param name The test name.
- */
- public PingLatencyTestPerf(String name)
- {
- super(name);
-
- // Sets up the test parameters with defaults.
- ParsedProperties.setSysPropertyIfNull(TEST_RESULTS_BATCH_SIZE_PROPNAME,
- Integer.toString(DEFAULT_TEST_RESULTS_BATCH_SIZE));
- }
-
- /** Compile all the tests into a test suite. */
- public static Test suite()
- {
- // Build a new test suite
- TestSuite suite = new TestSuite("Ping Latency Tests");
-
- // Run performance tests in read committed mode.
- suite.addTest(new PingLatencyTestPerf("testPingLatency"));
-
- return suite;
- }
-
- /**
- * Accepts a timing controller from the test runner.
- *
- * @param timingController The timing controller to register mutliple timings with.
- */
- public void setTimingController(TimingController timingController)
- {
- _timingController = timingController;
- }
-
- /**
- * Gets the timing controller passed in by the test runner.
- *
- * @return The timing controller passed in by the test runner.
- */
- public TimingController getTimingController()
- {
- return _timingController;
- }
-
- /**
- * 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.
- */
- public void testPingLatency(int numPings) throws Exception
- {
- _logger.debug("public void testPingLatency(int numPings): called");
-
- // Ensure that at least one ping was requeusted.
- if (numPings == 0)
- {
- _logger.error("Number of pings requested was zero.");
- }
-
- // Get the per thread test setup to run the test through.
- PerThreadSetup perThreadSetup = threadSetup.get();
- PingClient pingClient = perThreadSetup._pingClient;
-
- // Advance the correlation id of messages to send, to make it unique for this run.
- String messageCorrelationId = Long.toString(corellationIdGenerator.incrementAndGet());
- _logger.debug("messageCorrelationId = " + messageCorrelationId);
-
- // Initialize the count and timing controller for the new correlation id.
- PerCorrelationId perCorrelationId = new PerCorrelationId();
- TimingController tc = getTimingController().getControllerForCurrentThread();
- perCorrelationId._tc = tc;
- perCorrelationId._expectedCount = numPings;
- perCorrelationIds.put(messageCorrelationId, perCorrelationId);
-
- // Attach the chained message listener to the ping producer to listen asynchronously for the replies to these
- // messages.
- pingClient.setChainedMessageListener(batchedResultsListener);
-
- // 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));
-
- // Send the requested number of messages, and wait until they have all been received.
- long timeout = Long.parseLong(testParameters.getProperty(PingPongProducer.TIMEOUT_PROPNAME));
- int numReplies = pingClient.pingAndWaitForReply(msg, numPings, timeout, null);
-
- // Check that all the replies were received and log a fail if they were not.
- if (numReplies < numPings)
- {
- tc.completeTest(false, 0);
- }
-
- // Remove the chained message listener from the ping producer.
- pingClient.removeChainedMessageListener();
-
- // Remove the expected count and timing controller for the message correlation id, to ensure they are cleaned up.
- perCorrelationIds.remove(messageCorrelationId);
- }
-
- /** 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");
-
- try
- {
- // Call the set up method in the super class. This creates a PingClient pinger.
- super.threadSetUp();
-
- // Create the chained message listener, only if it has not already been created. This is set up with the
- // batch size property, to tell it what batch size to output results on. A synchronized block is used to
- // ensure that only one thread creates this.
- synchronized (this)
- {
- if (batchedResultsListener == null)
- {
- int batchSize = Integer.parseInt(testParameters.getProperty(TEST_RESULTS_BATCH_SIZE_PROPNAME));
- batchedResultsListener = new BatchedResultsListener(batchSize);
- }
- }
-
- // Get the set up that the super class created.
- PerThreadSetup perThreadSetup = threadSetup.get();
-
- // Register the chained message listener on the pinger to do its asynchronous test timings from.
- perThreadSetup._pingClient.setChainedMessageListener(batchedResultsListener);
- }
- catch (Exception e)
- {
- _logger.warn("There was an exception during per thread setup.", e);
- }
- }
-
- /**
- * 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.
- */
- 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.
- *
- * @param batchSize The batch size to use.
- */
- public BatchedResultsListener(int batchSize)
- {
- _batchSize = batchSize;
- _strictAMQP =
- Boolean.parseBoolean(System.getProperties().getProperty(AMQSession.STRICT_AMQP,
- AMQSession.STRICT_AMQP_DEFAULT));
- }
-
- /**
- * This callback method is called from all of the pingers that this test creates. It uses the correlation id
- * from the message to identify the timing controller for the test thread that was responsible for sending those
- * messages.
- *
- * @param message The message.
- * @param remainingCount The count of messages remaining to be received with a particular correlation id.
- *
- * @throws javax.jms.JMSException Any underlying JMSException is allowed to fall through.
- */
- public void onMessage(Message message, int remainingCount, long latency) throws JMSException
- {
- _logger.debug("public void onMessage(Message message, int remainingCount = " + remainingCount + "): called");
-
- // Check if a batch boundary has been crossed.
- if ((remainingCount % _batchSize) == 0)
- {
- // Extract the correlation id from the message.
- String correlationId = message.getJMSCorrelationID();
-
- // Get the details for the correlation id and check that they are not null. They can become null
- // if a test times out.
- PerCorrelationId perCorrelationId = perCorrelationIds.get(correlationId);
- if (perCorrelationId != null)
- {
- // Get the timing controller and expected count for this correlation id.
- TimingController tc = perCorrelationId._tc;
- int expected = perCorrelationId._expectedCount;
-
- // Calculate how many messages were actually received in the last batch. This will be the batch size
- // except where the number expected is not a multiple of the batch size and this is the first remaining
- // count to cross a batch size boundary, in which case it will be the number expected modulo the batch
- // size.
- int receivedInBatch = ((expected - remainingCount) < _batchSize) ? (expected % _batchSize) : _batchSize;
-
- // Register a test result for the correlation id.
- try
- {
- tc.completeTest(true, receivedInBatch, latency);
- }
- catch (InterruptedException e)
- {
- // Ignore this. It means the test runner wants to stop as soon as possible.
- _logger.warn("Got InterruptedException.", e);
- }
- }
- // Else ignore, test timed out. Should log a fail here?
- }
- }
- }
-
- /**
- * 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
- {
- public int _expectedCount;
- public TimingController _tc;
- }
-}
+/*
+ *
+ * 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.ping;
+
+import junit.framework.Test;
+import junit.framework.TestSuite;
+
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.requestreply.PingPongProducer;
+
+import org.apache.qpid.junit.extensions.TimingController;
+import org.apache.qpid.junit.extensions.TimingControllerAware;
+import org.apache.qpid.junit.extensions.util.ParsedProperties;
+
+import javax.jms.JMSException;
+import javax.jms.Message;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * PingLatencyTestPerf is a performance test that outputs multiple timings from its test method, using the timing
+ * controller interface supplied by the test runner from a seperate listener thread. It outputs round trip timings for
+ * individual ping messages rather than for how long a complete batch of messages took to process. It also differs from
+ * the {@link PingTestPerf} test that it extends because it can output timings as replies are received, rather than
+ * waiting until all expected replies are received.
+ *
+ * <p/>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}.
+ *
+ * <p/>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.
+ *
+ * <p/><table id="crc"><caption>CRC Card</caption> <tr><td> Responsibilities <th> Collaborations <tr><td> Send many ping
+ * messages and output timings for sampled individual pings. </table>
+ */
+public class PingLatencyTestPerf extends PingTestPerf implements TimingControllerAware
+{
+ private static Logger _logger = Logger.getLogger(PingLatencyTestPerf.class);
+
+ /** Holds the name of the property to get the test results logging batch size. */
+ public static final String TEST_RESULTS_BATCH_SIZE_PROPNAME = "batchSize";
+
+ /** Holds the default test results logging batch size. */
+ public static final int DEFAULT_TEST_RESULTS_BATCH_SIZE = 1000;
+
+ /** Used to hold the timing controller passed from the test runner. */
+ private TimingController _timingController;
+
+ /** 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.
+ */
+ private Map<String, PerCorrelationId> perCorrelationIds =
+ Collections.synchronizedMap(new HashMap<String, PerCorrelationId>());
+
+ /** Holds the batched results listener, that does logging on batch boundaries. */
+ private BatchedResultsListener batchedResultsListener = null;
+
+ /**
+ * Creates a new asynchronous ping performance test with the specified name.
+ *
+ * @param name The test name.
+ */
+ public PingLatencyTestPerf(String name)
+ {
+ super(name);
+
+ // Sets up the test parameters with defaults.
+ ParsedProperties.setSysPropertyIfNull(TEST_RESULTS_BATCH_SIZE_PROPNAME,
+ Integer.toString(DEFAULT_TEST_RESULTS_BATCH_SIZE));
+ }
+
+ /** Compile all the tests into a test suite. */
+ public static Test suite()
+ {
+ // Build a new test suite
+ TestSuite suite = new TestSuite("Ping Latency Tests");
+
+ // Run performance tests in read committed mode.
+ suite.addTest(new PingLatencyTestPerf("testPingLatency"));
+
+ return suite;
+ }
+
+ /**
+ * Accepts a timing controller from the test runner.
+ *
+ * @param timingController The timing controller to register mutliple timings with.
+ */
+ public void setTimingController(TimingController timingController)
+ {
+ _timingController = timingController;
+ }
+
+ /**
+ * Gets the timing controller passed in by the test runner.
+ *
+ * @return The timing controller passed in by the test runner.
+ */
+ public TimingController getTimingController()
+ {
+ return _timingController;
+ }
+
+ /**
+ * 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.
+ */
+ public void testPingLatency(int numPings) throws Exception
+ {
+ _logger.debug("public void testPingLatency(int numPings): called");
+
+ // Ensure that at least one ping was requeusted.
+ if (numPings == 0)
+ {
+ _logger.error("Number of pings requested was zero.");
+ }
+
+ // Get the per thread test setup to run the test through.
+ PerThreadSetup perThreadSetup = threadSetup.get();
+ PingClient pingClient = perThreadSetup._pingClient;
+
+ // Advance the correlation id of messages to send, to make it unique for this run.
+ String messageCorrelationId = Long.toString(corellationIdGenerator.incrementAndGet());
+ _logger.debug("messageCorrelationId = " + messageCorrelationId);
+
+ // Initialize the count and timing controller for the new correlation id.
+ PerCorrelationId perCorrelationId = new PerCorrelationId();
+ TimingController tc = getTimingController().getControllerForCurrentThread();
+ perCorrelationId._tc = tc;
+ perCorrelationId._expectedCount = numPings;
+ perCorrelationIds.put(messageCorrelationId, perCorrelationId);
+
+ // Attach the chained message listener to the ping producer to listen asynchronously for the replies to these
+ // messages.
+ pingClient.setChainedMessageListener(batchedResultsListener);
+
+ // 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));
+
+ // Send the requested number of messages, and wait until they have all been received.
+ long timeout = Long.parseLong(testParameters.getProperty(PingPongProducer.TIMEOUT_PROPNAME));
+ int numReplies = pingClient.pingAndWaitForReply(msg, numPings, timeout, null);
+
+ // Check that all the replies were received and log a fail if they were not.
+ if (numReplies < numPings)
+ {
+ tc.completeTest(false, 0);
+ }
+
+ // Remove the chained message listener from the ping producer.
+ pingClient.removeChainedMessageListener();
+
+ // Remove the expected count and timing controller for the message correlation id, to ensure they are cleaned up.
+ perCorrelationIds.remove(messageCorrelationId);
+ }
+
+ /** 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");
+
+ try
+ {
+ // Call the set up method in the super class. This creates a PingClient pinger.
+ super.threadSetUp();
+
+ // Create the chained message listener, only if it has not already been created. This is set up with the
+ // batch size property, to tell it what batch size to output results on. A synchronized block is used to
+ // ensure that only one thread creates this.
+ synchronized (this)
+ {
+ if (batchedResultsListener == null)
+ {
+ int batchSize = Integer.parseInt(testParameters.getProperty(TEST_RESULTS_BATCH_SIZE_PROPNAME));
+ batchedResultsListener = new BatchedResultsListener(batchSize);
+ }
+ }
+
+ // Get the set up that the super class created.
+ PerThreadSetup perThreadSetup = threadSetup.get();
+
+ // Register the chained message listener on the pinger to do its asynchronous test timings from.
+ perThreadSetup._pingClient.setChainedMessageListener(batchedResultsListener);
+ }
+ catch (Exception e)
+ {
+ _logger.warn("There was an exception during per thread setup.", e);
+ }
+ }
+
+ /**
+ * 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.
+ */
+ 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.
+ *
+ * @param batchSize The batch size to use.
+ */
+ public BatchedResultsListener(int batchSize)
+ {
+ _batchSize = batchSize;
+ _strictAMQP =
+ Boolean.parseBoolean(System.getProperties().getProperty(AMQSession.STRICT_AMQP,
+ AMQSession.STRICT_AMQP_DEFAULT));
+ }
+
+ /**
+ * This callback method is called from all of the pingers that this test creates. It uses the correlation id
+ * from the message to identify the timing controller for the test thread that was responsible for sending those
+ * messages.
+ *
+ * @param message The message.
+ * @param remainingCount The count of messages remaining to be received with a particular correlation id.
+ *
+ * @throws javax.jms.JMSException Any underlying JMSException is allowed to fall through.
+ */
+ public void onMessage(Message message, int remainingCount, long latency) throws JMSException
+ {
+ _logger.debug("public void onMessage(Message message, int remainingCount = " + remainingCount + "): called");
+
+ // Check if a batch boundary has been crossed.
+ if ((remainingCount % _batchSize) == 0)
+ {
+ // Extract the correlation id from the message.
+ String correlationId = message.getJMSCorrelationID();
+
+ // Get the details for the correlation id and check that they are not null. They can become null
+ // if a test times out.
+ PerCorrelationId perCorrelationId = perCorrelationIds.get(correlationId);
+ if (perCorrelationId != null)
+ {
+ // Get the timing controller and expected count for this correlation id.
+ TimingController tc = perCorrelationId._tc;
+ int expected = perCorrelationId._expectedCount;
+
+ // Calculate how many messages were actually received in the last batch. This will be the batch size
+ // except where the number expected is not a multiple of the batch size and this is the first remaining
+ // count to cross a batch size boundary, in which case it will be the number expected modulo the batch
+ // size.
+ int receivedInBatch = ((expected - remainingCount) < _batchSize) ? (expected % _batchSize) : _batchSize;
+
+ // Register a test result for the correlation id.
+ try
+ {
+ tc.completeTest(true, receivedInBatch, latency);
+ }
+ catch (InterruptedException e)
+ {
+ // Ignore this. It means the test runner wants to stop as soon as possible.
+ _logger.warn("Got InterruptedException.", e);
+ }
+ }
+ // Else ignore, test timed out. Should log a fail here?
+ }
+ }
+ }
+
+ /**
+ * 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
+ {
+ public int _expectedCount;
+ public TimingController _tc;
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingLatencyTestPerf.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingSendOnlyClient.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingSendOnlyClient.java?rev=651325&r1=651324&r2=651325&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingSendOnlyClient.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingSendOnlyClient.java Thu Apr 24 10:49:03 2008
@@ -1,93 +1,93 @@
-/*
- *
- * 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.ping;
-
-import java.util.Properties;
-
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.ObjectMessage;
-
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.client.message.TestMessageFactory;
-import org.apache.qpid.util.CommandLineParser;
-
-/**
- * <p><table id="crc"><caption>CRC Card</caption>
- * <tr><th> Responsibilities <th> Collaborations
- * </table>
- */
-public class PingSendOnlyClient extends PingDurableClient
-{
- private static final Logger log = Logger.getLogger(PingSendOnlyClient.class);
-
- public PingSendOnlyClient(Properties overrides) throws Exception
- {
- super(overrides);
- }
-
- /**
- * Starts the ping/wait/receive process.
- *
- * @param args The command line arguments.
- */
- public static void main(String[] args)
- {
- try
- {
- // Create a ping producer overriding its defaults with all options passed on the command line.
- Properties options = CommandLineParser.processCommandLine(args, new CommandLineParser(new String[][] {}), System.getProperties());
- PingSendOnlyClient pingProducer = new PingSendOnlyClient(options);
-
- // 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.getConnection().setExceptionListener(pingProducer);
-
- // Run the test procedure.
- int sent = pingProducer.send();
- pingProducer.waitForUser("Press return to close connection and quit.");
- pingProducer.closeConnection();
-
- System.exit(0);
- }
- catch (Exception e)
- {
- System.err.println(e.getMessage());
- log.error("Top level handler caught execption.", e);
- System.exit(1);
- }
- }
-
- public Message getTestMessage(Destination replyQueue, int messageSize, boolean persistent) throws JMSException
- {
- Message msg = TestMessageFactory.newTextMessage(_producerSession, messageSize);
-
- // Timestamp the message in nanoseconds.
- msg.setLongProperty(MESSAGE_TIMESTAMP_PROPNAME, System.nanoTime());
-
- return msg;
- }
-}
+/*
+ *
+ * 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.ping;
+
+import java.util.Properties;
+
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.ObjectMessage;
+
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.client.message.TestMessageFactory;
+import org.apache.qpid.util.CommandLineParser;
+
+/**
+ * <p><table id="crc"><caption>CRC Card</caption>
+ * <tr><th> Responsibilities <th> Collaborations
+ * </table>
+ */
+public class PingSendOnlyClient extends PingDurableClient
+{
+ private static final Logger log = Logger.getLogger(PingSendOnlyClient.class);
+
+ public PingSendOnlyClient(Properties overrides) throws Exception
+ {
+ super(overrides);
+ }
+
+ /**
+ * Starts the ping/wait/receive process.
+ *
+ * @param args The command line arguments.
+ */
+ public static void main(String[] args)
+ {
+ try
+ {
+ // Create a ping producer overriding its defaults with all options passed on the command line.
+ Properties options = CommandLineParser.processCommandLine(args, new CommandLineParser(new String[][] {}), System.getProperties());
+ PingSendOnlyClient pingProducer = new PingSendOnlyClient(options);
+
+ // 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.getConnection().setExceptionListener(pingProducer);
+
+ // Run the test procedure.
+ int sent = pingProducer.send();
+ pingProducer.waitForUser("Press return to close connection and quit.");
+ pingProducer.closeConnection();
+
+ System.exit(0);
+ }
+ catch (Exception e)
+ {
+ System.err.println(e.getMessage());
+ log.error("Top level handler caught execption.", e);
+ System.exit(1);
+ }
+ }
+
+ public Message getTestMessage(Destination replyQueue, int messageSize, boolean persistent) throws JMSException
+ {
+ Message msg = TestMessageFactory.newTextMessage(_producerSession, messageSize);
+
+ // Timestamp the message in nanoseconds.
+ msg.setLongProperty(MESSAGE_TIMESTAMP_PROPNAME, System.nanoTime());
+
+ return msg;
+ }
+}
Propchange: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingSendOnlyClient.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingTestPerf.java
URL: http://svn.apache.org/viewvc/incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingTestPerf.java?rev=651325&r1=651324&r2=651325&view=diff
==============================================================================
--- incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingTestPerf.java (original)
+++ incubator/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/ping/PingTestPerf.java Thu Apr 24 10:49:03 2008
@@ -1,196 +1,196 @@
-/*
- *
- * 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.ping;
-
-import junit.framework.Assert;
-import junit.framework.Test;
-import junit.framework.TestSuite;
-
-import org.apache.log4j.Logger;
-
-import org.apache.qpid.requestreply.PingPongProducer;
-
-import org.apache.qpid.junit.extensions.AsymptoticTestCase;
-import org.apache.qpid.junit.extensions.TestThreadAware;
-import org.apache.qpid.junit.extensions.util.ParsedProperties;
-import org.apache.qpid.junit.extensions.util.TestContextProperties;
-
-import javax.jms.*;
-
-/**
- * PingTestPerf is a ping test, that has been written with the intention of being scaled up to run many times
- * simultaneously to simluate many clients/producers/connections.
- *
- * <p/>A single run of the test using the default JUnit test runner will result in the sending and timing of a single
- * full round trip ping. This test may be scaled up using a suitable JUnit test runner.
- *
- * <p/>The setup/teardown cycle establishes a connection to a broker and sets up a queue to send ping messages to and a
- * temporary queue for replies. This setup is only established once for all the test repeats/threads that may be run,
- * except if the connection is lost in which case an attempt to re-establish the setup is made.
- *
- * <p/>The test cycle is: Connects to a queue, creates a temporary queue, creates messages containing a property that
- * is the name of the temporary queue, fires off a message on the original queue and waits for a response on the
- * temporary queue.
- *
- * <p/>Configurable test properties: message size, transacted or not, persistent or not. Broker connection details.
- *
- * <p><table id="crc"><caption>CRC Card</caption>
- * <tr><th> Responsibilities <th> Collaborations
- * </table>
- */
-public class PingTestPerf extends AsymptoticTestCase implements TestThreadAware
-{
- private static Logger _logger = Logger.getLogger(PingTestPerf.class);
-
- /** Thread local to hold the per-thread test setup fields. */
- ThreadLocal<PerThreadSetup> threadSetup = new ThreadLocal<PerThreadSetup>();
-
- /** Holds a property reader to extract the test parameters from. */
- protected ParsedProperties testParameters =
- TestContextProperties.getInstance(PingPongProducer.defaults /*System.getProperties()*/);
-
- public PingTestPerf(String name)
- {
- super(name);
-
- _logger.debug("testParameters = " + testParameters);
- }
-
- /**
- * Compile all the tests into a test suite.
- * @return The test method testPingOk.
- */
- public static Test suite()
- {
- // Build a new test suite
- TestSuite suite = new TestSuite("Ping Performance Tests");
-
- // Run performance tests in read committed mode.
- suite.addTest(new PingTestPerf("testPingOk"));
-
- return suite;
- }
-
- public void testPingOk(int numPings) throws Exception
- {
- if (numPings == 0)
- {
- Assert.fail("Number of pings requested was zero.");
- }
-
- // Get the per thread test setup to run the test through.
- PerThreadSetup perThreadSetup = threadSetup.get();
-
- if (perThreadSetup == null)
- {
- Assert.fail("Could not get per thread test setup, it was null.");
- }
-
- // Generate a sample message. This message is already time stamped and has its reply-to destination set.
- Message msg =
- perThreadSetup._pingClient.getTestMessage(perThreadSetup._pingClient.getReplyDestinations().get(0),
- testParameters.getPropertyAsInteger(PingPongProducer.MESSAGE_SIZE_PROPNAME),
- testParameters.getPropertyAsBoolean(PingPongProducer.PERSISTENT_MODE_PROPNAME));
-
- // start the test
- long timeout = Long.parseLong(testParameters.getProperty(PingPongProducer.TIMEOUT_PROPNAME));
- int numReplies = perThreadSetup._pingClient.pingAndWaitForReply(msg, numPings, timeout, null);
-
- // Fail the test if the timeout was exceeded.
- if (numReplies != perThreadSetup._pingClient.getExpectedNumPings(numPings))
- {
- Assert.fail("The ping timed out after " + timeout + " ms. Messages Sent = " + numPings + ", MessagesReceived = "
- + numReplies);
- }
- }
-
- /**
- * 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");
-
- try
- {
- PerThreadSetup perThreadSetup = new PerThreadSetup();
-
- // This is synchronized because there is a race condition, which causes one connection to sleep if
- // all threads try to create connection concurrently.
- synchronized (this)
- {
- // Establish a client to ping a Destination and listen the reply back from same Destination
- perThreadSetup._pingClient = new PingClient(testParameters);
- perThreadSetup._pingClient.establishConnection(true, true);
- }
- // Start the client connection
- perThreadSetup._pingClient.start();
-
- // Attach the per-thread set to the thread.
- threadSetup.set(perThreadSetup);
- }
- catch (Exception e)
- {
- _logger.warn("There was an exception during per thread setup.", e);
- }
- }
-
- /**
- * Performs test fixture clean
- */
- public void threadTearDown()
- {
- _logger.debug("public void threadTearDown(): called");
-
- try
- {
- // Get the per thread test fixture.
- PerThreadSetup perThreadSetup = threadSetup.get();
-
- // Close the pingers so that it cleans up its connection cleanly.
- synchronized (this)
- {
- if ((perThreadSetup != null) && (perThreadSetup._pingClient != null))
- {
- perThreadSetup._pingClient.close();
- }
- }
- }
- catch (JMSException e)
- {
- _logger.warn("There was an exception during per thread tear down.");
- }
- finally
- {
- // Ensure the per thread fixture is reclaimed.
- threadSetup.remove();
- }
- }
-
- protected static class PerThreadSetup
- {
- /**
- * Holds the test ping client.
- */
- protected PingClient _pingClient;
- protected String _correlationId;
- }
-}
+/*
+ *
+ * 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.ping;
+
+import junit.framework.Assert;
+import junit.framework.Test;
+import junit.framework.TestSuite;
+
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.requestreply.PingPongProducer;
+
+import org.apache.qpid.junit.extensions.AsymptoticTestCase;
+import org.apache.qpid.junit.extensions.TestThreadAware;
+import org.apache.qpid.junit.extensions.util.ParsedProperties;
+import org.apache.qpid.junit.extensions.util.TestContextProperties;
+
+import javax.jms.*;
+
+/**
+ * PingTestPerf is a ping test, that has been written with the intention of being scaled up to run many times
+ * simultaneously to simluate many clients/producers/connections.
+ *
+ * <p/>A single run of the test using the default JUnit test runner will result in the sending and timing of a single
+ * full round trip ping. This test may be scaled up using a suitable JUnit test runner.
+ *
+ * <p/>The setup/teardown cycle establishes a connection to a broker and sets up a queue to send ping messages to and a
+ * temporary queue for replies. This setup is only established once for all the test repeats/threads that may be run,
+ * except if the connection is lost in which case an attempt to re-establish the setup is made.
+ *
+ * <p/>The test cycle is: Connects to a queue, creates a temporary queue, creates messages containing a property that
+ * is the name of the temporary queue, fires off a message on the original queue and waits for a response on the
+ * temporary queue.
+ *
+ * <p/>Configurable test properties: message size, transacted or not, persistent or not. Broker connection details.
+ *
+ * <p><table id="crc"><caption>CRC Card</caption>
+ * <tr><th> Responsibilities <th> Collaborations
+ * </table>
+ */
+public class PingTestPerf extends AsymptoticTestCase implements TestThreadAware
+{
+ private static Logger _logger = Logger.getLogger(PingTestPerf.class);
+
+ /** Thread local to hold the per-thread test setup fields. */
+ ThreadLocal<PerThreadSetup> threadSetup = new ThreadLocal<PerThreadSetup>();
+
+ /** Holds a property reader to extract the test parameters from. */
+ protected ParsedProperties testParameters =
+ TestContextProperties.getInstance(PingPongProducer.defaults /*System.getProperties()*/);
+
+ public PingTestPerf(String name)
+ {
+ super(name);
+
+ _logger.debug("testParameters = " + testParameters);
+ }
+
+ /**
+ * Compile all the tests into a test suite.
+ * @return The test method testPingOk.
+ */
+ public static Test suite()
+ {
+ // Build a new test suite
+ TestSuite suite = new TestSuite("Ping Performance Tests");
+
+ // Run performance tests in read committed mode.
+ suite.addTest(new PingTestPerf("testPingOk"));
+
+ return suite;
+ }
+
+ public void testPingOk(int numPings) throws Exception
+ {
+ if (numPings == 0)
+ {
+ Assert.fail("Number of pings requested was zero.");
+ }
+
+ // Get the per thread test setup to run the test through.
+ PerThreadSetup perThreadSetup = threadSetup.get();
+
+ if (perThreadSetup == null)
+ {
+ Assert.fail("Could not get per thread test setup, it was null.");
+ }
+
+ // Generate a sample message. This message is already time stamped and has its reply-to destination set.
+ Message msg =
+ perThreadSetup._pingClient.getTestMessage(perThreadSetup._pingClient.getReplyDestinations().get(0),
+ testParameters.getPropertyAsInteger(PingPongProducer.MESSAGE_SIZE_PROPNAME),
+ testParameters.getPropertyAsBoolean(PingPongProducer.PERSISTENT_MODE_PROPNAME));
+
+ // start the test
+ long timeout = Long.parseLong(testParameters.getProperty(PingPongProducer.TIMEOUT_PROPNAME));
+ int numReplies = perThreadSetup._pingClient.pingAndWaitForReply(msg, numPings, timeout, null);
+
+ // Fail the test if the timeout was exceeded.
+ if (numReplies != perThreadSetup._pingClient.getExpectedNumPings(numPings))
+ {
+ Assert.fail("The ping timed out after " + timeout + " ms. Messages Sent = " + numPings + ", MessagesReceived = "
+ + numReplies);
+ }
+ }
+
+ /**
+ * 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");
+
+ try
+ {
+ PerThreadSetup perThreadSetup = new PerThreadSetup();
+
+ // This is synchronized because there is a race condition, which causes one connection to sleep if
+ // all threads try to create connection concurrently.
+ synchronized (this)
+ {
+ // Establish a client to ping a Destination and listen the reply back from same Destination
+ perThreadSetup._pingClient = new PingClient(testParameters);
+ perThreadSetup._pingClient.establishConnection(true, true);
+ }
+ // Start the client connection
+ perThreadSetup._pingClient.start();
+
+ // Attach the per-thread set to the thread.
+ threadSetup.set(perThreadSetup);
+ }
+ catch (Exception e)
+ {
+ _logger.warn("There was an exception during per thread setup.", e);
+ }
+ }
+
+ /**
+ * Performs test fixture clean
+ */
+ public void threadTearDown()
+ {
+ _logger.debug("public void threadTearDown(): called");
+
+ try
+ {
+ // Get the per thread test fixture.
+ PerThreadSetup perThreadSetup = threadSetup.get();
+
+ // Close the pingers so that it cleans up its connection cleanly.
+ synchronized (this)
+ {
+ if ((perThreadSetup != null) && (perThreadSetup._pingClient != null))
+ {
+ perThreadSetup._pingClient.close();
+ }
+ }
+ }
+ catch (JMSException e)
+ {
+ _logger.warn("There was an exception during per thread tear down.");
+ }
+ finally
+ {
+ // Ensure the per thread fixture is reclaimed.
+ threadSetup.remove();
+ }
+ }
+
+ protected static class PerThreadSetup
+ {
+ /**
+ * Holds the test ping client.
+ */
+ protected PingClient _pingClient;
+ protected String _correlationId;
+ }
+}