INT-1429: tidy up JMX

- rename some classes and add Statistics abstraction
- INT-1429: rename SimpleChannelMonitor

spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DirectChannelMonitor.java

@@ -33,7 +33,7 @@ import org.springframework.util.StopWatch;
  * @author Helena Edelson
  */
 @ManagedResource
-public class SimpleMessageChannelMonitor implements MethodInterceptor, MessageChannelMonitor {
+public class DirectChannelMonitor implements MethodInterceptor, MessageChannelMonitor {
 
 	protected final Log logger = LogFactory.getLog(getClass());
 
@@ -43,16 +43,16 @@ public class SimpleMessageChannelMonitor implements MethodInterceptor, MessageCh
 
 	public static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10;
 
-	private ExponentialMovingAverageCumulativeHistory sendDuration = new ExponentialMovingAverageCumulativeHistory(
+	private ExponentialMovingAverage sendDuration = new ExponentialMovingAverage(
 			DEFAULT_MOVING_AVERAGE_WINDOW);
 
-	private final ExponentialMovingAverageRateCumulativeHistory sendErrorRate = new ExponentialMovingAverageRateCumulativeHistory(
+	private final ExponentialMovingAverageRate sendErrorRate = new ExponentialMovingAverageRate(
 			ONE_SECOND_SECONDS, ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW);
 
-	private final ExponentialMovingAverageRatioCumulativeHistory sendSuccessRatio = new ExponentialMovingAverageRatioCumulativeHistory(
+	private final ExponentialMovingAverageRatio sendSuccessRatio = new ExponentialMovingAverageRatio(
 			ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW);
 
-	private final ExponentialMovingAverageRateCumulativeHistory sendRate = new ExponentialMovingAverageRateCumulativeHistory(
+	private final ExponentialMovingAverageRate sendRate = new ExponentialMovingAverageRate(
 			ONE_SECOND_SECONDS, ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW);
 
 	private final AtomicInteger sendCount = new AtomicInteger();
@@ -61,7 +61,7 @@ public class SimpleMessageChannelMonitor implements MethodInterceptor, MessageCh
 
 	private final String name;
 
-	public SimpleMessageChannelMonitor(String name) {
+	public DirectChannelMonitor(String name) {
 		this.name = name;
 	}
 
@@ -107,14 +107,20 @@ public class SimpleMessageChannelMonitor implements MethodInterceptor, MessageCh
 			Object result = invocation.proceed();
 
 			timer.stop();
-			sendSuccessRatio.success();
-			sendDuration.append(timer.getTotalTimeSeconds());
+			if ((Boolean)result) {
+				sendSuccessRatio.success();
+				sendDuration.append(timer.getTotalTimeSeconds());
+			} else {
+				sendSuccessRatio.failure();
+				sendErrorCount.incrementAndGet();
+				sendErrorRate.increment();
+			}
 			return result;
 
 		}
 		catch (Throwable e) {
-			sendErrorCount.incrementAndGet();
 			sendSuccessRatio.failure();
+			sendErrorCount.incrementAndGet();
 			sendErrorRate.increment();
 			throw e;
 		}
@@ -141,17 +147,17 @@ public class SimpleMessageChannelMonitor implements MethodInterceptor, MessageCh
 	}
 
 	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Rate per Second")
-	public double getSendRate() {
+	public double getMeanSendRate() {
 		return sendRate.getMean();
 	}
 
 	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Error Rate per Second")
-	public double getErrorRate() {
+	public double getMeanErrorRate() {
 		return sendErrorRate.getMean();
 	}
 
 	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Mean Channel Error Ratio per Minute")
-	public double getErrorRatio() {
+	public double getMeanErrorRatio() {
 		return 1 - sendSuccessRatio.getMean();
 	}
 
@@ -174,6 +180,18 @@ public class SimpleMessageChannelMonitor implements MethodInterceptor, MessageCh
 	public double getStandardDeviationSendDuration() {
 		return sendDuration.getStandardDeviation();
 	}
+	
+	public Statistics getSendDuration() {
+		return sendDuration.getStatistics();
+	}
+	
+	public Statistics getSendRate() {
+		return sendRate.getStatistics();
+	}
+
+	public Statistics getErrorRate() {
+		return sendErrorRate.getStatistics();
+	}
 
 	@Override
 	public String toString() {

spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverage.java

@@ -20,7 +20,7 @@ package org.springframework.integration.monitor;
  * @author Dave Syer
  * 
  */
-public class ExponentialMovingAverageCumulativeHistory {
+public class ExponentialMovingAverage {
 
 	private int count;
 
@@ -39,7 +39,7 @@ public class ExponentialMovingAverageCumulativeHistory {
 	/**
 	 * @param window the exponential lapse window (number of measurements)
 	 */
-	public ExponentialMovingAverageCumulativeHistory(int window) {
+	public ExponentialMovingAverage(int window) {
 		this.decay = 1 - 1. / window;
 	}
 
@@ -76,10 +76,13 @@ public class ExponentialMovingAverageCumulativeHistory {
 		return min;
 	}
 
+	public Statistics getStatistics() {
+		return new Statistics(count, min, max, getMean(), getStandardDeviation());
+	}
+
 	@Override
 	public String toString() {
-		return String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f]", count, min, max, getMean(),
-				getStandardDeviation());
+		return getStatistics().toString();
 	}
 
 }

spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRate.java

@@ -19,9 +19,9 @@ package org.springframework.integration.monitor;
  * @author Dave Syer
  * 
  */
-public class ExponentialMovingAverageRateCumulativeHistory {
+public class ExponentialMovingAverageRate {
 
-	private final ExponentialMovingAverageCumulativeHistory rates;
+	private final ExponentialMovingAverage rates;
 
 	private double weight;
 
@@ -42,8 +42,8 @@ public class ExponentialMovingAverageRateCumulativeHistory {
 	 * @param lapsePeriod the exponential lapse rate for the rate average (in seconds)
 	 * @param window the exponential lapse window (number of measurements)
 	 */
-	public ExponentialMovingAverageRateCumulativeHistory(double period, double lapsePeriod, int window) {
-		rates = new ExponentialMovingAverageCumulativeHistory(10);
+	public ExponentialMovingAverageRate(double period, double lapsePeriod, int window) {
+		rates = new ExponentialMovingAverage(10);
 		this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to millisecs
 		this.period = period * 1000; // convert to millisecs
 	}
@@ -99,10 +99,13 @@ public class ExponentialMovingAverageRateCumulativeHistory {
 		return max > 0 ? 1 / max : 0;
 	}
 
+	public Statistics getStatistics() {
+		return new Statistics(getCount(), min, max, getMean(), getStandardDeviation());
+	}
+
 	@Override
 	public String toString() {
-		return String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f, timeSinceLast=%f]", getCount(), getMin(),
-				getMax(), getMean(), getStandardDeviation(), getTimeSinceLastMeasurement());
+		return String.format("[%s, timeSinceLast=%f]", getStatistics(), getTimeSinceLastMeasurement());
 	}
 
 }

spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRatio.java

@@ -20,7 +20,7 @@ package org.springframework.integration.monitor;
  * @author Dave Syer
  * 
  */
-public class ExponentialMovingAverageRatioCumulativeHistory {
+public class ExponentialMovingAverageRatio {
 
 	private double weight;
 
@@ -30,14 +30,14 @@ public class ExponentialMovingAverageRatioCumulativeHistory {
 
 	private final double lapse;
 
-	private final ExponentialMovingAverageCumulativeHistory cumulative;
+	private final ExponentialMovingAverage cumulative;
 
 	/**
 	 * @param lapsePeriod the exponential lapse rate for the rate average (in seconds)
 	 * @param window the exponential lapse window (number of measurements)
 	 */
-	public ExponentialMovingAverageRatioCumulativeHistory(double lapsePeriod, int window) {
-		this.cumulative = new ExponentialMovingAverageCumulativeHistory(window);
+	public ExponentialMovingAverageRatio(double lapsePeriod, int window) {
+		this.cumulative = new ExponentialMovingAverage(window);
 		this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to millisecs
 	}
 
@@ -93,10 +93,13 @@ public class ExponentialMovingAverageRatioCumulativeHistory {
 		return cumulative.getMin();
 	}
 
+	public Statistics getStatistics() {
+		return new Statistics(getCount(), getMin(), getMax(), getMean(), getStandardDeviation());
+	}
+
 	@Override
 	public String toString() {
-		return String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f, timeSinceLast=%f]", getCount(), getMin(),
-				getMax(), getMean(), getStandardDeviation(), getTimeSinceLastMeasurement());
+		return String.format("[%s, timeSinceLast=%f]", getStatistics(), getTimeSinceLastMeasurement());
 	}
 
 }

spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java

@@ -81,9 +81,9 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 
 	private Set<SimpleMessageHandlerMonitor> handlers = new HashSet<SimpleMessageHandlerMonitor>();
 
-	private Set<SimpleMessageChannelMonitor> channels = new HashSet<SimpleMessageChannelMonitor>();
+	private Set<DirectChannelMonitor> channels = new HashSet<DirectChannelMonitor>();
 
-	private Map<String, SimpleMessageChannelMonitor> channelsByName = new HashMap<String, SimpleMessageChannelMonitor>();
+	private Map<String, DirectChannelMonitor> channelsByName = new HashMap<String, DirectChannelMonitor>();
 
 	private Map<String, MessageHandlerMonitor> handlersByName = new HashMap<String, MessageHandlerMonitor>();
 
@@ -149,7 +149,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 			return monitor;
 		}
 		if (bean instanceof MessageChannel) {
-			SimpleMessageChannelMonitor monitor;
+			DirectChannelMonitor monitor;
 			if (bean instanceof PollableChannel) {
 				Object target = extractTarget(bean);
 				if (target instanceof QueueChannel) {
@@ -160,7 +160,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 				}
 			}
 			else {
-				monitor = new SimpleMessageChannelMonitor(beanName);
+				monitor = new DirectChannelMonitor(beanName);
 			}
 			Object advised = applyChannelInterceptor(bean, monitor, beanClassLoader);
 			channels.add(monitor);
@@ -288,32 +288,16 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 	public Map<String, String> getObjectNames() {
 		return Collections.unmodifiableMap(objectNamesByName);
 	}
-
-	public double getHandlerMeanDuration(String name) {
+	
+	public Statistics getHandlerDuration(String name) {
 		if (handlersByName.containsKey(name)) {
-			return handlersByName.get(name).getMeanDuration();
+			return handlersByName.get(name).getDuration();
 		}
 		logger.debug("No handler found for (" + name + ")");
-		return -1;
+		return null;		
 	}
 
-	public long getChannelSendCount(String name) {
-		if (channelsByName.containsKey(name)) {
-			return channelsByName.get(name).getSendCount();
-		}
-		logger.debug("No channel found for (" + name + ")");
-		return -1;
-	}
-
-	public long getChannelSendErrorCount(String name) {
-		if (channelsByName.containsKey(name)) {
-			return channelsByName.get(name).getSendErrorCount();
-		}
-		logger.debug("No channel found for (" + name + ")");
-		return -1;
-	}
-
-	public long getChannelReceiveCount(String name) {
+	public int getChannelReceiveCount(String name) {
 		if (channelsByName.containsKey(name)) {
 			if (channelsByName.get(name) instanceof PollableChannelMonitor) {
 				return ((PollableChannelMonitor) channelsByName.get(name)).getReceiveCount();
@@ -323,32 +307,24 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 		return -1;
 	}
 
-	public double getChannelSendRate(String name) {
+	public Statistics getChannelSendRate(String name) {
 		if (channelsByName.containsKey(name)) {
 			return channelsByName.get(name).getSendRate();
 		}
 		logger.debug("No channel found for (" + name + ")");
-		return -1;
+		return null;
 	}
 
-	public double getChannelErrorRate(String name) {
+	public Statistics getChannelErrorRate(String name) {
 		if (channelsByName.containsKey(name)) {
 			return channelsByName.get(name).getErrorRate();
 		}
 		logger.debug("No channel found for (" + name + ")");
-		return -1;
-	}
-
-	public double getChannelMeanSendDuration(String name) {
-		if (channelsByName.containsKey(name)) {
-			return channelsByName.get(name).getMeanSendDuration();
-		}
-		logger.debug("No channel found for (" + name + ")");
-		return -1;
+		return null;
 	}
 
 	private void registerChannels() {
-		for (SimpleMessageChannelMonitor monitor : channels) {
+		for (DirectChannelMonitor monitor : channels) {
 			String name = monitor.getName();
 			// Only register once...
 			if (!channelsByName.containsKey(name)) {
@@ -379,7 +355,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 		}
 	}
 
-	private Object applyChannelInterceptor(Object bean, SimpleMessageChannelMonitor interceptor,
+	private Object applyChannelInterceptor(Object bean, DirectChannelMonitor interceptor,
 			ClassLoader beanClassLoader) {
 		NameMatchMethodPointcutAdvisor channelsAdvice = new NameMatchMethodPointcutAdvisor(interceptor);
 		channelsAdvice.addMethodName("send");

spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMonitor.java

@@ -76,6 +76,10 @@ public class LifecycleMessageHandlerMonitor implements MessageHandlerMonitor {
 		return delegate.getStandardDeviationDuration();
 	}
 
+	public Statistics getDuration() {
+		return delegate.getDuration();
+	}
+
 	public String getName() {
 		return delegate.getName();
 	}

spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMonitor.java

@@ -34,13 +34,13 @@ public interface MessageChannelMonitor {
 	double getTimeSinceLastSend();
 
 	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Rate per Second")
-	double getSendRate();
+	double getMeanSendRate();
 
 	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Error Rate per Second")
-	double getErrorRate();
+	double getMeanErrorRate();
 
 	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Mean Channel Error Ratio per Minute")
-	double getErrorRatio();
+	double getMeanErrorRatio();
 
 	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Mean Duration")
 	double getMeanSendDuration();
@@ -54,4 +54,10 @@ public interface MessageChannelMonitor {
 	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Standard Deviation Duration")
 	double getStandardDeviationSendDuration();
 
+	Statistics getSendDuration();
+
+	Statistics getSendRate();
+
+	Statistics getErrorRate();
+
 }

spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMonitor.java

@@ -42,6 +42,8 @@ public interface MessageHandlerMonitor {
 
 	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Standard Deviation Duration")
 	double getStandardDeviationDuration();
+	
+	Statistics getDuration();
 
 	String getName();
 

spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMonitor.java

@@ -15,7 +15,7 @@
  */
 package org.springframework.integration.monitor;
 
-import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.aopalliance.intercept.MethodInvocation;
 import org.springframework.integration.MessageChannel;
@@ -28,11 +28,11 @@ import org.springframework.jmx.support.MetricType;
  * @since 2.0
  * 
  */
-public class PollableChannelMonitor extends SimpleMessageChannelMonitor {
+public class PollableChannelMonitor extends DirectChannelMonitor {
 
-	private final AtomicLong receiveCount = new AtomicLong();
+	private final AtomicInteger receiveCount = new AtomicInteger();
 
-	private final AtomicLong receiveErrorCount = new AtomicLong();
+	private final AtomicInteger receiveErrorCount = new AtomicInteger();
 
 	/**
 	 * @param name
@@ -68,12 +68,12 @@ public class PollableChannelMonitor extends SimpleMessageChannelMonitor {
 	}
 
 	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receives")
-	public long getReceiveCount() {
+	public int getReceiveCount() {
 		return receiveCount.get();
 	}
 
 	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Errors")
-	public long getReceiveErrorCount() {
+	public int getReceiveErrorCount() {
 		return receiveErrorCount.get();
 	}
 

spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMonitor.java

@@ -36,7 +36,7 @@ public class SimpleMessageHandlerMonitor implements MessageHandler, MessageHandl
 
 	private final AtomicInteger errorCount = new AtomicInteger();
 
-	private final ExponentialMovingAverageCumulativeHistory duration = new ExponentialMovingAverageCumulativeHistory(
+	private final ExponentialMovingAverage duration = new ExponentialMovingAverage(
 			DEFAULT_MOVING_AVERAGE_WINDOW);
 
 	private String name;
@@ -128,6 +128,10 @@ public class SimpleMessageHandlerMonitor implements MessageHandler, MessageHandl
 	public double getStandardDeviationDuration() {
 		return duration.getStandardDeviation();
 	}
+	
+	public Statistics getDuration() {
+		return duration.getStatistics();
+	}
 
 	@Override
 	public String toString() {

spring-integration-jmx/src/main/java/org/springframework/integration/monitor/Statistics.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ *
+ * Licensed 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.springframework.integration.monitor;
+
+/**
+ * @author Dave Syer
+ * 
+ * @since 2.0
+ *
+ */
+public class Statistics {
+	
+	private final int count;
+	private final double min;
+	private final double max;
+	private final double mean;
+	private final double standardDeviation;
+
+	/**
+	 * 
+	 */
+	public Statistics(int count, double min, double max, double mean, double standardDeviation) {
+		this.count = count;
+		this.min = min;
+		this.max = max;
+		this.mean = mean;
+		this.standardDeviation = standardDeviation;
+	}
+
+	public int getCount() {
+		return count;
+	}
+
+	public double getMin() {
+		return min;
+	}
+
+	public double getMax() {
+		return max;
+	}
+
+	public double getMean() {
+		return mean;
+	}
+
+	public double getStandardDeviation() {
+		return standardDeviation;
+	}
+
+	@Override
+	public String toString() {
+		return String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f]", count, min, max, getMean(),
+				getStandardDeviation());
+	}
+
+}