5 files changed, 444 insertions, 0 deletions
diff --git a/qpid/java/testkit/README.txt b/qpid/java/testkit/README.txt
new file mode 100644
index 0000000000..df7996f471
--- /dev/null
+++ b/qpid/java/testkit/README.txt
@@ -0,0 +1,6 @@
+Introduction
+============
+
+The Test kit for the JMS client is based on a multi-broker python framework.
+So far it could only be used with the c++ broker.
+
diff --git a/qpid/java/testkit/bin/run_soak_client.sh b/qpid/java/testkit/bin/run_soak_client.sh
new file mode 100644
index 0000000000..ea1721d988
--- /dev/null
+++ b/qpid/java/testkit/bin/run_soak_client.sh
@@ -0,0 +1,70 @@
+#!/bin/bash
+#
+# 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 is a sample script for a soak test on
+# linux environment.
+# This will start n of Producers processors and record their CPU and memory stats.
+# Also the Producer out will be saved to a file as well.
+
+if [ "$JAR_PATH" = "" ] ; then
+    echo "ERROR: Please set JAR_PATH to point to the Qpid libraries ...."
+    exit 1
+fi
+
+#1 PID, $2 freq, $3 count
+calc_stats(){
+
+for ((  i = 0 ;  i <= $3; i++  ))
+ do
+     cpu=`ps auxw | grep $1 | grep -v 'grep' | awk '{print $3}'`
+     mem=`pmap $1 | grep total | grep -v 'grep' | awk '{print substr($2,0,length($2)-1)}'`
+     echo $i","$mem","$cpu
+     sleep $2
+     cpu="0.0"
+     mem="0"
+ done
+ kill -9 $1
+}
+
+# Num of producer processors to start
+num=$1
+# Log frequency in seconds
+log_freq=$2
+# Num of iterations
+log_iter=$3
+
+class_name=$4
+log_file_name=`echo $class_name | cut -d. -f6`
+
+# The total time for the test is determined by the
+# log_freq * log_iter.
+
+shift 4
+CLASSPATH=`find $JAR_PATH -name '*.jar' | tr '\n' ":"`
+
+JVM_ARGS="-Xmx1500M $@"
+echo "Starting $log_file_name with the following params $JVM_ARGS"
+
+for ((  c = 1 ;  c <= $num;  c++  ))
+do
+  $JAVA_HOME/bin/java $JVM_ARGS -cp $CLASSPATH $class_name > ${log_file_name}_${c}.log &
+  pid=`jobs -l %% | awk '{print $2}'`
+  calc_stats $pid $log_freq $log_iter > ${log_file_name}_process_${c}.log &
+done
diff --git a/qpid/java/testkit/bin/soak_report.sh b/qpid/java/testkit/bin/soak_report.sh
new file mode 100644
index 0000000000..9da8bfa234
--- /dev/null
+++ b/qpid/java/testkit/bin/soak_report.sh
@@ -0,0 +1,161 @@
+#!/bin/sh
+#
+# 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.
+#
+
+# Sample script to run a soak test with MultiThreadedProducer/Consumer.
+# You need to provide the log freq and no of iterations
+# Ex to run 10 hours and collect 1 second samples 
+# soak_report.sh 1 36000
+
+# This script assumes that a suitable broker instance is started.
+
+log_freq=$1
+log_iter=$2
+shift 2
+JVM_ARGS=$@
+
+if [ "$QPID_TEST_HOME" = "" ] ; then
+    echo "ERROR: Please set QPID_TEST_HOME ...."
+    exit 1
+fi
+
+print_rates()
+{
+ cat $1 | awk '{
+ FS = ",";
+ count = 0;
+ total_latency = 0; 
+ min_latency = 9223372036854775807;
+ max_latency = 0;
+
+ total_tp = 0;
+ min_tp = 50000;
+ max_tp = 0;
+
+ while ((getline) == 1) 
+ { 
+     total_latency = total_latency + $3
+     total_tp = total_tp + $2
+
+     if ($3 > 0)
+     { 
+        min_latency = (($3 < min_latency) ? $3 : min_latency);
+        max_latency = (($3 > max_latency) ? $3 : max_latency);
+     }
+     if ($2 > 0)
+     { 
+        min_tp = (($2 < min_tp) ? $2 : min_tp);
+        max_tp = (($2 > max_tp) ? $2 : max_tp);
+     }
+
+     count =  count + 1
+ }
+
+ print "Avg Latency (ms) : " total_latency/count
+ print "Max Latency (ms) : " max_latency
+ print "Min Latency (ms) : " min_latency
+
+ print ""
+ print "Avg Throughput (msg/sec) : " total_tp/count
+ print "Max Throughput (msg/sec) : " max_tp
+ print "Min Throughput (msg/sec) : " min_tp
+
+ print ""
+ print "Total Iteratons " count
+
+ }'
+}
+
+print_system_stats()
+{
+ cat $1 | awk '{
+ FS = ",";
+ count = 0;
+ total_memory = 0;
+ min_memory = 9223372036854775807;
+ max_memory = 0;
+
+ total_cp = 0;
+ min_cp = 50000;
+ max_cp = 0;
+
+ while ((getline) == 1) 
+ { 
+     total_memory = total_memory + $2
+     total_cp = total_cp + $3
+
+     if ($2 > 0)
+     { 
+        min_memory = (($2 < min_memory) ? $2 : min_memory);
+        max_memory = (($2 > max_memory) ? $2 : max_memory);
+     }
+     if ($3 > 0)
+     { 
+        min_cp = (($3 < min_cp) ? $3 : min_cp);
+        max_cp = (($3 > max_cp) ? $3 : max_cp);
+     }
+
+     count =  count + 1
+ }
+
+ print "Avg Memory (MB) : " total_memory/(count*1024)
+ print "Max Memory (MB) : " max_memory/1024
+ print "Min Memory (MB) : " min_memory/1024
+
+ print ""
+ print "Avg CPU         : " total_cp/count
+ print "Max CPU         : " max_cp
+ print "Min CPU         : " min_cp
+
+ print ""
+ print "Total Iteratons " count
+
+ }'
+}
+
+
+cleanup()
+{
+  kill -9 `ps aux | grep java | grep soak | awk '{ print $2 }'`
+}
+
+print_results()
+{  
+  printf "\n======================================================= \n" 
+  print_rates MultiThreadedConsumer_1.log
+  printf "\nConsumer process stats "
+  printf "\n----------------------- \n"
+  print_system_stats MultiThreadedConsumer_process_1.log
+  printf "\nProducer process stats "
+  printf "\n----------------------- \n"
+  print_system_stats MultiThreadedProducer_process_1.log
+  printf "\n------------------------------------------------------- \n"
+}
+
+trap cleanup EXIT
+
+# runs a single instance of the MultiThreadedConsumer and MultiThreadedProducer
+sh $QPID_TEST_HOME/bin/run_soak_client.sh 1 $log_freq $log_iter org.apache.qpid.testkit.soak.MultiThreadedConsumer $JVM_ARGS
+sh $QPID_TEST_HOME/bin/run_soak_client.sh 1 $log_freq $log_iter org.apache.qpid.testkit.soak.MultiThreadedProducer $JVM_ARGS
+
+sleep_time=$((log_freq * log_iter))
+echo "sleep time : " $sleep_time
+sleep $((log_freq * log_iter))
+
+print_results
diff --git a/qpid/java/testkit/build.xml b/qpid/java/testkit/build.xml
new file mode 100644
index 0000000000..80a3373379
--- /dev/null
+++ b/qpid/java/testkit/build.xml
@@ -0,0 +1,27 @@
+<!--
+ -
+ - 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.
+ -
+ -->
+<project name="Test Kit" default="build">
+
+    <property name="module.depends" value="client common tools"/>
+
+    <import file="../module.xml"/>
+
+</project>
diff --git a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/ResourceLeakTest.java b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/ResourceLeakTest.java
new file mode 100644
index 0000000000..c240ecdf2e
--- /dev/null
+++ b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/ResourceLeakTest.java
@@ -0,0 +1,180 @@
+/*
+ *
+ * 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.testkit.soak;
+
+
+import java.util.Random;
+
+import javax.jms.BytesMessage;
+import javax.jms.Destination;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+
+import org.apache.qpid.client.AMQConnection;
+import org.apache.qpid.client.AMQQueue;
+import org.apache.qpid.client.AMQSession;
+import org.apache.qpid.client.BasicMessageConsumer;
+import org.apache.qpid.client.BasicMessageProducer;
+import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.testkit.TestLauncher;
+import org.apache.qpid.thread.Threading;
+
+/**
+ * Test Description
+ * ================
+ * This test will open x number of connections where each
+ * connection will create a session and a producer/consumer pair,
+ * and then a randomly selected set of connections (about 1/4th)
+ * will send a configurable no of messages and try to receive them.
+ * It will then sleep for configurable time interval and
+ * tear down the connections/sessions/consumers.
+ * It will then repeat the process again until the test is stopped.
+ *
+ * Purpose of the test
+ * ===================
+ * To find if the broker has leaks when cleaning resources.
+ * To find if the client has leaks with resources.
+ */
+public class ResourceLeakTest extends TestLauncher
+{
+  /*  protected long connection_idle_time = 5000;    
+    protected Random rand = new Random();
+    
+    public ResourceLeakTest()
+    {
+        super();        
+    }
+
+    public void test()
+    {
+        try
+        {
+
+            AMQConnection[] cons = new AMQConnection[connection_count];
+            Session[] sessions = new Session[connection_count];
+            MessageConsumer[] msgCons = new MessageConsumer[connection_count];
+            MessageProducer [] msgProds = new MessageProducer[connection_count];
+            
+            while (true)
+            {
+                for (int i = 0; i < connection_count; i++)
+                {
+                    AMQConnection con = new AMQConnection(url);
+                    con.start();
+                    cons[i] = con;
+                    Session ssn = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
+                    sessions[i] = ssn;
+                    Destination dest = new AMQQueue(new AMQShortString(exchange_name),
+                            new AMQShortString(routing_key + i),
+                            new AMQShortString(queue_name + i),
+                            true, //exclusive
+                            true  // auto delete
+                            );
+                    MessageConsumer msgCon = ssn.createConsumer(dest);
+                    msgCons[i] = msgCon;
+                    MessageProducer msgProd = ssn.createProducer(dest);
+                    msgProds[i] = msgProd;
+                }
+                
+                // Select some connections randomly and send/recv messages
+                // Exercise around quarter of the connections
+                for (int i=0; i < connection_count/4; i++)
+                {    
+                    int k = rand.nextInt(connection_count);
+                    
+                    BytesMessage msg = sessions[k].createBytesMessage();
+                    msg.writeBytes("Test Msg".getBytes());
+
+                    for (int j = 0; j < msg_count;j++)
+                    {
+                        msgProds[k].send(msg);
+                    }
+
+                    int j = 0;
+                    while (j < msg_count)
+                    {
+                      msgCons[k].receive();
+                      j++;
+                    }
+                }
+                System.out.println(df.format(System.currentTimeMillis()));
+                Thread.sleep(connection_idle_time);
+
+                try
+                {
+                    for (int i = 0; i < connection_count; i++)
+                    {
+                        if (!((BasicMessageConsumer)msgCons[i]).isClosed())
+                        {
+                            msgCons[i].close();
+                        }
+                        
+                        if (!((BasicMessageProducer)msgProds[i]).isClosed())
+                        {
+                            msgProds[i].close();
+                        }
+                        
+                        if (!((AMQSession)sessions[i]).isClosed())
+                        {
+                            sessions[i].close();
+                        }
+                        if (!((AMQConnection)cons[i]).isClosed())
+                        {
+                            cons[i].close();
+                        }
+                    }
+                }
+                catch (Exception e)
+                {
+                    handleError(e,"Exception closing resources");
+                }
+            }
+        }
+        catch (Exception e)
+        {
+            handleError(e,"Exception in setting up the test");
+        }
+
+    }
+
+    public static void main(String[] args)
+    {
+        final ResourceLeakTest test = new ResourceLeakTest();
+        Runnable r = new Runnable(){    
+            public void run()
+            {
+                test.test();
+            }
+        };    
+        
+        Thread t;
+        try
+        {
+            t = Threading.getThreadFactory().createThread(r);                      
+        }
+        catch(Exception e)
+        {
+            throw new Error("Error creating test thread",e);
+        }
+    }*/
+
+}