You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ro...@apache.org on 2012/04/05 18:04:16 UTC
svn commit: r1309918 [3/10] - in /qpid/trunk/qpid/java: ./ perftests/
perftests/etc/ perftests/example/ perftests/example/brokerconfig/
perftests/src/ perftests/src/main/ perftests/src/main/java/
perftests/src/main/java/org/ perftests/src/main/java/org...
Added: qpid/trunk/qpid/java/perftests/example/brokerconfig/log4j.xml
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/example/brokerconfig/log4j.xml?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/example/brokerconfig/log4j.xml (added)
+++ qpid/trunk/qpid/java/perftests/example/brokerconfig/log4j.xml Thu Apr 5 16:04:10 2012
@@ -0,0 +1,125 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ -
+ - 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.
+ -
+ --><!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
+
+<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/" debug="null" threshold="null">
+ <appender class="org.apache.log4j.QpidCompositeRollingAppender" name="ArchivingFileAppender">
+ <!-- Ensure that logs always have the DatePattern appended to the filename.
+ DEFAULT IF NOT CONFIGURED: true -->
+ <param name="StaticLogFileName" value="true"/>
+ <param name="file" value="${QPID_WORK}/log/${logprefix}qpid${logsuffix}.log"/>
+ <!-- Style of rolling to use, by:
+ File Size(1)
+ Date(2)
+ Both(3)
+
+ When Date (or Both) is enabled then the value of DatePattern will determine
+ when the new file is made. e.g. a DatePattern of "'.'yyyy-MM-dd-HH-mm"
+ which includes minutes will cause a new backup file to be made every minute.
+
+ DEFAULT IF NOT CONFIGURED: 3 -->
+ <param name="RollingStyle" value="1"/>
+ <!-- Set the count direction:
+ Negative numbers mean backups are numbered <latest>, .0, .1, .2,..., .n
+ 0 means backup is DatePattern stamped and followed with a Positive number
+ if the DatePattern stamp clashes with other existing backups.
+ Positive numbers mean backups are numbered 0, 1, 2, ..., n, <latest>
+
+ DEFAULT IF NOT CONFIGURED: -1 -->
+ <param name="CountDirection" value="0"/>
+ <!-- Maximum File Size:
+ DEFAULT IF NOT CONFIGURED: 10MB -->
+ <param name="MaxFileSize" value="1MB"/>
+ <!-- Date Pattern:
+ DEFAULT IF NOT CONFIGURED: "'.'yyyy-MM-dd" -->
+ <param name="DatePattern" value="'.'yyyy-MM-dd-HH-mm"/>
+ <!-- Maximum number of backup files:
+ 0 means no backups
+ -1 means infinite backups
+
+ DEFAULT IF NOT CONFIGURED: 0 -->
+ <param name="MaxSizeRollBackups" value="-1"/>
+ <!-- Compress(gzip) the backup files to the backup location:
+ DEFAULT IF NOT CONFIGURED: FALSE -->
+ <param name="CompressBackupFiles" value="true"/>
+ <!-- Compress the backup files using a second thread:
+ DEFAULT IF NOT CONFIGURED: FALSE -->
+ <param name="CompressAsync" value="true"/>
+ <!-- Backup Location:
+ DEFAULT IF NOT CONFIGURED: same dir as log file -->
+ <param name="backupFilesToPath" value="${QPID_WORK}/backup/log"/>
+
+ <layout class="org.apache.log4j.PatternLayout">
+ <param name="ConversionPattern" value="%d %-5p [%t] (%F:%L) - %m%n"/>
+ </layout>
+ </appender>
+
+ <appender class="org.apache.log4j.FileAppender" name="FileAppender">
+ <param name="File" value="${QPID_WORK}/log/${logprefix}qpid${logsuffix}.log"/>
+ <param name="Append" value="false"/>
+
+ <layout class="org.apache.log4j.PatternLayout">
+ <param name="ConversionPattern" value="%d %-5p [%t] (%F:%L) - %m%n"/>
+ </layout>
+ </appender>
+
+ <appender class="org.apache.log4j.ConsoleAppender" name="STDOUT">
+ <layout class="org.apache.log4j.PatternLayout">
+ <param name="ConversionPattern" value="%d %-5p [%t] (%F:%L) - %m%n"/>
+ </layout>
+ </appender>
+
+ <category additivity="true" name="org.apache.qpid.server.queue.AMQQueueMBean">
+ <priority value="info"/>
+ </category>
+
+ <!-- Provide warnings to standard output -->
+ <category additivity="true" name="org.apache.qpid">
+ <priority value="warn"/>
+ </category>
+
+ <!-- Enable info messages for the status-logging hierarchy -->
+ <logger additivity="true" name="qpid.message">
+ <level value="info"/>
+ </logger>
+
+
+
+ <!-- Examples of additional logging settings -->
+ <!-- Used to generate extra debug. See debug.log4j.xml -->
+
+ <!--<category additivity="true" name="org.apache.qpid.server.store">
+ <priority value="debug"/>
+ </category-->
+
+ <!-- Set the commons logging that the XML parser uses to WARN, it is very chatty at debug -->
+ <logger name="org.apache.commons">
+ <level value="WARN"/>
+ </logger>
+
+ <!-- Log all info events to file -->
+ <root>
+ <priority value="info"/>
+ <appender-ref ref="FileAppender"/>
+ <!--appender-ref ref="ArchivingFileAppender"/-->
+ </root>
+
+</log4j:configuration>
Added: qpid/trunk/qpid/java/perftests/example/brokerconfig/passwd
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/example/brokerconfig/passwd?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/example/brokerconfig/passwd (added)
+++ qpid/trunk/qpid/java/perftests/example/brokerconfig/passwd Thu Apr 5 16:04:10 2012
@@ -0,0 +1,23 @@
+#
+# 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.
+#
+guest:guest
+client:guest
+server:guest
+admin:admin
+
Added: qpid/trunk/qpid/java/perftests/example/brokerconfig/virtualhosts.xml
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/example/brokerconfig/virtualhosts.xml?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/example/brokerconfig/virtualhosts.xml (added)
+++ qpid/trunk/qpid/java/perftests/example/brokerconfig/virtualhosts.xml Thu Apr 5 16:04:10 2012
@@ -0,0 +1,67 @@
+<?xml version="1.0" encoding="ISO-8859-1"?>
+<!--
+ -
+ - 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.
+ -
+ -->
+<virtualhosts>
+ <default>test</default>
+
+ <virtualhost>
+ <name>test</name>
+ <test>
+ <store>
+ <!-- class>org.apache.qpid.server.store.MemoryMessageStore</class -->
+
+ <class>org.apache.qpid.server.store.berkeleydb.BDBMessageStore</class>
+ <environment-path>/home/V510279/dev/qpid/qpid/java/build/work/bdbstore/test-store</environment-path>
+ </store>
+
+ <queues>
+ <minimumAlertRepeatGap>30000</minimumAlertRepeatGap>
+ <maximumMessageCount>50</maximumMessageCount>
+ <queue>
+ <name>queue</name>
+ <queue>
+ <exchange>amq.direct</exchange>
+ <maximumQueueDepth>4235264</maximumQueueDepth>
+ <!-- 4Mb -->
+ <maximumMessageSize>2117632</maximumMessageSize>
+ <!-- 2Mb -->
+ <maximumMessageAge>600000</maximumMessageAge>
+ <!-- 10 mins -->
+ </queue>
+ </queue>
+ <queue>
+ <name>ping</name>
+ <ping>
+ <exchange>amq.direct</exchange>
+ <maximumQueueDepth>4235264</maximumQueueDepth>
+ <!-- 4Mb -->
+ <maximumMessageSize>2117632</maximumMessageSize>
+ <!-- 2Mb -->
+ <maximumMessageAge>600000</maximumMessageAge>
+ <!-- 10 mins -->
+ </ping>
+ </queue>
+ </queues>
+ </test>
+ </virtualhost>
+</virtualhosts>
+
+
Added: qpid/trunk/qpid/java/perftests/example/config-one-test.json
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/example/config-one-test.json?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/example/config-one-test.json (added)
+++ qpid/trunk/qpid/java/perftests/example/config-one-test.json Thu Apr 5 16:04:10 2012
@@ -0,0 +1,55 @@
+{
+ "_tests":[
+ {
+ "_name": "One producer to one consumer";
+ "_queues":[
+ {
+ "_name": "direct://amq.direct//testQueue"
+ }
+ ];
+ "_clients":[
+ {
+ "_name": "producingClient",
+ "_connections":[
+ {
+ "_name": "connection1",
+ "_factory": "connectionfactory",
+ "_sessions": [
+ {
+ "_sessionName": "session1",
+ "_producers": [
+ {
+ "_name": "Producer1",
+ "_destinationName": "direct://amq.direct//testQueue",
+ "_numberOfMessages": 5000
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ },
+ {
+ "_name": "consumingClient",
+ "_connections":[
+ {
+ "_name": "connection1",
+ "_factory": "connectionfactory",
+ "_sessions": [
+ {
+ "_sessionName": "session1",
+ "_consumers": [
+ {
+ "_name": "Consumer1",
+ "_destinationName": "direct://amq.direct//testQueue",
+ "_numberOfMessages": 5000
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ }]
+}
Added: qpid/trunk/qpid/java/perftests/example/config-stress-testing-manyp-c1-with-psfc.json
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/example/config-stress-testing-manyp-c1-with-psfc.json?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/example/config-stress-testing-manyp-c1-with-psfc.json (added)
+++ qpid/trunk/qpid/java/perftests/example/config-stress-testing-manyp-c1-with-psfc.json Thu Apr 5 16:04:10 2012
@@ -0,0 +1,162 @@
+{
+ "_tests":[
+ {
+ "_name": "Many producers to one consumer w/producer flow-control";
+ "_queues":[
+ {
+ "_name": "direct://amq.direct//testQueue",
+ "_attributes":
+ {
+ "comment": "does not work yet - configure in virtualhost.xml",
+ "x-qpid-capacity": 10485760,
+ "x-qpid-flow-resume-capacity": 8388608
+ }
+ }
+ ];
+ "_clients":[
+ {
+ "_name": "producingClient1",
+ "_connections":[
+ {
+ "_name": "connection1",
+ "_factory": "connectionfactory",
+ "_sessions": [
+ {
+ "_sessionName": "session1",
+ "_acknowledgeMode": 1,
+ "_producers": [
+ {
+ "_name": "Producer1",
+ "_destinationName": "direct://amq.direct//testQueue",
+ "_numberOfMessages": 1000000,
+ "_deliveryMode": 1,
+ "_messageSize": 10000
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ },
+ {
+ "_name": "producingClient2",
+ "_connections":[
+ {
+ "_name": "connection2",
+ "_factory": "connectionfactory",
+ "_sessions": [
+ {
+ "_sessionName": "session2",
+ "_acknowledgeMode": 1,
+ "_producers": [
+ {
+ "_name": "Producer2",
+ "_destinationName": "direct://amq.direct//testQueue",
+ "_numberOfMessages": 1000000,
+ "_deliveryMode": 1,
+ "_messageSize": 10000
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ },
+ {
+ "_name": "producingClient3",
+ "_connections":[
+ {
+ "_name": "connection3",
+ "_factory": "connectionfactory",
+ "_sessions": [
+ {
+ "_sessionName": "session3",
+ "_acknowledgeMode": 1,
+ "_producers": [
+ {
+ "_name": "Producer3",
+ "_destinationName": "direct://amq.direct//testQueue",
+ "_numberOfMessages": 1000000,
+ "_deliveryMode": 1,
+ "_messageSize": 10000
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ },
+ {
+ "_name": "producingClient4",
+ "_connections":[
+ {
+ "_name": "connection4",
+ "_factory": "connectionfactory",
+ "_sessions": [
+ {
+ "_sessionName": "session4",
+ "_acknowledgeMode": 1,
+ "_producers": [
+ {
+ "_name": "Producer4",
+ "_destinationName": "direct://amq.direct//testQueue",
+ "_numberOfMessages": 1000000,
+ "_deliveryMode": 1,
+ "_messageSize": 10000
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ },
+ {
+ "_name": "producingClient5",
+ "_connections":[
+ {
+ "_name": "connection5",
+ "_factory": "connectionfactory",
+ "_sessions": [
+ {
+ "_sessionName": "session5",
+ "_acknowledgeMode": "1",
+ "_producers": [
+ {
+ "_name": "Producer5",
+ "_destinationName": "direct://amq.direct//testQueue",
+ "_numberOfMessages": 1000000,
+ "_deliveryMode": 1,
+ "_messageSize": 10000
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ },
+ {
+ "_name": "consumingClient",
+ "_connections":[
+ {
+ "_name": "connection5",
+ "_factory": "connectionfactory",
+ "_sessions": [
+ {
+ "_sessionName": "session5",
+ "_acknowledgeMode": 1,
+ "_consumers": [
+ {
+ "_name": "Consumer1",
+ "_destinationName": "direct://amq.direct//testQueue",
+ "_numberOfMessages": 5000000
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ }
+ ]
+}
Added: qpid/trunk/qpid/java/perftests/example/config-stress-testing.json
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/example/config-stress-testing.json?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/example/config-stress-testing.json (added)
+++ qpid/trunk/qpid/java/perftests/example/config-stress-testing.json Thu Apr 5 16:04:10 2012
@@ -0,0 +1,84 @@
+{
+ "_tests":[
+ {
+ "_name": "One producer to one consumer";
+ "_iterations":[
+ {
+ "_messageSize": 10000
+ },
+ {
+ "_messageSize": 10000
+ },
+ {
+ "_messageSize": 10000
+ },
+ {
+ "_messageSize": 10000
+ },
+ {
+ "_messageSize": 10000
+ },
+ {
+ "_messageSize": 10000
+ },
+ {
+ "_messageSize": 10000
+ },
+ {
+ "_messageSize": 10000
+ }
+ ],
+ "_queues":[
+ {
+ "_name": "direct://amq.direct//testQueue"
+ }
+ ],
+ "_clients":[
+ {
+ "_name": "producingClient",
+ "_connections":[
+ {
+ "_name": "connection1",
+ "_factory": "connectionfactory",
+ "_sessions": [
+ {
+ "_sessionName": "session1",
+ "_acknowledgeMode": 1,
+ "_producers": [
+ {
+ "_name": "Producer1",
+ "_destinationName": "direct://amq.direct//testQueue",
+ "_numberOfMessages": 1000000,
+ "_deliveryMode": 1
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ },
+ {
+ "_name": "consumingClient",
+ "_connections":[
+ {
+ "_name": "connection1",
+ "_factory": "connectionfactory",
+ "_sessions": [
+ {
+ "_sessionName": "session1",
+ "_acknowledgeMode": 1,
+ "_consumers": [
+ {
+ "_name": "Consumer1",
+ "_destinationName": "direct://amq.direct//testQueue",
+ "_numberOfMessages": 1000000
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ }]
+}
Added: qpid/trunk/qpid/java/perftests/example/config-two-tests.json
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/example/config-two-tests.json?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/example/config-two-tests.json (added)
+++ qpid/trunk/qpid/java/perftests/example/config-two-tests.json Thu Apr 5 16:04:10 2012
@@ -0,0 +1,117 @@
+{
+ "_tests":[
+ {
+ "_name": "One producer to one consumer";
+ "_queues":[
+ {
+ "_name": "direct://amq.direct//testQueue"
+ }
+ ];
+ "_clients":[
+ {
+ "_name": "producingClient",
+ "_connections":[
+ {
+ "_name": "connection1",
+ "_factory": "connectionfactory",
+ "_sessions": [
+ {
+ "_sessionName": "session1",
+ "_producers": [
+ {
+ "_name": "Producer1",
+ "_destinationName": "direct://amq.direct//testQueue",
+ "_numberOfMessages": 5000
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ },
+ {
+ "_name": "consumingClient",
+ "_connections":[
+ {
+ "_name": "connection1",
+ "_factory": "connectionfactory",
+ "_sessions": [
+ {
+ "_sessionName": "session1",
+ "_consumers": [
+ {
+ "_name": "Consumer1",
+ "_destinationName": "direct://amq.direct//testQueue",
+ "_numberOfMessages": 5000
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ },
+ {
+ "_name": "One producer to two consumers";
+ "_queues":[
+ {
+ "_name": "direct://amq.direct//testQueue"
+ }
+ ];
+ "_clients":[
+ {
+ "_name": "producingClient",
+ "_connections":[
+ {
+ "_name": "connection1",
+ "_factory": "connectionfactory",
+ "_sessions": [
+ {
+ "_sessionName": "session1",
+ "_producers": [
+ {
+ "_name": "Producer",
+ "_destinationName": "direct://amq.direct//testQueue",
+ "_numberOfMessages": 5000
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ },
+ {
+ "_name": "consumingClient",
+ "_connections":[
+ {
+ "_name": "connection1",
+ "_factory": "connectionfactory",
+ "_sessions": [
+ {
+ "_sessionName": "session1",
+ "_consumers": [
+ {
+ "_name": "Consumer1",
+ "_destinationName": "direct://amq.direct//testQueue",
+ "_numberOfMessages": 2500
+ }
+ ]
+ },
+ {
+ "_sessionName": "session2",
+ "_consumers": [
+ {
+ "_name": "Consumer2",
+ "_destinationName": "direct://amq.direct//testQueue",
+ "_numberOfMessages": 2500
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ }
+ ]
+ }]
+}
Added: qpid/trunk/qpid/java/perftests/example/log4j-client.properties
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/example/log4j-client.properties?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/example/log4j-client.properties (added)
+++ qpid/trunk/qpid/java/perftests/example/log4j-client.properties Thu Apr 5 16:04:10 2012
@@ -0,0 +1,29 @@
+#
+# 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.
+#
+log4j.rootLogger=ERROR,console
+
+log4j.logger.org.apache.qpid=WARN
+log4j.logger.org.apache.qpid.disttest=INFO
+
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.Threshold=all
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+
+log4j.appender.console.layout.ConversionPattern=%d %p [%c{1}] %m%n
+
Added: qpid/trunk/qpid/java/perftests/example/log4j.properties
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/example/log4j.properties?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/example/log4j.properties (added)
+++ qpid/trunk/qpid/java/perftests/example/log4j.properties Thu Apr 5 16:04:10 2012
@@ -0,0 +1,35 @@
+#
+# 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.
+#
+log4j.rootLogger=ERROR,console,fileApp
+
+log4j.logger.org.apache.qpid=WARN
+log4j.logger.org.apache.qpid.disttest=INFO
+
+log4j.appender.console=org.apache.log4j.ConsoleAppender
+log4j.appender.console.Threshold=all
+log4j.appender.console.layout=org.apache.log4j.PatternLayout
+
+log4j.appender.console.layout.ConversionPattern=%d %p [%c{1}] %m%n
+
+log4j.appender.fileApp=org.apache.log4j.FileAppender
+log4j.appender.fileApp.file=perftests.log
+log4j.appender.fileApp.Threshold=info
+log4j.appender.fileApp.append=false
+log4j.appender.fileApp.layout=org.apache.log4j.PatternLayout
+log4j.appender.fileApp.layout.ConversionPattern=%d %p [%c{1}] %m%n
\ No newline at end of file
Added: qpid/trunk/qpid/java/perftests/example/perftests-jndi.properties
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/example/perftests-jndi.properties?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/example/perftests-jndi.properties (added)
+++ qpid/trunk/qpid/java/perftests/example/perftests-jndi.properties Thu Apr 5 16:04:10 2012
@@ -0,0 +1,26 @@
+# 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.
+
+# this file is used for running system tests of the performance test framework,
+# (i.e. not for running the performance tests themselves!)
+
+java.naming.factory.initial = org.apache.qpid.jndi.PropertiesFileInitialContextFactory
+
+# use QpidBrokerTestCase's default port
+connectionfactory.connectionfactory = amqp://guest:guest@clientid/test?brokerlist='tcp://localhost:5672'
+
+destination.controllerqueue = direct://amq.direct//controllerqueue
Added: qpid/trunk/qpid/java/perftests/example/run-client.sh
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/example/run-client.sh?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/example/run-client.sh (added)
+++ qpid/trunk/qpid/java/perftests/example/run-client.sh Thu Apr 5 16:04:10 2012
@@ -0,0 +1 @@
+java -cp ".:${QPID_HOME}/lib/*" -Dlog4j.configuration=log4j-client.properties -Dqpid.dest_syntax=BURL org.apache.qpid.disttest.ClientRunner jndi-config=perftests-jndi.properties
Propchange: qpid/trunk/qpid/java/perftests/example/run-client.sh
------------------------------------------------------------------------------
svn:executable = *
Added: qpid/trunk/qpid/java/perftests/example/run.sh
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/example/run.sh?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/example/run.sh (added)
+++ qpid/trunk/qpid/java/perftests/example/run.sh Thu Apr 5 16:04:10 2012
@@ -0,0 +1,2 @@
+java -cp ".:${QPID_HOME}/lib/*" -Dqpid.dest_syntax=BURL org.apache.qpid.disttest.ControllerRunner jndi-config=perftests-jndi.properties test-config=$1 distributed=true
+
Propchange: qpid/trunk/qpid/java/perftests/example/run.sh
------------------------------------------------------------------------------
svn:executable = *
Added: qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/AbstractRunner.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/AbstractRunner.java?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/AbstractRunner.java (added)
+++ qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/AbstractRunner.java Thu Apr 5 16:04:10 2012
@@ -0,0 +1,74 @@
+/*
+ *
+ * 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.disttest;
+
+import java.io.FileInputStream;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+import javax.naming.Context;
+import javax.naming.InitialContext;
+
+public class AbstractRunner
+{
+ public static final String JNDI_CONFIG_PROP = "jndi-config";
+ public static final String JNDI_CONFIG_DEFAULT = "perftests-jndi.properties";
+
+ private Map<String,String> _cliOptions = new HashMap<String, String>();
+ {
+ getCliOptions().put(JNDI_CONFIG_PROP, JNDI_CONFIG_DEFAULT);
+ }
+
+ protected Context getContext()
+ {
+ Context context = null;
+
+ try
+ {
+ final Properties properties = new Properties();
+ properties.load(new FileInputStream(getJndiConfig()));
+ context = new InitialContext(properties);
+ }
+ catch (Exception e)
+ {
+ throw new DistributedTestException("Exception while loading JNDI properties", e);
+ }
+
+ return context;
+ }
+
+ public void parseArgumentsIntoConfig(String[] args)
+ {
+ ArgumentParser argumentParser = new ArgumentParser();
+ argumentParser.parseArgumentsIntoConfig(getCliOptions(), args);
+ }
+
+ protected String getJndiConfig()
+ {
+ return getCliOptions().get(AbstractRunner.JNDI_CONFIG_PROP);
+ }
+
+ protected Map<String,String> getCliOptions()
+ {
+ return _cliOptions;
+ }
+}
\ No newline at end of file
Added: qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/ArgumentParser.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/ArgumentParser.java?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/ArgumentParser.java (added)
+++ qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/ArgumentParser.java Thu Apr 5 16:04:10 2012
@@ -0,0 +1,44 @@
+/*
+ *
+ * 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.disttest;
+
+import java.util.Map;
+
+public class ArgumentParser
+{
+ public void parseArgumentsIntoConfig(Map<String, String> initialValues, String[] args)
+ {
+ for(String arg: args)
+ {
+ String[] splitArg = arg.split("=");
+ if(splitArg.length != 2)
+ {
+ throw new IllegalArgumentException("arguments must have format <name>=<value>: " + arg);
+ }
+
+ if(initialValues.put(splitArg[0], splitArg[1]) == null)
+ {
+ throw new IllegalArgumentException("not a valid configuration property: " + arg);
+ }
+ }
+
+ }
+}
Added: qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/ClientRunner.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/ClientRunner.java?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/ClientRunner.java (added)
+++ qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/ClientRunner.java Thu Apr 5 16:04:10 2012
@@ -0,0 +1,94 @@
+/*
+ * 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.disttest;
+
+import javax.naming.Context;
+import javax.naming.NamingException;
+
+import org.apache.qpid.disttest.client.Client;
+import org.apache.qpid.disttest.jms.ClientJmsDelegate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClientRunner extends AbstractRunner
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(ClientRunner.class);
+
+ public static final String NUM_OF_CLIENTS_PROP = "number-of-clients";
+
+ public static final String NUM_OF_CLIENTS_DEFAULT = "1";
+
+ public ClientRunner()
+ {
+ getCliOptions().put(NUM_OF_CLIENTS_PROP, NUM_OF_CLIENTS_DEFAULT);
+ }
+
+ public static void main(String[] args)
+ {
+ ClientRunner runner = new ClientRunner();
+ runner.parseArgumentsIntoConfig(args);
+ runner.runClients();
+ }
+
+ public void setJndiPropertiesFileLocation(String jndiConfigFileLocation)
+ {
+ getCliOptions().put(JNDI_CONFIG_PROP, jndiConfigFileLocation);
+ }
+
+ /**
+ * Run the clients in the background
+ */
+ public void runClients()
+ {
+ int numClients = Integer.parseInt(getCliOptions().get(NUM_OF_CLIENTS_PROP));
+
+ Context context = getContext();
+
+ for(int i = 1; i <= numClients; i++)
+ {
+ createBackgroundClient(context);
+ }
+ }
+
+ private void createBackgroundClient(Context context)
+ {
+ try
+ {
+ final Client client = new Client(new ClientJmsDelegate(context));
+
+ final Thread clientThread = new Thread(new Runnable()
+ {
+ @Override
+ public void run()
+ {
+ LOGGER.info("Starting client " + client.getClientName());
+ client.start();
+ client.waitUntilStopped();
+ LOGGER.info("Stopped client " + client.getClientName());
+ }
+ });
+ clientThread.start();
+ }
+ catch (NamingException e)
+ {
+ throw new DistributedTestException("Exception while creating client instance", e);
+ }
+ }
+}
Added: qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/ControllerRunner.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/ControllerRunner.java?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/ControllerRunner.java (added)
+++ qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/ControllerRunner.java Thu Apr 5 16:04:10 2012
@@ -0,0 +1,179 @@
+/*
+ * 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.disttest;
+
+import java.io.FileNotFoundException;
+import java.io.FileWriter;
+import java.io.IOException;
+
+import javax.naming.Context;
+
+import org.apache.qpid.disttest.controller.Controller;
+import org.apache.qpid.disttest.controller.ResultsForAllTests;
+import org.apache.qpid.disttest.controller.config.Config;
+import org.apache.qpid.disttest.controller.config.ConfigReader;
+import org.apache.qpid.disttest.jms.ControllerJmsDelegate;
+import org.apache.qpid.disttest.results.aggregation.Aggregator;
+import org.apache.qpid.disttest.results.formatting.CSVFormater;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ControllerRunner extends AbstractRunner
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(ControllerRunner.class);
+
+ public static final String TEST_CONFIG_PROP = "test-config";
+ public static final String DISTRIBUTED_PROP = "distributed";
+ public static final String OUTPUT_FILE_PROP = "outputfile";
+
+ private static final String TEST_CONFIG_DEFAULT = "perftests-config.json";
+ private static final String DISTRIBUTED_DEFAULT = "false";
+ private static final String OUTPUT_FILE_DEFAULT = "output.csv";
+
+ private Controller _controller;
+
+ private final Aggregator _aggregator = new Aggregator();
+
+
+ public ControllerRunner()
+ {
+ getCliOptions().put(TEST_CONFIG_PROP, TEST_CONFIG_DEFAULT);
+ getCliOptions().put(DISTRIBUTED_PROP, DISTRIBUTED_DEFAULT);
+ getCliOptions().put(OUTPUT_FILE_PROP, OUTPUT_FILE_DEFAULT);
+ }
+
+ public static void main(String[] args) throws Exception
+ {
+ ControllerRunner runner = new ControllerRunner();
+ runner.parseArgumentsIntoConfig(args);
+ runner.runController();
+ }
+
+ public void runController() throws Exception
+ {
+ Context context = getContext();
+
+ ControllerJmsDelegate jmsDelegate = new ControllerJmsDelegate(context);
+
+ try
+ {
+ runTests(jmsDelegate);
+ }
+ finally
+ {
+ jmsDelegate.closeConnections();
+ }
+ }
+
+ private void runTests(ControllerJmsDelegate jmsDelegate)
+ {
+
+ _controller = new Controller(jmsDelegate,
+ DistributedTestConstants.REGISTRATION_TIMEOUT, DistributedTestConstants.COMMAND_RESPONSE_TIMEOUT);
+
+ Config testConfig = getTestConfig();
+ _controller.setConfig(testConfig);
+
+ if(!isDistributed())
+ {
+ //we must create the required test clients, running in single-jvm mode
+ int numClients = testConfig.getTotalNumberOfClients();
+ for (int i = 1; i <= numClients; i++)
+ {
+ ClientRunner clientRunner = new ClientRunner();
+ clientRunner.setJndiPropertiesFileLocation(getJndiConfig());
+ clientRunner.runClients();
+ }
+ }
+
+ ResultsForAllTests resultsForAllTests = null;
+ try
+ {
+ _controller.awaitClientRegistrations();
+
+ ResultsForAllTests rawResultsForAllTests = _controller.runAllTests();
+ resultsForAllTests = _aggregator.aggregateResults(rawResultsForAllTests);
+ }
+ finally
+ {
+ _controller.stopAllRegisteredClients();
+ }
+
+ final String outputFile = getOutputFile();
+ writeResultsToFile(resultsForAllTests, outputFile);
+ }
+
+ private void writeResultsToFile(ResultsForAllTests resultsForAllTests, String outputFile)
+ {
+ FileWriter writer = null;
+ try
+ {
+ final String outputCsv = new CSVFormater().format(resultsForAllTests);
+ writer = new FileWriter(outputFile);
+ writer.write(outputCsv);
+ LOGGER.info("Wrote " + resultsForAllTests.getTestResults().size() + " test result(s) to output file " + outputFile);
+ }
+ catch (IOException e)
+ {
+ throw new DistributedTestException("Unable to write output file " + outputFile, e);
+ }
+ finally
+ {
+ if (writer != null)
+ {
+ try
+ {
+ writer.close();
+ }
+ catch (IOException e)
+ {
+ LOGGER.error("Failed to close stream for file " + outputFile, e);
+ }
+ }
+ }
+ }
+
+ private String getOutputFile()
+ {
+ return String.valueOf(getCliOptions().get(ControllerRunner.OUTPUT_FILE_PROP));
+ }
+
+ private Config getTestConfig()
+ {
+ ConfigReader configReader = new ConfigReader();
+ Config testConfig;
+ try
+ {
+ testConfig = configReader.getConfigFromFile(getCliOptions().get(ControllerRunner.TEST_CONFIG_PROP));
+ }
+ catch (FileNotFoundException e)
+ {
+ throw new DistributedTestException("Exception while loading test config", e);
+ }
+ return testConfig;
+ }
+
+ private boolean isDistributed()
+ {
+ return Boolean.valueOf(getCliOptions().get(ControllerRunner.DISTRIBUTED_PROP));
+ }
+
+
+}
Added: qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/DistributedTestConstants.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/DistributedTestConstants.java?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/DistributedTestConstants.java (added)
+++ qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/DistributedTestConstants.java Thu Apr 5 16:04:10 2012
@@ -0,0 +1,33 @@
+/*
+ * 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.disttest;
+
+public abstract class DistributedTestConstants
+{
+ public static final String PERF_TEST_PROPERTIES_FILE = "perftests.properties";
+
+ public static final String MSG_COMMAND_PROPERTY = "COMMAND";
+ public static final String MSG_JSON_PROPERTY = "JSON";
+
+ public static final long REGISTRATION_TIMEOUT = 60000;
+ public static final long COMMAND_RESPONSE_TIMEOUT = 10000;
+
+ public static final String CONTROLLER_QUEUE_JNDI_NAME = "controllerqueue";
+}
Added: qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/DistributedTestException.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/DistributedTestException.java?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/DistributedTestException.java (added)
+++ qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/DistributedTestException.java Thu Apr 5 16:04:10 2012
@@ -0,0 +1,66 @@
+/*
+ * 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.disttest;
+
+import org.apache.qpid.disttest.message.Command;
+
+public class DistributedTestException extends RuntimeException
+{
+ private static final long serialVersionUID = 1L;
+ private final Command causeCommand;
+
+ public DistributedTestException(final String message)
+ {
+ this(message, (Command) null);
+ }
+
+ public DistributedTestException(final Throwable cause)
+ {
+ this(cause, null);
+ }
+
+ public DistributedTestException(final String message, final Throwable cause)
+ {
+ this(message, cause, null);
+ }
+
+ public DistributedTestException(final String message, final Command commandCause)
+ {
+ super(message);
+ causeCommand = commandCause;
+ }
+
+ public DistributedTestException(final Throwable cause, final Command commandCause)
+ {
+ super(cause);
+ causeCommand = commandCause;
+ }
+
+ public DistributedTestException(final String message, final Throwable cause, final Command commandCause)
+ {
+ super(message, cause);
+ causeCommand = commandCause;
+ }
+
+ public Command getCauseCommand()
+ {
+ return causeCommand;
+ }
+}
Added: qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/Visitor.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/Visitor.java?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/Visitor.java (added)
+++ qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/Visitor.java Thu Apr 5 16:04:10 2012
@@ -0,0 +1,76 @@
+/*
+ * 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.disttest;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+
+/**
+ * A variation of the visitor pattern that uses reflection to call the correct
+ * visit method. By convention, subclasses should provide public
+ * <pre>visit(SpecificClass)</pre> methods.
+ */
+public abstract class Visitor
+{
+
+ private static final String VISITOR_METHOD_NAME = "visit";
+
+ public void visit(Object targetObject)
+ {
+ Class<? extends Object> targetObjectClass = targetObject.getClass();
+ final Method method = findVisitMethodForTargetObjectClass(targetObjectClass);
+ invokeVisitMethod(targetObject, method);
+ }
+
+ private Method findVisitMethodForTargetObjectClass(
+ Class<? extends Object> targetObjectClass)
+ {
+ final Method method;
+ try
+ {
+ method = getClass().getDeclaredMethod(VISITOR_METHOD_NAME, targetObjectClass);
+ }
+ catch (Exception e)
+ {
+ throw new DistributedTestException("Failed to find method " + VISITOR_METHOD_NAME + " on object of class " + targetObjectClass, e);
+ }
+ return method;
+ }
+
+ private void invokeVisitMethod(Object targetObject, final Method method)
+ {
+ try
+ {
+ method.invoke(this, targetObject);
+ }
+ catch (IllegalArgumentException e)
+ {
+ throw new DistributedTestException(e);
+ }
+ catch (IllegalAccessException e)
+ {
+ throw new DistributedTestException(e);
+ }
+ catch (InvocationTargetException e)
+ {
+ throw new DistributedTestException(e.getCause());
+ }
+ }
+}
Added: qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java (added)
+++ qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/Client.java Thu Apr 5 16:04:10 2012
@@ -0,0 +1,204 @@
+/*
+ * 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.disttest.client;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import javax.naming.NamingException;
+
+import org.apache.commons.lang.builder.ToStringBuilder;
+import org.apache.commons.lang.builder.ToStringStyle;
+import org.apache.qpid.disttest.DistributedTestException;
+import org.apache.qpid.disttest.Visitor;
+import org.apache.qpid.disttest.jms.ClientJmsDelegate;
+import org.apache.qpid.disttest.message.Command;
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.apache.qpid.disttest.message.Response;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class Client
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(Client.class);
+
+ private final ClientJmsDelegate _clientJmsDelegate;
+
+ private final CountDownLatch _latch = new CountDownLatch(1);
+ private Visitor _visitor;
+ private final AtomicReference<ClientState> _state;
+ private ParticipantExecutorRegistry _participantRegistry = new ParticipantExecutorRegistry();
+
+ public Client(final ClientJmsDelegate delegate) throws NamingException
+ {
+ _clientJmsDelegate = delegate;
+ _state = new AtomicReference<ClientState>(ClientState.CREATED);
+ _visitor = new ClientCommandVisitor(this, _clientJmsDelegate);
+ }
+
+ /**
+ * Register with the controller
+ */
+ public void start()
+ {
+ _clientJmsDelegate.setInstructionListener(this);
+ _clientJmsDelegate.sendRegistrationMessage();
+ _state.set(ClientState.READY);
+ }
+
+ public void stop()
+ {
+ _state.set(ClientState.STOPPED);
+ _latch.countDown();
+ }
+
+ public void addParticipantExecutor(final ParticipantExecutor participant)
+ {
+ _participantRegistry.add(participant);
+ }
+
+ public void waitUntilStopped()
+ {
+ waitUntilStopped(0);
+ }
+
+ public void waitUntilStopped(final long timeout)
+ {
+ try
+ {
+ if (timeout == 0)
+ {
+ _latch.await();
+ }
+ else
+ {
+ _latch.await(timeout, TimeUnit.MILLISECONDS);
+ }
+ }
+ catch (final InterruptedException ie)
+ {
+ Thread.currentThread().interrupt();
+ }
+
+ _clientJmsDelegate.destroy();
+ }
+
+ public void processInstruction(final Command command)
+ {
+ String responseMessage = null;
+ try
+ {
+ command.accept(_visitor);
+ }
+ catch (final Exception e)
+ {
+ LOGGER.error("Error processing instruction", e);
+ responseMessage = e.getMessage();
+ }
+ finally
+ {
+ _clientJmsDelegate.sendResponseMessage(new Response(_clientJmsDelegate.getClientName(), command.getType(), responseMessage));
+ }
+ }
+
+ public ClientState getState()
+ {
+ return _state.get();
+ }
+
+ public String getClientName()
+ {
+ return _clientJmsDelegate.getClientName();
+ }
+
+ public void setClientCommandVisitor(final ClientCommandVisitor visitor)
+ {
+ _visitor = visitor;
+ }
+
+ public void startTest()
+ {
+ if (_state.compareAndSet(ClientState.READY, ClientState.RUNNING_TEST))
+ {
+ try
+ {
+ _clientJmsDelegate.startConnections();
+ for (final ParticipantExecutor executor : _participantRegistry.executors())
+ {
+ executor.start(this);
+ }
+ }
+ catch (final Exception e)
+ {
+ try
+ {
+ tearDownTest();
+ }
+ catch (final Exception e2)
+ {
+ // ignore
+ }
+ throw new DistributedTestException("Error starting test: " + _clientJmsDelegate.getClientName(), e);
+ }
+ }
+ else
+ {
+ throw new DistributedTestException("Client '" + _clientJmsDelegate.getClientName()
+ + "' is not in READY state:" + _state.get());
+ }
+ }
+
+ public void tearDownTest()
+ {
+ if (_state.compareAndSet(ClientState.RUNNING_TEST, ClientState.READY))
+ {
+ LOGGER.info("Tearing down test on client: " + _clientJmsDelegate.getClientName());
+
+ _clientJmsDelegate.closeTestConnections();
+ }
+ else
+ {
+ throw new DistributedTestException("Client '" + _clientJmsDelegate.getClientName() + "' is not in RUNNING_TEST state! Ignoring tearDownTest");
+ }
+
+
+ _participantRegistry.clear();
+ }
+
+ public void sendResults(ParticipantResult testResult)
+ {
+ _clientJmsDelegate.sendResponseMessage(testResult);
+ LOGGER.info("Sent test results " + testResult);
+ }
+
+ @Override
+ public String toString()
+ {
+ return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)
+ .append("clientJmsDelegate", _clientJmsDelegate).toString();
+ }
+
+ void setParticipantRegistry(ParticipantExecutorRegistry participantRegistry)
+ {
+ _participantRegistry = participantRegistry;
+ }
+
+}
Added: qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ClientCommandVisitor.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ClientCommandVisitor.java?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ClientCommandVisitor.java (added)
+++ qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ClientCommandVisitor.java Thu Apr 5 16:04:10 2012
@@ -0,0 +1,98 @@
+/*
+ * 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.disttest.client;
+
+import org.apache.qpid.disttest.Visitor;
+import org.apache.qpid.disttest.jms.ClientJmsDelegate;
+import org.apache.qpid.disttest.message.CreateConnectionCommand;
+import org.apache.qpid.disttest.message.CreateConsumerCommand;
+import org.apache.qpid.disttest.message.CreateMessageProviderCommand;
+import org.apache.qpid.disttest.message.CreateProducerCommand;
+import org.apache.qpid.disttest.message.CreateSessionCommand;
+import org.apache.qpid.disttest.message.NoOpCommand;
+import org.apache.qpid.disttest.message.StartTestCommand;
+import org.apache.qpid.disttest.message.StopClientCommand;
+import org.apache.qpid.disttest.message.TearDownTestCommand;
+
+public class ClientCommandVisitor extends Visitor
+{
+ private final Client _client;
+ private final ClientJmsDelegate _clientJmsDelegate;
+
+ public ClientCommandVisitor(final Client client, final ClientJmsDelegate clientJmsDelegate)
+ {
+ super();
+ _client = client;
+ _clientJmsDelegate = clientJmsDelegate;
+ }
+
+ public void visit(final StopClientCommand command)
+ {
+ _client.stop();
+ }
+
+ public void visit(final NoOpCommand command)
+ {
+ // no-op
+ }
+
+ public void visit(final CreateConnectionCommand command)
+ {
+ _clientJmsDelegate.createConnection(command);
+ }
+
+ public void visit(final CreateSessionCommand command)
+ {
+ _clientJmsDelegate.createSession(command);
+ }
+
+ public void visit(final CreateProducerCommand command)
+ {
+
+ final ProducerParticipant participant = new ProducerParticipant(_clientJmsDelegate, command);
+ _clientJmsDelegate.createProducer(command);
+ final ParticipantExecutor executor = new ParticipantExecutor(participant);
+ _client.addParticipantExecutor(executor);
+ }
+
+ public void visit(final CreateConsumerCommand command)
+ {
+ final ConsumerParticipant participant = new ConsumerParticipant(_clientJmsDelegate, command);
+ _clientJmsDelegate.createConsumer(command);
+ final ParticipantExecutor executor = new ParticipantExecutor(participant);
+ _client.addParticipantExecutor(executor);
+ }
+
+ public void visit(final StartTestCommand command)
+ {
+ _client.startTest();
+ }
+
+ public void visit(final TearDownTestCommand command)
+ {
+ _client.tearDownTest();
+ }
+
+ public void visit(final CreateMessageProviderCommand command)
+ {
+ _clientJmsDelegate.createMessageProvider(command);
+ }
+
+}
Added: qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ClientState.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ClientState.java?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ClientState.java (added)
+++ qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ClientState.java Thu Apr 5 16:04:10 2012
@@ -0,0 +1,25 @@
+/*
+ * 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.disttest.client;
+
+public enum ClientState
+{
+ CREATED, READY, STOPPED, RUNNING_TEST;
+}
Added: qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ConsumerParticipant.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ConsumerParticipant.java?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ConsumerParticipant.java (added)
+++ qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ConsumerParticipant.java Thu Apr 5 16:04:10 2012
@@ -0,0 +1,235 @@
+/*
+ * 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.disttest.client;
+
+
+import java.util.Date;
+import java.util.NavigableSet;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import javax.jms.Message;
+import javax.jms.MessageListener;
+
+import org.apache.qpid.disttest.DistributedTestException;
+import org.apache.qpid.disttest.jms.ClientJmsDelegate;
+import org.apache.qpid.disttest.message.ConsumerParticipantResult;
+import org.apache.qpid.disttest.message.CreateConsumerCommand;
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ConsumerParticipant implements Participant
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(ConsumerParticipant.class);
+
+ private final AtomicInteger _totalNumberOfMessagesReceived = new AtomicInteger(0);
+ private final NavigableSet<Integer> _allConsumedPayloadSizes = new ConcurrentSkipListSet<Integer>();
+ private final AtomicLong _totalPayloadSizeOfAllMessagesReceived = new AtomicLong(0);
+ private final CountDownLatch _asyncRunHasFinished = new CountDownLatch(1);
+ private final ClientJmsDelegate _jmsDelegate;
+ private final CreateConsumerCommand _command;
+ private final ParticipantResultFactory _resultFactory;
+
+ private long _startTime;
+
+ private volatile Exception _asyncMessageListenerException;
+
+ public ConsumerParticipant(final ClientJmsDelegate delegate, final CreateConsumerCommand command)
+ {
+ _jmsDelegate = delegate;
+ _command = command;
+ _resultFactory = new ParticipantResultFactory();
+ }
+
+ @Override
+ public ParticipantResult doIt(String registeredClientName) throws Exception
+ {
+ final Date start = new Date();
+
+ if (_command.getMaximumDuration() == 0 && _command.getNumberOfMessages() == 0)
+ {
+ throw new DistributedTestException("number of messages and duration cannot both be zero");
+ }
+
+ if (_command.isSynchronous())
+ {
+ synchronousRun();
+ }
+ else
+ {
+ _jmsDelegate.registerListener(_command.getParticipantName(), new MessageListener(){
+
+ @Override
+ public void onMessage(Message message)
+ {
+ processAsynchMessage(message);
+ }
+
+ });
+
+ waitUntilMsgListenerHasFinished();
+ rethrowAnyAsyncMessageListenerException();
+ }
+
+ Date end = new Date();
+ int numberOfMessagesSent = _totalNumberOfMessagesReceived.get();
+ long totalPayloadSize = _totalPayloadSizeOfAllMessagesReceived.get();
+ int payloadSize = getPayloadSizeForResultIfConstantOrZeroOtherwise(_allConsumedPayloadSizes);
+
+ ConsumerParticipantResult result = _resultFactory.createForConsumer(
+ getName(),
+ registeredClientName,
+ _command,
+ numberOfMessagesSent,
+ payloadSize,
+ totalPayloadSize,
+ start,
+ end);
+
+ return result;
+ }
+
+ private void synchronousRun()
+ {
+ LOGGER.debug("entered synchronousRun: " + this);
+
+ _startTime = System.currentTimeMillis();
+
+ Message message = null;
+
+ do
+ {
+ message = _jmsDelegate.consumeMessage(_command.getParticipantName(),
+ _command.getReceiveTimeout());
+ } while (processMessage(message));
+ }
+
+ /**
+ * @return whether to continue running (ie returns false if the message quota has been reached)
+ */
+ private boolean processMessage(Message message)
+ {
+ int messageCount = _totalNumberOfMessagesReceived.incrementAndGet();
+ if (LOGGER.isTraceEnabled())
+ {
+ LOGGER.trace("message " + messageCount + " received by " + this);
+ }
+ int messagePayloadSize = _jmsDelegate.calculatePayloadSizeFrom(message);
+ _allConsumedPayloadSizes.add(messagePayloadSize);
+ _totalPayloadSizeOfAllMessagesReceived.addAndGet(messagePayloadSize);
+
+ boolean batchEnabled = _command.getBatchSize() > 0;
+ boolean batchComplete = batchEnabled && messageCount % _command.getBatchSize() == 0;
+
+ if (!batchEnabled || batchComplete)
+ {
+ _jmsDelegate.commitOrAcknowledgeMessage(message, _command.getSessionName());
+ }
+
+ boolean reachedExpectedNumberOfMessages = _command.getNumberOfMessages() > 0 && messageCount >= _command.getNumberOfMessages();
+ boolean reachedMaximumDuration = _command.getMaximumDuration() > 0 && System.currentTimeMillis() - _startTime >= _command.getMaximumDuration();
+ boolean finishedConsuming = reachedExpectedNumberOfMessages || reachedMaximumDuration;
+
+ if (finishedConsuming)
+ {
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("message " + messageCount
+ + " reachedExpectedNumberOfMessages " + reachedExpectedNumberOfMessages
+ + " reachedMaximumDuration " + reachedMaximumDuration);
+ }
+
+ if (batchEnabled && !batchComplete)
+ {
+ // commit/acknowledge remaining messages if necessary
+ _jmsDelegate.commitOrAcknowledgeMessage(message, _command.getSessionName());
+ }
+ return false;
+ }
+
+ return true;
+ }
+
+
+ /**
+ * Intended to be called from a {@link MessageListener}. Updates {@link #_asyncRunHasFinished} if
+ * no more messages should be processed, causing {@link #doIt(String)} to exit.
+ */
+ public void processAsynchMessage(Message message)
+ {
+ boolean continueRunning = true;
+ try
+ {
+ if (_startTime == 0)
+ {
+ // reset counter and start time on receiving of first message
+ _startTime = System.currentTimeMillis();
+ }
+
+ continueRunning = processMessage(message);
+ }
+ catch (Exception e)
+ {
+ LOGGER.error("Error occured consuming message " + _totalNumberOfMessagesReceived, e);
+ continueRunning = false;
+ _asyncMessageListenerException = e;
+ }
+
+ if(!continueRunning)
+ {
+ _asyncRunHasFinished.countDown();
+ }
+ }
+
+ @Override
+ public void releaseResources()
+ {
+ _jmsDelegate.closeTestConsumer(_command.getParticipantName());
+ }
+
+ private int getPayloadSizeForResultIfConstantOrZeroOtherwise(NavigableSet<Integer> allSizes)
+ {
+ return allSizes.size() == 1 ? _allConsumedPayloadSizes.first() : 0;
+ }
+
+ private void rethrowAnyAsyncMessageListenerException()
+ {
+ if (_asyncMessageListenerException != null)
+ {
+ throw new DistributedTestException(_asyncMessageListenerException);
+ }
+ }
+
+ private void waitUntilMsgListenerHasFinished() throws Exception
+ {
+ LOGGER.debug("waiting until message listener has finished for " + this);
+ _asyncRunHasFinished.await();
+ LOGGER.debug("Message listener has finished for " + this);
+ }
+
+ @Override
+ public String getName()
+ {
+ return _command.getParticipantName();
+ }
+}
Added: qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/MessageProvider.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/MessageProvider.java?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/MessageProvider.java (added)
+++ qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/MessageProvider.java Thu Apr 5 16:04:10 2012
@@ -0,0 +1,212 @@
+/*
+ * 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.disttest.client;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import org.apache.commons.beanutils.BeanUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.qpid.disttest.DistributedTestException;
+import org.apache.qpid.disttest.client.property.PropertyValue;
+import org.apache.qpid.disttest.message.CreateProducerCommand;
+
+public class MessageProvider
+{
+ public static final String TTL = "ttl";
+
+ public static final String DELIVERY_MODE = "deliveryMode";
+
+ public static final String PRIORITY = "priority";
+
+ public static final String[] STANDARD_JMS_PROPERTIES = { "correlationID", DELIVERY_MODE,
+ "expiration", "messageID", PRIORITY, "redelivered", "replyTo", "timestamp", "type", TTL };
+
+ private Map<String, PropertyValue> _messageProperties;
+ private ConcurrentMap<Integer, Future<String>> _payloads;
+
+ public MessageProvider(Map<String, PropertyValue> messageProperties)
+ {
+ _messageProperties = messageProperties;
+ _payloads = new ConcurrentHashMap<Integer, Future<String>>();
+ }
+
+ public Message nextMessage(Session session, CreateProducerCommand command) throws JMSException
+ {
+ Message message = createTextMessage(session, command);
+ setMessageProperties(message);
+ return message;
+ }
+
+ public boolean isPropertySet(String name)
+ {
+ return _messageProperties != null && _messageProperties.containsKey(name);
+ }
+
+ public void setMessageProperties(Message message) throws JMSException
+ {
+ if (_messageProperties != null)
+ {
+ for (Entry<String, PropertyValue> entry : _messageProperties.entrySet())
+ {
+ String propertyName = entry.getKey();
+ Object propertyValue = entry.getValue().getValue();
+ if (isStandardProperty(propertyName))
+ {
+ setStandardProperty(message, propertyName, propertyValue);
+ }
+ else
+ {
+ setCustomProperty(message, propertyName, propertyValue);
+ }
+ }
+ }
+ }
+
+ protected void setCustomProperty(Message message, String propertyName, Object propertyValue) throws JMSException
+ {
+ if (propertyValue instanceof Integer)
+ {
+ message.setIntProperty(propertyName, ((Integer) propertyValue).intValue());
+ }
+ else if (propertyValue instanceof Long)
+ {
+ message.setLongProperty(propertyName, ((Long) propertyValue).longValue());
+ }
+ else if (propertyValue instanceof Boolean)
+ {
+ message.setBooleanProperty(propertyName, ((Boolean) propertyValue).booleanValue());
+ }
+ else if (propertyValue instanceof Byte)
+ {
+ message.setByteProperty(propertyName, ((Byte) propertyValue).byteValue());
+ }
+ else if (propertyValue instanceof Double)
+ {
+ message.setDoubleProperty(propertyName, ((Double) propertyValue).doubleValue());
+ }
+ else if (propertyValue instanceof Float)
+ {
+ message.setFloatProperty(propertyName, ((Float) propertyValue).floatValue());
+ }
+ else if (propertyValue instanceof Short)
+ {
+ message.setShortProperty(propertyName, ((Short) propertyValue).shortValue());
+ }
+ else if (propertyValue instanceof String)
+ {
+ message.setStringProperty(propertyName, (String) propertyValue);
+ }
+ else
+ {
+ message.setObjectProperty(propertyName, propertyValue);
+ }
+ }
+
+ protected void setStandardProperty(Message message, String property, Object propertyValue) throws JMSException
+ {
+ String propertyName = "JMS" + StringUtils.capitalize(property);
+ try
+ {
+ BeanUtils.setProperty(message, propertyName, propertyValue);
+ }
+ catch (IllegalAccessException e)
+ {
+ throw new DistributedTestException("Unable to set property " + propertyName + " :" + e.getMessage(), e);
+ }
+ catch (InvocationTargetException e)
+ {
+ if (e.getCause() instanceof JMSException)
+ {
+ throw ((JMSException) e.getCause());
+ }
+ else
+ {
+ throw new DistributedTestException("Unable to set property " + propertyName + " :" + e.getMessage(), e);
+ }
+ }
+ }
+
+ protected boolean isStandardProperty(String propertyName)
+ {
+ for (int i = 0; i < STANDARD_JMS_PROPERTIES.length; i++)
+ {
+ if (propertyName.equals(STANDARD_JMS_PROPERTIES[i]))
+ {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ protected Message createTextMessage(Session ssn, final CreateProducerCommand command) throws JMSException
+ {
+ String payload = getMessagePayload(command);
+ TextMessage msg = ssn.createTextMessage();
+ msg.setText(payload);
+ return msg;
+ }
+
+ protected String getMessagePayload(final CreateProducerCommand command)
+ {
+ FutureTask<String> createTextFuture = new FutureTask<String>(new Callable<String>()
+ {
+ @Override
+ public String call() throws Exception
+ {
+ return StringUtils.repeat("a", command.getMessageSize());
+ }
+ });
+
+ Future<String> future = _payloads.putIfAbsent(command.getMessageSize(), createTextFuture);
+ if (future == null)
+ {
+ createTextFuture.run();
+ future = createTextFuture;
+ }
+ String payload = null;
+ try
+ {
+ payload = future.get();
+ }
+ catch (Exception e)
+ {
+ throw new DistributedTestException("Unable to create message payload :" + e.getMessage(), e);
+ }
+ return payload;
+ }
+
+ @Override
+ public String toString()
+ {
+ return "MessageProvider [_messageProperties=" + _messageProperties + "]";
+ }
+
+}
Added: qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/Participant.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/Participant.java?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/Participant.java (added)
+++ qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/Participant.java Thu Apr 5 16:04:10 2012
@@ -0,0 +1,31 @@
+/*
+ * 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.disttest.client;
+
+import org.apache.qpid.disttest.message.ParticipantResult;
+
+public interface Participant
+{
+ ParticipantResult doIt(String registeredClientName) throws Exception;
+
+ void releaseResources();
+
+ String getName();
+
+}
Added: qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantExecutor.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantExecutor.java?rev=1309918&view=auto
==============================================================================
--- qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantExecutor.java (added)
+++ qpid/trunk/qpid/java/perftests/src/main/java/org/apache/qpid/disttest/client/ParticipantExecutor.java Thu Apr 5 16:04:10 2012
@@ -0,0 +1,132 @@
+/*
+ * 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.disttest.client;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+
+import org.apache.commons.lang.builder.ToStringBuilder;
+import org.apache.commons.lang.builder.ToStringStyle;
+import org.apache.qpid.disttest.message.ParticipantResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ParticipantExecutor
+{
+ private static final Logger LOGGER = LoggerFactory.getLogger(ParticipantExecutor.class);
+
+ private static final ExecutorService SHARED_UNBOUNDED_THREAD_POOL = Executors.newCachedThreadPool(new DaemonThreadFactory());
+
+ private Executor _executor = SHARED_UNBOUNDED_THREAD_POOL;
+
+ private Client _client;
+
+ private final Participant _participant;
+
+ private final ParticipantResultFactory _factory;
+
+ public ParticipantExecutor(Participant participant)
+ {
+ _participant = participant;
+ _factory = new ParticipantResultFactory();
+ }
+
+ /**
+ * Schedules the test participant to be run in a background thread.
+ */
+ public void start(Client client)
+ {
+ _client = client;
+
+ LOGGER.info("Starting test participant in background thread: " + this);
+ _executor.execute(new ParticipantRunnable());
+ }
+
+ public String getParticipantName()
+ {
+ return _participant.getName();
+ }
+
+ void setExecutor(Executor executor)
+ {
+ _executor = executor;
+ }
+
+ private class ParticipantRunnable implements Runnable
+ {
+ @Override
+ public final void run()
+ {
+ Thread currentThread = Thread.currentThread();
+ final String initialThreadName = currentThread.getName();
+ currentThread.setName(initialThreadName + "-" + getParticipantName());
+
+ try
+ {
+ runParticipantAndSendResults();
+ }
+ finally
+ {
+ currentThread.setName(initialThreadName);
+ }
+ }
+
+ private void runParticipantAndSendResults()
+ {
+ ParticipantResult result = null;
+ try
+ {
+ result = _participant.doIt(_client.getClientName());
+ }
+ catch (Throwable t)
+ {
+ String errorMessage = "Unhandled error: " + t.getMessage();
+ LOGGER.error(errorMessage, t);
+ result = _factory.createForError(_participant.getName(), _client.getClientName(), errorMessage);
+ }
+ finally
+ {
+ _client.sendResults(result);
+ _participant.releaseResources();
+ }
+ }
+ }
+
+ private static final class DaemonThreadFactory implements ThreadFactory
+ {
+ @Override
+ public Thread newThread(Runnable r)
+ {
+ Thread thread = new Thread(r);
+ thread.setDaemon(true);
+ return thread;
+ }
+ }
+
+ @Override
+ public String toString()
+ {
+ return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE)
+ .append("participantName", getParticipantName())
+ .append("client", _client)
+ .toString();
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org