From c47834db296c0fda1bc97c662412a6310d788a5a Mon Sep 17 00:00:00 2001 From: Keith Wall Date: Thu, 5 Jul 2012 14:56:45 +0000 Subject: QPID-4110 added topic support to performance tests. Specifically: - Added support for creating and tearing down durable subscriptions - Improved IterationValueTest so that we test the format for boolean properties (such as durableSubscription) - Added test and chart definitions Applied patch from Philip Harvey git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1357650 13f79535-47bb-0310-9956-ffa450edef68 --- .../org/apache/qpid/disttest/client/Client.java | 2 +- .../disttest/controller/config/QueueConfig.java | 3 - .../qpid/disttest/jms/ClientJmsDelegate.java | 80 +++++++++++++++++----- .../aggregation/ParticipantResultAggregator.java | 35 ++++++++++ 4 files changed, 100 insertions(+), 20 deletions(-) (limited to 'qpid/java/perftests/src/main') diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java index 2478b49cfd..0d5457c992 100644 --- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java +++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java @@ -176,7 +176,7 @@ public class Client { LOGGER.debug("Tearing down test on client: " + _clientJmsDelegate.getClientName()); - _clientJmsDelegate.closeTestConnections(); + _clientJmsDelegate.tearDownTest(); } else { diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/controller/config/QueueConfig.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/controller/config/QueueConfig.java index cffc2b7c50..45a4551cbc 100644 --- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/controller/config/QueueConfig.java +++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/controller/config/QueueConfig.java @@ -38,7 +38,6 @@ public class QueueConfig public QueueConfig(String name, boolean durable, Map attributes) { - super(); this._name = name; this._durable = durable; this._attributes = attributes; @@ -49,8 +48,6 @@ public class QueueConfig return _name; } - // TODO x-qpid-capacity and x-qpid-flow-resume-capacity need to be typed as numeric but we currrently - // pass these as a string. public Map getAttributes() { return _attributes; diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/jms/ClientJmsDelegate.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/jms/ClientJmsDelegate.java index f1dd911f0b..15c4f5f70f 100644 --- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/jms/ClientJmsDelegate.java +++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/jms/ClientJmsDelegate.java @@ -35,6 +35,7 @@ import javax.jms.MessageProducer; import javax.jms.Queue; import javax.jms.Session; import javax.jms.TextMessage; +import javax.jms.Topic; import javax.naming.Context; import javax.naming.NamingException; @@ -72,6 +73,7 @@ public class ClientJmsDelegate private Map _testSessions; private Map _testProducers; private Map _testConsumers; + private Map _testSubscriptions; private Map _testMessageProviders; private final MessageProvider _defaultMessageProvider; @@ -92,6 +94,7 @@ public class ClientJmsDelegate _testSessions = new HashMap(); _testProducers = new HashMap(); _testConsumers = new HashMap(); + _testSubscriptions = new HashMap(); _testMessageProviders = new HashMap(); _defaultMessageProvider = new MessageProvider(null); } @@ -255,9 +258,31 @@ public class ClientJmsDelegate synchronized(session) { - final Destination destination = command.isTopic() ? session.createTopic(command.getDestinationName()) - : session.createQueue(command.getDestinationName()); - final MessageConsumer jmsConsumer = session.createConsumer(destination, command.getSelector()); + Destination destination; + MessageConsumer jmsConsumer; + if(command.isTopic()) + { + Topic topic = session.createTopic(command.getDestinationName()); + if(command.isDurableSubscription()) + { + String subscription = "subscription-" + command.getParticipantName() + System.currentTimeMillis(); + jmsConsumer = session.createDurableSubscriber(topic, subscription); + + _testSubscriptions.put(subscription, session); + LOGGER.debug("created durable suscription " + subscription + " to topic " + topic); + } + else + { + jmsConsumer = session.createConsumer(topic, command.getSelector()); + } + + destination = topic; + } + else + { + destination = session.createQueue(command.getDestinationName()); + jmsConsumer = session.createConsumer(destination, command.getSelector()); + } _testConsumers.put(command.getParticipantName(), jmsConsumer); } @@ -521,36 +546,59 @@ public class ClientJmsDelegate return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE).append("clientName", _clientName).toString(); } - public void closeTestConnections() + public void tearDownTest() { StringBuilder jmsErrorMessages = new StringBuilder(); - int failedCloseCounter = 0; - for (final Map.Entry entry : _testConnections.entrySet()) + int failureCounter = 0; + + for(String subscription : _testSubscriptions.keySet()) { - final Connection connection = entry.getValue(); + Session session = _testSubscriptions.get(subscription); + try + { + session.unsubscribe(subscription); + } + catch (JMSException e) + { + LOGGER.error("Failed to unsubscribe '" + subscription + "' :" + e.getLocalizedMessage(), e); + failureCounter++; + appendErrorMessage(jmsErrorMessages, e); + } + } + + for (Map.Entry entry : _testConnections.entrySet()) + { + Connection connection = entry.getValue(); try { connection.close(); } - catch (final JMSException e) + catch (JMSException e) { LOGGER.error("Failed to close connection '" + entry.getKey() + "' :" + e.getLocalizedMessage(), e); - failedCloseCounter++; - if (jmsErrorMessages.length() > 0) - { - jmsErrorMessages.append('\n'); - } - jmsErrorMessages.append(e.getMessage()); + failureCounter++; + appendErrorMessage(jmsErrorMessages, e); } } + _testConnections.clear(); _testSessions.clear(); _testProducers.clear(); _testConsumers.clear(); - if (failedCloseCounter > 0) + + if (failureCounter > 0) + { + throw new DistributedTestException("Tear down test encountered " + failureCounter + " failures with the following errors: " + jmsErrorMessages.toString()); + } + } + + private void appendErrorMessage(StringBuilder errorMessages, JMSException e) + { + if (errorMessages.length() > 0) { - throw new DistributedTestException("Close failed for " + failedCloseCounter + " connection(s) with the following errors: " + jmsErrorMessages.toString()); + errorMessages.append('\n'); } + errorMessages.append(e.getMessage()); } public void closeTestConsumer(String consumerName) diff --git a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregator.java b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregator.java index dde717c71b..4dcabe6c7b 100644 --- a/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregator.java +++ b/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/results/aggregation/ParticipantResultAggregator.java @@ -25,6 +25,7 @@ import java.util.TreeSet; import org.apache.qpid.disttest.message.ConsumerParticipantResult; import org.apache.qpid.disttest.message.ParticipantResult; +import org.apache.qpid.disttest.message.ProducerParticipantResult; public class ParticipantResultAggregator { @@ -43,6 +44,9 @@ public class ParticipantResultAggregator private NavigableSet _encounteredIterationNumbers = new TreeSet(); private NavigableSet _encounteredBatchSizes = new TreeSet(); private NavigableSet _encounteredAcknowledgeMode = new TreeSet(); + private NavigableSet _encounteredDeliveryModes = new TreeSet(); + private NavigableSet _encounteredDurableSubscriptions = new TreeSet(); + private NavigableSet _encounteredTopics = new TreeSet(); private NavigableSet _encountedTestNames = new TreeSet(); private SeriesStatistics _latencyStatistics = new SeriesStatistics(); @@ -116,6 +120,17 @@ public class ParticipantResultAggregator _encounteredIterationNumbers.add(result.getIterationNumber()); _encounteredBatchSizes.add(result.getBatchSize()); _encounteredAcknowledgeMode.add(result.getAcknowledgeMode()); + if (result instanceof ProducerParticipantResult) + { + ProducerParticipantResult producerParticipantResult = (ProducerParticipantResult) result; + _encounteredDeliveryModes.add(producerParticipantResult.getDeliveryMode()); + } + else if(result instanceof ConsumerParticipantResult) + { + ConsumerParticipantResult consumerParticipantResult = (ConsumerParticipantResult)result; + _encounteredDurableSubscriptions.add(consumerParticipantResult.isDurableSubscription()); + _encounteredTopics.add(consumerParticipantResult.isTopic()); + } } private void setComputedVariableAttributes(ParticipantResult aggregatedResult) @@ -151,6 +166,26 @@ public class ParticipantResultAggregator { aggregatedResult.setAcknowledgeMode(_encounteredAcknowledgeMode.first()); } + if (aggregatedResult instanceof ProducerParticipantResult) + { + ProducerParticipantResult producerParticipantResult = (ProducerParticipantResult) aggregatedResult; + if(_encounteredDeliveryModes.size() == 1) + { + producerParticipantResult.setDeliveryMode(_encounteredDeliveryModes.first()); + } + } + if (aggregatedResult instanceof ConsumerParticipantResult) + { + ConsumerParticipantResult consumerParticipantResult = (ConsumerParticipantResult) aggregatedResult; + if(_encounteredDurableSubscriptions.size() == 1) + { + consumerParticipantResult.setDurableSubscription(_encounteredDurableSubscriptions.first()); + } + if(_encounteredTopics.size() == 1) + { + consumerParticipantResult.setTopic(_encounteredTopics.first()); + } + } } private double calculateThroughputInKiloBytesPerSecond() -- cgit v1.2.1