QPID-3941 Handles the stat collecting and reporting module for the jms

perf tools housed under the java/tools module. This again closely
followed the c++ versions. The Mercury tool has it's own stat and report
implementation to facilitate exisitng scripts.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1311678 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 459 insertions, 0 deletions

diff --git a/qpid/java/tools/src/main/java/org/apache/qpid/tools/report/BasicReporter.java b/qpid/java/tools/src/main/java/org/apache/qpid/tools/report/BasicReporter.java
new file mode 100644
index 0000000000..a9896c1d4e
--- /dev/null
+++ b/qpid/java/tools/src/main/java/org/apache/qpid/tools/report/BasicReporter.java
@@ -0,0 +1,113 @@
+/*
+ *
+ * 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.tools.report;
+
+import java.io.PrintStream;
+import java.lang.reflect.Constructor;
+
+import javax.jms.Message;
+
+public class BasicReporter implements Reporter
+{
+    PrintStream out;
+    int batchSize = 0;
+    int batchCount = 0;
+    boolean headerPrinted = false;
+    protected Statistics overall;
+    Statistics batch;
+    Constructor<? extends Statistics> statCtor;
+
+    public BasicReporter(Class<? extends Statistics> clazz, PrintStream out,
+            int batchSize, boolean printHeader) throws Exception
+    {
+        this.out = out;
+        this.batchSize = batchSize;
+        this.headerPrinted = !printHeader;
+        statCtor = clazz.getConstructor();
+        overall = statCtor.newInstance();
+        if (batchSize > 0)
+        {
+            batch = statCtor.newInstance();
+        }        
+    }
+
+    /* (non-Javadoc)
+     * @see org.apache.qpid.tools.report.Reporter#message(javax.jms.Message)
+     */
+    @Override
+    public void message(Message msg)
+    {
+        overall.message(msg);
+        if (batchSize > 0) 
+        {
+            batch.message(msg);            
+            if (++batchCount == batchSize) 
+            {
+                if (!headerPrinted)
+                {
+                    header(); 
+                }
+                batch.report(out);
+                batch.clear();
+                batchCount = 0;
+            }
+        }
+    }
+
+    /* (non-Javadoc)
+     * @see org.apache.qpid.tools.report.Reporter#report()
+     */
+    @Override
+    public void report()
+    {
+        if (!headerPrinted)
+        {
+            header();
+        }
+        overall.report(out);
+    }
+
+    /* (non-Javadoc)
+     * @see org.apache.qpid.tools.report.Reporter#header()
+     */
+    @Override
+    public void header()
+    {
+        headerPrinted = true;
+        overall.header(out);
+    }
+
+    /* (non-Javadoc)
+     * @see org.apache.qpid.tools.report.Reporter#log()
+     */
+    @Override
+    public void log(String s)
+    {
+        // NOOP
+    }
+
+    @Override
+    public void clear()
+    {
+        batch.clear();
+        overall.clear();
+    }
+}
diff --git a/qpid/java/tools/src/main/java/org/apache/qpid/tools/report/MercuryReporter.java b/qpid/java/tools/src/main/java/org/apache/qpid/tools/report/MercuryReporter.java
new file mode 100644
index 0000000000..e9bf7100c1
--- /dev/null
+++ b/qpid/java/tools/src/main/java/org/apache/qpid/tools/report/MercuryReporter.java
@@ -0,0 +1,167 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.tools.report;
+
+import java.io.PrintStream;
+
+import org.apache.qpid.tools.report.Statistics.Throughput;
+import org.apache.qpid.tools.report.Statistics.ThroughputAndLatency;
+
+public class MercuryReporter extends BasicReporter
+{
+    MercuryStatistics stats;
+
+    public MercuryReporter(Class<? extends MercuryStatistics> clazz, PrintStream out,
+            int batchSize, boolean printHeader) throws Exception
+    {
+        super(clazz, out, batchSize, printHeader);
+        stats = (MercuryStatistics)overall;
+    }
+
+    public double getRate()
+    {
+        return stats.getRate();
+    }
+
+    public double getAvgLatency()
+    {
+        return stats.getAvgLatency();
+    }
+
+    public double getStdDev()
+    {
+        return stats.getStdDev();
+    }
+
+    public double getMinLatency()
+    {
+        return stats.getMinLatency();
+    }
+
+    public double getMaxLatency()
+    {
+        return stats.getMaxLatency();
+    }
+
+    public int getSampleSize()
+    {
+        return stats.getSampleSize();
+    }
+
+    public interface MercuryStatistics extends Statistics
+    {
+        public double getRate();
+        public long getMinLatency();
+        public long getMaxLatency();
+        public double getAvgLatency();
+        public double getStdDev();
+        public int getSampleSize();
+    }
+
+    public static class MercuryThroughput extends Throughput implements MercuryStatistics
+    {
+        double rate = 0;
+
+        @Override
+        public void report(PrintStream out)
+        {
+            long elapsed = System.currentTimeMillis() - start;
+            rate = (double)messages/(double)elapsed;
+        }
+
+        @Override
+        public void clear()
+        {
+            super.clear();
+            rate = 0;
+        }
+
+        public double getRate()
+        {
+            return rate;
+        }
+
+        public int getSampleSize()
+        {
+            return messages;
+        }
+
+        public long getMinLatency() { return 0; }
+        public long getMaxLatency() { return 0; }
+        public double getAvgLatency(){ return 0; }
+        public double getStdDev(){ return 0; }
+
+    }
+
+    public static class MercuryThroughputAndLatency extends ThroughputAndLatency implements MercuryStatistics
+    {
+        double rate = 0;
+        double avgLatency = 0;
+        double stdDev;
+
+        @Override
+        public void report(PrintStream out)
+        {
+            long elapsed = System.currentTimeMillis() - start;
+            rate = (double)messages/(double)elapsed;
+            avgLatency = totalLatency/(double)sampleCount;
+        }
+
+        @Override
+        public void clear()
+        {
+            super.clear();
+            rate = 0;
+            avgLatency = 0;
+        }
+
+        public double getRate()
+        {
+            return rate;
+        }
+
+        public long getMinLatency()
+        { 
+            return minLatency; 
+        }
+
+        public long getMaxLatency() 
+        { 
+            return maxLatency; 
+        }
+
+        public double getAvgLatency()
+        { 
+            return avgLatency; 
+        }
+
+        public double getStdDev()
+        { 
+            return stdDev; 
+        }
+
+        public int getSampleSize()
+        {
+            return messages;
+        }
+    }
+
+}
diff --git a/qpid/java/tools/src/main/java/org/apache/qpid/tools/report/Reporter.java b/qpid/java/tools/src/main/java/org/apache/qpid/tools/report/Reporter.java
new file mode 100644
index 0000000000..5e481458be
--- /dev/null
+++ b/qpid/java/tools/src/main/java/org/apache/qpid/tools/report/Reporter.java
@@ -0,0 +1,40 @@
+/*
+ *
+ * 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.tools.report;
+
+import javax.jms.Message;
+
+public interface Reporter
+{
+
+    public void message(Message msg);
+
+    public void report();
+
+    public void header();
+
+    // Will be used by some reporters to print statements which are greped by
+    // scripts. Example see java/tools/bin/perf-report
+    public void log(String s);
+    
+    public void clear();
+
+}
\ No newline at end of file
diff --git a/qpid/java/tools/src/main/java/org/apache/qpid/tools/report/Statistics.java b/qpid/java/tools/src/main/java/org/apache/qpid/tools/report/Statistics.java
new file mode 100644
index 0000000000..73efd1f1e0
--- /dev/null
+++ b/qpid/java/tools/src/main/java/org/apache/qpid/tools/report/Statistics.java
@@ -0,0 +1,139 @@
+/*
+ *
+ * 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.tools.report;
+
+import java.io.PrintStream;
+import java.text.DecimalFormat;
+
+import javax.jms.Message;
+
+public interface Statistics
+{
+    public void message(Message msg);
+    public void report(PrintStream out);
+    public void header(PrintStream out);
+    public void clear();
+
+    static class Throughput implements Statistics 
+    {
+        DecimalFormat df = new DecimalFormat("###.##");
+        int messages = 0;
+        long start = 0;
+        boolean started = false;
+
+        @Override
+        public void message(Message msg)
+        {
+            ++messages;
+            if (!started) 
+            {
+                start = System.currentTimeMillis();
+                started = true;
+            }
+        }
+
+        @Override
+        public void report(PrintStream out)
+        {
+            long elapsed = System.currentTimeMillis() - start;
+            out.print(df.format((double)messages/(double)elapsed));            
+        }
+
+        @Override
+        public void header(PrintStream out)
+        {
+            out.print("tp(m/s)");            
+        }        
+
+        public void clear()
+        {
+            messages = 0;
+            start = 0;
+            started = false;
+        }
+    }
+
+    static class ThroughputAndLatency extends Throughput 
+    {
+        long minLatency = Long.MAX_VALUE;
+        long maxLatency = Long.MIN_VALUE;
+        double totalLatency = 0;
+        int sampleCount = 0;
+
+        @Override
+        public void message(Message msg)
+        {
+            super.message(msg);
+            try
+            {
+                long ts = msg.getLongProperty("ts");
+                long latency = System.currentTimeMillis() - ts;
+                minLatency = Math.min(latency, minLatency);
+                maxLatency = Math.min(latency, maxLatency);
+                totalLatency = totalLatency + latency;
+                sampleCount++;
+            }
+            catch(Exception e)
+            {
+                System.out.println("Error calculating latency");
+            }
+        }
+
+        @Override
+        public void report(PrintStream out)
+        {
+            super.report(out);
+            double avgLatency = totalLatency/(double)sampleCount;
+            out.append('\t')
+            .append(String.valueOf(minLatency))
+            .append('\t')
+            .append(String.valueOf(maxLatency))
+            .append('\t')
+            .append(df.format(avgLatency));   
+
+            out.flush();
+        }
+
+        @Override
+        public void header(PrintStream out)
+        {
+            super.header(out);
+            out.append('\t')
+            .append("l-min")
+            .append('\t')
+            .append("l-max")
+            .append('\t')
+            .append("l-avg");
+
+            out.flush();
+        }       
+
+        public void clear()
+        {
+            super.clear();
+            minLatency = 0;
+            maxLatency = 0;
+            totalLatency = 0;
+            sampleCount = 0;
+        }
+    }
+
+}