diff options
Diffstat (limited to 'qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation')
2 files changed, 363 insertions, 0 deletions
diff --git a/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargePublisher.java b/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargePublisher.java new file mode 100644 index 0000000000..2fe01fc126 --- /dev/null +++ b/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargePublisher.java @@ -0,0 +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.fragmentation; + +import org.apache.qpid.AMQException; +import org.apache.qpid.framing.AMQShortString; +import org.apache.qpid.exchange.ExchangeDefaults; +import org.apache.qpid.url.URLSyntaxException; +import org.apache.qpid.client.AMQConnection; +import org.apache.qpid.client.AMQTopic; +import org.apache.qpid.client.AMQSession; +import org.apache.qpid.jms.MessageProducer; +import org.apache.qpid.jms.Session; +import org.apache.log4j.Logger; + +import javax.jms.BytesMessage; +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.MessageListener; +import java.net.InetAddress; +import java.net.UnknownHostException; + +/** + * A client that behaves as follows: + * <ul><li>Connects to a queue, whose name is specified as a cmd-line argument</li> + * <li>Creates a temporary queue</li> + * <li>Creates messages containing a property that is the name of the temporary queue</li> + * <li>Fires off a message on the original queue and waits for a response on the temporary queue</li> + * </ul> + */ +public class TestLargePublisher +{ + private static final Logger _log = Logger.getLogger(TestLargePublisher.class); + + private AMQConnection _connection; + + private AMQSession _session; + + private class CallbackHandler implements MessageListener + { + private int _expectedMessageCount; + + private int _actualMessageCount; + + private long _startTime; + + public CallbackHandler(int expectedMessageCount, long startTime) + { + _expectedMessageCount = expectedMessageCount; + _startTime = startTime; + } + + public void onMessage(Message m) + { + if (_log.isDebugEnabled()) + { + _log.debug("Message received: " + m); + } + _actualMessageCount++; + if (_actualMessageCount%1000 == 0) + { + _log.info("Received message count: " + _actualMessageCount); + } + /*if (!"henson".equals(m.toString())) + { + _log.error("AbstractJMSMessage response not correct: expected 'henson' but got " + m.toString()); + } + else + { + if (_log.isDebugEnabled()) + { + _log.debug("AbstractJMSMessage " + m + " received"); + } + else + { + _log.info("AbstractJMSMessage received"); + } + } */ + + if (_actualMessageCount == _expectedMessageCount) + { + long timeTaken = System.currentTimeMillis() - _startTime; + System.out.println("Total time taken to receive " + _expectedMessageCount+ " messages was " + + timeTaken + "ms, equivalent to " + + (_expectedMessageCount/(timeTaken/1000.0)) + " messages per second"); + } + } + } + + public TestLargePublisher(String host, int port, String clientID, + final int messageCount) throws AMQException,URLSyntaxException + { + try + { + createConnection(host, port, clientID); + + _session = (AMQSession) _connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + AMQTopic destination = new AMQTopic(_session.getDefaultTopicExchangeName(), new AMQShortString("large")); + MessageProducer producer = (MessageProducer) _session.createProducer(destination); + + _connection.start(); + //TextMessage msg = _session.createTextMessage(tempDestination.getQueueName() + "/Presented to in conjunction with Mahnah Mahnah and the Snowths"); + final long startTime = System.currentTimeMillis(); + + for (int i = 0; i < messageCount; i++) + { + BytesMessage msg = _session.createBytesMessage(); + populateMessage(msg); + producer.send(msg); + } + _log.info("Finished sending " + messageCount + " messages"); + } + catch (JMSException e) + { + e.printStackTrace(); + } + } + + private void createConnection(String host, int port, String clientID) throws AMQException , URLSyntaxException + { + _connection = new AMQConnection(host, port, "guest", "guest", + clientID, "/test"); + } + + private void populateMessage(BytesMessage msg) throws JMSException + { + int size = 1024 * 187; // 187k + byte[] data = new byte[size]; + for (int i = 0; i < data.length; i++) + { + data[i] = (byte)(i%25); + } + msg.writeBytes(data); + } + + /** + * + * @param args argument 1 if present specifies the name of the temporary queue to create. Leaving it blank + * means the server will allocate a name. + */ + public static void main(String[] args) throws URLSyntaxException + { + final String host; + final int port; + final int numMessages; + if (args.length == 0) + { + host = "localhost"; + port = 5672; + numMessages = 100; +// System.err.println("Usage: TestLargePublisher <host> <port> <number of messages>"); + } + else + { + host = args[0]; + port = Integer.parseInt(args[1]); + numMessages = Integer.parseInt(args[2]); + } + + try + { + InetAddress address = InetAddress.getLocalHost(); + String clientID = address.getHostName() + System.currentTimeMillis(); + TestLargePublisher client = new TestLargePublisher(host, port, clientID, numMessages); + } + catch (UnknownHostException e) + { + e.printStackTrace(); + } + catch (AMQException e) + { + System.err.println("Error in client: " + e); + e.printStackTrace(); + } + + //System.exit(0); + } +} diff --git a/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargeSubscriber.java b/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargeSubscriber.java new file mode 100644 index 0000000000..b0cde22349 --- /dev/null +++ b/qpid/java/client/src/old_test/java/org/apache/qpid/fragmentation/TestLargeSubscriber.java @@ -0,0 +1,167 @@ +/* + * + * 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.fragmentation; + +import org.apache.qpid.client.AMQConnection; +import org.apache.qpid.client.AMQTopic; +import org.apache.qpid.client.AMQSession; +import org.apache.qpid.jms.Session; +import org.apache.qpid.exchange.ExchangeDefaults; +import org.apache.qpid.framing.AMQShortString; +import org.apache.log4j.Logger; + +import javax.jms.*; +import java.net.InetAddress; + +public class TestLargeSubscriber +{ + private static final Logger _logger = Logger.getLogger(TestLargeSubscriber.class); + + private static MessageProducer _destinationProducer; + + private static String _destinationName; + + public static void main(String[] args) + { + _logger.info("Starting..."); + + final String host; + final int port; + final String username; + final String password; + final String virtualPath; + final int numExpectedMessages; + if (args.length == 0) + { + host = "localhost"; + port = 5672; + username = "guest"; + password = "guest"; + virtualPath = "/test"; + numExpectedMessages = 100; + } + else if (args.length == 6) + { + host = args[0]; + port = Integer.parseInt(args[1]); + username = args[2]; + password = args[3]; + virtualPath = args[4]; + numExpectedMessages = Integer.parseInt(args[5]); + } + else + { + System.out.println("Usage: host port username password virtual-path expectedMessageCount"); + System.exit(1); + throw new RuntimeException("cannot be reached"); + } + + try + { + InetAddress address = InetAddress.getLocalHost(); + AMQConnection con = new AMQConnection(host, port, username, password, + address.getHostName(), virtualPath); + final AMQSession session = (AMQSession) con.createSession(false, Session.AUTO_ACKNOWLEDGE); + + final int expectedMessageCount = numExpectedMessages; + + MessageConsumer consumer = session.createConsumer(new AMQTopic(session.getDefaultTopicExchangeName(), + new AMQShortString("large")), + 100, true, false, null); + + consumer.setMessageListener(new MessageListener() + { + private int _messageCount; + + private long _startTime = 0; + + public void onMessage(Message message) + { + validateMessage(message); + if (_messageCount++ == 0) + { + _startTime = System.currentTimeMillis(); + } + if (_logger.isInfoEnabled()) + { + _logger.info("Got message '" + message + "'"); + } + if (_messageCount == expectedMessageCount) + { + long totalTime = System.currentTimeMillis() - _startTime; + _logger.error("Total time to receive " + _messageCount + " messages was " + + totalTime + "ms. Rate is " + (_messageCount/(totalTime/1000.0))); + } + } + + private void validateMessage(Message message) + { + if (!(message instanceof BytesMessage)) + { + _logger.error("Message is not of correct type - should be BytesMessage and is " + + message.getClass()); + } + BytesMessage bm = (BytesMessage) message; + final int expectedSize = 1024 * 187; // 187k + try + { + if (bm.getBodyLength() != expectedSize) + { + _logger.error("Message is not correct length - should be " + expectedSize + " and is " + + bm.getBodyLength()); + } + } + catch (JMSException e) + { + _logger.error("Failed to validate message: " + e, e); + } + try + { + byte[] data = new byte[(int)bm.getBodyLength()]; + bm.readBytes(data); + for (int i = 0; i < data.length; i++) + { + if (data[i] != (byte)(i%25)) + { + _logger.error("byte " + i + " of message is wrong - should be " + i%25 + " but is " + + data[i]); + } + } + _logger.info("***** Validated message successfully"); + } + catch (JMSException e) + { + _logger.error("Failed to validate message: " + e, e); + } + } + }); + con.start(); + } + catch (Throwable t) + { + System.err.println("Fatal error: " + t); + t.printStackTrace(); + } + + System.out.println("Waiting..."); + } +} + |