From 56c9945fab16ae5c2df607b4705a79ba85128674 Mon Sep 17 00:00:00 2001 From: Bhupendra Bhusman Bhardwaj Date: Fri, 19 Jan 2007 17:02:11 +0000 Subject: Added class to ping itself and a junit test for it. git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@497878 13f79535-47bb-0310-9956-ffa450edef68 --- .../java/org/apache/qpid/ping/TestPingItself.java | 116 +++++++++++++++++++++ .../apache/qpid/requestreply/PingPongProducer.java | 32 ++++-- 2 files changed, 137 insertions(+), 11 deletions(-) create mode 100644 qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingItself.java (limited to 'qpid/java/perftests/src/main') diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingItself.java b/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingItself.java new file mode 100644 index 0000000000..6bb4c08e6d --- /dev/null +++ b/qpid/java/perftests/src/main/java/org/apache/qpid/ping/TestPingItself.java @@ -0,0 +1,116 @@ +/* + * + * Copyright (c) 2006 The Apache Software Foundation + * + * Licensed 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.qpid.requestreply.PingPongProducer; +import org.apache.qpid.client.AMQQueue; +import org.apache.qpid.client.AMQConnection; +import org.apache.qpid.jms.Session; +import org.apache.qpid.jms.MessageProducer; +import org.apache.log4j.Logger; + +import javax.jms.JMSException; +import javax.jms.MessageConsumer; +import javax.jms.ObjectMessage; +import javax.jms.Connection; +import javax.jms.DeliveryMode; +import javax.jms.Queue; +import java.net.InetAddress; + +/** + * This class is used to test sending and receiving messages to (pingQueue) and from a queue (replyQueue). + * The producer and consumer created by this test send and receive messages to and from the same Queue. ie. + * pingQueue and replyQueue are same. + * This class extends @see org.apache.qpid.requestreply.PingPongProducer which different ping and reply Queues + */ +public class TestPingItself extends PingPongProducer +{ + private static final Logger _logger = Logger.getLogger(TestPingItself.class); + + public TestPingItself(String brokerDetails, String username, String password, String virtualpath, String queueName, + String selector, boolean transacted, boolean persistent, int messageSize, boolean verbose) + throws Exception + { + super(brokerDetails, username, password, virtualpath, queueName, selector, transacted, persistent, messageSize, verbose); + } + + @Override + public void createConsumer(String selector) throws JMSException + { + // Create a message consumer to get the replies with and register this to be called back by it. + setReplyQueue(getPingQueue()); + MessageConsumer consumer = getConsumerSession().createConsumer(getReplyQueue(), PREFETCH, false, EXCLUSIVE, selector); + consumer.setMessageListener(this); + } + + + + /** + * Starts a ping-pong loop running from the command line. The bounce back client {@link org.apache.qpid.requestreply.PingPongBouncer} also needs + * to be started to bounce the pings back again. + * + *

The command line takes from 2 to 4 arguments: + *

+ *
brokerDetails The broker connection string. + *
virtualPath The virtual path. + *
transacted A boolean flag, telling this client whether or not to use transactions. + *
size The size of ping messages to use, in bytes. + *
+ * + * @param args The command line arguments as defined above. + */ + public static void main(String[] args) throws Exception + { + // Extract the command line. + if (args.length < 2) + { + System.err.println("Usage: TestPingPublisher [verbose (true/false)] " + + "[transacted (true/false)] [persistent (true/false)] [message size in bytes]"); + System.exit(0); + } + + String brokerDetails = args[0]; + String virtualpath = args[1]; + boolean verbose = (args.length >= 3) ? Boolean.parseBoolean(args[2]) : true; + boolean transacted = (args.length >= 4) ? Boolean.parseBoolean(args[3]) : false; + boolean persistent = (args.length >= 5) ? Boolean.parseBoolean(args[4]) : false; + int messageSize = (args.length >= 6) ? Integer.parseInt(args[5]) : DEFAULT_MESSAGE_SIZE; + + String queue = "ping_"+ System.currentTimeMillis(); + _logger.info("Queue:" + queue + ", Transacted:" + transacted + ", persistent:"+ persistent + + ",MessageSize:" + messageSize + " bytes"); + + // Create a ping producer to handle the request/wait/reply cycle. + TestPingItself pingItself = new TestPingItself(brokerDetails, "guest", "guest", virtualpath, queue, null, + transacted, persistent, messageSize, verbose); + pingItself.getConnection().start(); + + // Run a few priming pings to remove warm up time from test results. + pingItself.prime(PRIMING_LOOPS); + // Create a shutdown hook to terminate the ping-pong producer. + Runtime.getRuntime().addShutdownHook(pingItself.getShutdownHook()); + + // Ensure that the ping pong producer is registered to listen for exceptions on the connection too. + pingItself.getConnection().setExceptionListener(pingItself); + + // Create the ping loop thread and run it until it is terminated by the shutdown hook or exception. + Thread pingThread = new Thread(pingItself); + pingThread.run(); + pingThread.join(); + } +} 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 00b01f1025..031a5c5299 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 @@ -79,7 +79,7 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable, protected static final long TIMEOUT = 9000; /** Holds the name of the queue to send pings on. */ - private static final String PING_QUEUE_NAME = "ping"; + protected static final String PING_QUEUE_NAME = "ping"; /** The batch size. */ protected static final int BATCH_SIZE = 100; @@ -91,7 +91,7 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable, protected static final boolean EXCLUSIVE = false; /** The number of priming loops to run. */ - private static final int PRIMING_LOOPS = 3; + protected static final int PRIMING_LOOPS = 3; /** A source for providing sequential unique correlation ids. */ private AtomicLong idGenerator = new AtomicLong(0L); @@ -106,19 +106,19 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable, private Queue _pingQueue; /** Determines whether this producer sends persistent messages from the run method. */ - private boolean _persistent; + protected boolean _persistent; /** Holds the message size to send, from the run method. */ - private int _messageSize; + protected int _messageSize; /** Holds a map from message ids to latches on which threads wait for replies. */ private Map trafficLights = new HashMap(); /** Used to indicate that the ping loop should print out whenever it pings. */ - private boolean _verbose = false; - - private Session _consumerSession; + protected boolean _verbose = false; + protected Session _consumerSession; + /** * Creates a ping pong producer with the specified connection details and type. * @@ -151,9 +151,6 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable, createProducer(queueName, persistent); createConsumer(selector); - // Run a few priming pings to remove warm up time from test results. - prime(PRIMING_LOOPS); - _persistent = persistent; _messageSize = messageSize; _verbose = verbose; @@ -168,7 +165,8 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable, public void createProducer(String queueName, boolean persistent) throws JMSException { // Create a queue and producer to send the pings on. - _pingQueue = new AMQQueue(queueName); + if (_pingQueue == null) + _pingQueue = new AMQQueue(queueName); _producer = (MessageProducer) getProducerSession().createProducer(_pingQueue); _producer.setDisableMessageTimestamp(true); _producer.setDeliveryMode(persistent ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT); @@ -199,6 +197,11 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable, return _pingQueue; } + protected void setPingQueue(Queue queue) + { + _pingQueue = queue; + } + /** * Starts a ping-pong loop running from the command line. The bounce back client {@link org.apache.qpid.requestreply.PingPongBouncer} also needs * to be started to bounce the pings back again. @@ -235,6 +238,8 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable, persistent, messageSize, verbose); _pingProducer.getConnection().start(); + // Run a few priming pings to remove warm up time from test results. + _pingProducer.prime(PRIMING_LOOPS); // Create a shutdown hook to terminate the ping-pong producer. Runtime.getRuntime().addShutdownHook(_pingProducer.getShutdownHook()); @@ -445,6 +450,11 @@ public class PingPongProducer extends AbstractPingProducer implements Runnable, return _replyQueue; } + protected void setReplyQueue(Queue queue) + { + _replyQueue = queue; + } + /** * A connection listener that logs out any failover complete events. Could do more interesting things with this * at some point... -- cgit v1.2.1