INT-3636: JMX Eliminate Channel Metric Proxies

JIRA: https://jira.spring.io/browse/INT-3636

- Add `enableStats()` managed operation
- Delegate to a channel metrics object instead of using a proxy

INT-3636: Polishing; PR Comments

Change field name in `MBeanExporterHelper`.

INT-3636: Polishing

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java

@@ -26,11 +26,15 @@ import java.util.concurrent.CopyOnWriteArrayList;
 
 import org.springframework.core.OrderComparator;
 import org.springframework.core.convert.ConversionService;
+import org.springframework.integration.channel.management.ChannelSendMetrics;
+import org.springframework.integration.channel.management.MessageChannelMetrics;
 import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.history.MessageHistory;
 import org.springframework.integration.history.TrackableComponent;
 import org.springframework.integration.support.converter.DefaultDatatypeChannelMessageConverter;
+import org.springframework.integration.support.management.Statistics;
+import org.springframework.jmx.export.annotation.ManagedResource;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageDeliveryException;
@@ -38,6 +42,7 @@ import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.converter.MessageConverter;
 import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.util.Assert;
+import org.springframework.util.StopWatch;
 import org.springframework.util.StringUtils;
 
 /**
@@ -51,8 +56,9 @@ import org.springframework.util.StringUtils;
  * @author Gary Russell
  * @author Artem Bilan
  */
+@ManagedResource
 public abstract class AbstractMessageChannel extends IntegrationObjectSupport
-		implements MessageChannel, TrackableComponent, ChannelInterceptorAware {
+		implements MessageChannel, TrackableComponent, ChannelInterceptorAware, MessageChannelMetrics {
 
 	private final ChannelInterceptorList interceptors = new ChannelInterceptorList();
 
@@ -66,6 +72,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 
 	private volatile MessageConverter messageConverter;
 
+	private volatile boolean statsEnabled;
+
+	private volatile ChannelSendMetrics channelMetrics;
+
 	@Override
 	public String getComponentType() {
 		return "channel";
@@ -76,6 +86,16 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 		this.shouldTrack = shouldTrack;
 	}
 
+	@Override
+	public void enableStats(boolean statsEnabled) {
+		this.statsEnabled = statsEnabled;
+	}
+
+	@Override
+	public boolean isStatsEnabled() {
+		return this.statsEnabled;
+	}
+
 	/**
 	 * Specify the Message payload datatype(s) supported by this channel. If a
 	 * payload type does not match directly, but the 'conversionService' is
@@ -192,6 +212,86 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 		return this.interceptors;
 	}
 
+	@Override
+	public void reset() {
+		this.channelMetrics.reset();
+	}
+
+	@Override
+	public int getSendCount() {
+		return this.channelMetrics.getSendCount();
+	}
+
+	@Override
+	public long getSendCountLong() {
+		return this.channelMetrics.getSendCountLong();
+	}
+
+	@Override
+	public int getSendErrorCount() {
+		return this.channelMetrics.getSendErrorCount();
+	}
+
+	@Override
+	public long getSendErrorCountLong() {
+		return this.channelMetrics.getSendErrorCountLong();
+	}
+
+	@Override
+	public double getTimeSinceLastSend() {
+		return this.channelMetrics.getTimeSinceLastSend();
+	}
+
+	@Override
+	public double getMeanSendRate() {
+		return this.channelMetrics.getMeanSendRate();
+	}
+
+	@Override
+	public double getMeanErrorRate() {
+		return this.channelMetrics.getMeanErrorRate();
+	}
+
+	@Override
+	public double getMeanErrorRatio() {
+		return this.channelMetrics.getMeanErrorRatio();
+	}
+
+	@Override
+	public double getMeanSendDuration() {
+		return this.channelMetrics.getMeanSendDuration();
+	}
+
+	@Override
+	public double getMinSendDuration() {
+		return this.channelMetrics.getMinSendDuration();
+	}
+
+	@Override
+	public double getMaxSendDuration() {
+		return this.channelMetrics.getMaxSendDuration();
+	}
+
+	@Override
+	public double getStandardDeviationSendDuration() {
+		return this.channelMetrics.getStandardDeviationSendDuration();
+	}
+
+	@Override
+	public Statistics getSendDuration() {
+		return this.channelMetrics.getSendDuration();
+	}
+
+	@Override
+	public Statistics getSendRate() {
+		return this.channelMetrics.getSendRate();
+	}
+
+	@Override
+	public Statistics getErrorRate() {
+		return this.channelMetrics.getErrorRate();
+	}
+
 	@Override
 	protected void onInit() throws Exception {
 		super.onInit();
@@ -205,6 +305,15 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 				}
 			}
 		}
+		initMetrics();
+	}
+
+	protected void initMetrics() {
+		setChannelMetrics(new ChannelSendMetrics(getComponentName()));
+	}
+
+	protected void setChannelMetrics(ChannelSendMetrics channelMetrics) {
+		this.channelMetrics = channelMetrics;
 	}
 
 	/**
@@ -263,6 +372,8 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 
 		Deque<ChannelInterceptor> interceptorStack = null;
 		boolean sent = false;
+		boolean statsProcessed = false;
+		StopWatch timer = null;
 		try {
 			if (this.datatypes.length > 0) {
 				message = this.convertPayloadIfNecessary(message);
@@ -274,7 +385,14 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 					return false;
 				}
 			}
+			if (this.statsEnabled) {
+				timer = this.channelMetrics.beforeSend();
+			}
 			sent = this.doSend(message, timeout);
+			if (this.statsEnabled) {
+				this.channelMetrics.afterSend(timer, sent);
+				statsProcessed = true;
+			}
 			this.interceptors.postSend(message, this, sent);
 			if (interceptorStack != null) {
 				this.interceptors.afterSendCompletion(message, this, sent, null, interceptorStack);
@@ -282,6 +400,9 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 			return sent;
 		}
 		catch (Exception e) {
+			if (this.statsEnabled && !statsProcessed) {
+				this.channelMetrics.afterSend(timer, false);
+			}
 			if (interceptorStack != null) {
 				this.interceptors.afterSendCompletion(message, this, sent, e, interceptorStack);
 			}
@@ -293,6 +414,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 		}
 	}
 
+	protected ChannelSendMetrics getMetrics() {
+		return this.channelMetrics;
+	}
+
 	private Message<?> convertPayloadIfNecessary(Message<?> message) {
 		// first pass checks if the payload type already matches any of the datatypes
 		for (Class<?> datatype : this.datatypes) {
@@ -309,7 +434,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 						return (Message<?>) converted;
 					}
 					else {
-						return this.getMessageBuilderFactory().withPayload(converted).copyHeaders(message.getHeaders()).build();
+						return getMessageBuilderFactory()
+								.withPayload(converted)
+								.copyHeaders(message.getHeaders())
+								.build();
 					}
 				}
 			}

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -19,6 +19,8 @@ package org.springframework.integration.channel;
 import java.util.ArrayDeque;
 import java.util.Deque;
 
+import org.springframework.integration.channel.management.ChannelReceiveMetrics;
+import org.springframework.integration.channel.management.PollableChannelManagement;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
 import org.springframework.messaging.support.ChannelInterceptor;
@@ -28,8 +30,41 @@ import org.springframework.messaging.support.ChannelInterceptor;
  *
  * @author Mark Fisher
  * @author Artem Bilan
+ * @author Gary Russell
  */
-public abstract class AbstractPollableChannel extends AbstractMessageChannel implements PollableChannel {
+public abstract class AbstractPollableChannel extends AbstractMessageChannel implements PollableChannel,
+		PollableChannelManagement {
+
+
+	@Override
+	protected void initMetrics() {
+		setChannelMetrics(new ChannelReceiveMetrics(this.getComponentName()));
+	}
+
+	@Override
+	protected ChannelReceiveMetrics getMetrics() {
+		return (ChannelReceiveMetrics) super.getMetrics();
+	}
+
+	@Override
+	public int getReceiveCount() {
+		return getMetrics().getReceiveCount();
+	}
+
+	@Override
+	public long getReceiveCountLong() {
+		return getMetrics().getReceiveCountLong();
+	}
+
+	@Override
+	public int getReceiveErrorCount() {
+		return getMetrics().getReceiveErrorCount();
+	}
+
+	@Override
+	public long getReceiveErrorCountLong() {
+		return getMetrics().getReceiveErrorCountLong();
+	}
 
 	/**
 	 * Receive the first available message from this channel. If the channel
@@ -40,7 +75,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
 	 */
 	@Override
 	public final Message<?> receive() {
-		return this.receive(-1);
+		return receive(-1);
 	}
 
 	/**
@@ -58,7 +93,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
 	 */
 	@Override
 	public final Message<?> receive(long timeout) {
-		ChannelInterceptorList interceptorList = this.getInterceptors();
+		ChannelInterceptorList interceptorList = getInterceptors();
 		Deque<ChannelInterceptor> interceptorStack = null;
 		try {
 			if (interceptorList.getInterceptors().size() > 0) {
@@ -69,6 +104,9 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
 				}
 			}
 			Message<?> message = this.doReceive(timeout);
+			if (isStatsEnabled()) {
+				getMetrics().afterReceive();
+			}
 			message = interceptorList.postReceive(message, this);
 			if (interceptorStack != null) {
 				interceptorList.afterReceiveCompletion(message, this, null, interceptorStack);
@@ -76,6 +114,9 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
 			return message;
 		}
 		catch (RuntimeException e) {
+			if (isStatsEnabled()) {
+				getMetrics().afterError();
+			}
 			if (interceptorStack != null) {
 				interceptorList.afterReceiveCompletion(null, this, e, interceptorStack);
 			}

spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -19,40 +19,168 @@ package org.springframework.integration.channel;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.integration.channel.management.ChannelSendMetrics;
+import org.springframework.integration.channel.management.MessageChannelMetrics;
+import org.springframework.integration.support.context.NamedComponent;
+import org.springframework.integration.support.management.Statistics;
+import org.springframework.jmx.export.annotation.ManagedResource;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
 
+import reactor.util.StringUtils;
+
 /**
  * A channel implementation that essentially behaves like "/dev/null".
  * All receive() calls will return <em>null</em>, and all send() calls
  * will return <em>true</em> although no action is performed.
  * Note however that the invocations are logged at debug-level.
- * 
+ *
  * @author Mark Fisher
+ * @author Gary Russell
  */
-public class NullChannel implements PollableChannel {
+@ManagedResource
+public class NullChannel implements PollableChannel, MessageChannelMetrics, BeanNameAware, NamedComponent {
 
 	private final Log logger = LogFactory.getLog(this.getClass());
 
+	private volatile ChannelSendMetrics metrics = new ChannelSendMetrics("nullChannel");
 
+	private volatile boolean statsEnabled;
+
+	private String beanName;
+
+	@Override
+	public void setBeanName(String beanName) {
+		this.beanName = beanName;
+		this.metrics = new ChannelSendMetrics(getComponentName());
+	}
+
+	@Override
+	public String getComponentName() {
+		return StringUtils.hasText(this.beanName) ? this.beanName: "nullChannel";
+	}
+
+	@Override
+	public String getComponentType() {
+		return "channel";
+	}
+
+	@Override
+	public void reset() {
+		this.metrics.reset();
+	}
+
+	@Override
+	public void enableStats(boolean statsEnabled) {
+		this.statsEnabled = statsEnabled;
+	}
+
+	@Override
+	public boolean isStatsEnabled() {
+		return this.statsEnabled;
+	}
+
+	@Override
+	public int getSendCount() {
+		return this.metrics.getSendCount();
+	}
+
+	@Override
+	public long getSendCountLong() {
+		return this.metrics.getSendCountLong();
+	}
+
+	@Override
+	public int getSendErrorCount() {
+		return this.metrics.getSendErrorCount();
+	}
+
+	@Override
+	public long getSendErrorCountLong() {
+		return this.metrics.getSendErrorCountLong();
+	}
+
+	@Override
+	public double getTimeSinceLastSend() {
+		return this.metrics.getTimeSinceLastSend();
+	}
+
+	@Override
+	public double getMeanSendRate() {
+		return this.metrics.getMeanSendRate();
+	}
+
+	@Override
+	public double getMeanErrorRate() {
+		return this.metrics.getMeanErrorRate();
+	}
+
+	@Override
+	public double getMeanErrorRatio() {
+		return this.metrics.getMeanErrorRatio();
+	}
+
+	@Override
+	public double getMeanSendDuration() {
+		return this.metrics.getMeanSendDuration();
+	}
+
+	@Override
+	public double getMinSendDuration() {
+		return this.metrics.getMinSendDuration();
+	}
+
+	@Override
+	public double getMaxSendDuration() {
+		return this.metrics.getMaxSendDuration();
+	}
+
+	@Override
+	public double getStandardDeviationSendDuration() {
+		return this.metrics.getStandardDeviationSendDuration();
+	}
+
+	@Override
+	public Statistics getSendDuration() {
+		return this.metrics.getSendDuration();
+	}
+
+	@Override
+	public Statistics getSendRate() {
+		return this.metrics.getSendRate();
+	}
+
+	@Override
+	public Statistics getErrorRate() {
+		return this.metrics.getErrorRate();
+	}
+
+	@Override
 	public boolean send(Message<?> message) {
 		if (logger.isDebugEnabled()) {
 			logger.debug("message sent to null channel: " + message);
 		}
+		if (this.statsEnabled) {
+			this.metrics.afterSend(this.metrics.beforeSend(), true);
+		}
 		return true;
 	}
 
+	@Override
 	public boolean send(Message<?> message, long timeout) {
 		return this.send(message);
 	}
 
+	@Override
 	public Message<?> receive() {
 		if (logger.isDebugEnabled()) {
 			logger.debug("receive called on null channel");
-		}		
+		}
 		return null;
 	}
 
+	@Override
 	public Message<?> receive(long timeout) {
 		return this.receive();
 	}

spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java

@@ -47,10 +47,7 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
 
 	private volatile Integer maxSubscribers;
 
-	@Override
-	public String getComponentType(){
-		return "publish-subscribe-channel";
-	}
+
 	/**
 	 * Create a PublishSubscribeChannel that will use an {@link Executor}
 	 * to invoke the handlers. If this is null, each invocation will occur in
@@ -72,6 +69,11 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
 	}
 
 
+	@Override
+	public String getComponentType(){
+		return "publish-subscribe-channel";
+	}
+
 	/**
 	 * Provide an {@link ErrorHandler} strategy for handling Exceptions that
 	 * occur downstream from this channel. This will <i>only</i> be applied if
@@ -144,9 +146,11 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
 
 	/**
 	 * Callback method for initialization.
+	 * @throws Exception the exception.
 	 */
 	@Override
-	public final void onInit() {
+	public final void onInit() throws Exception {
+		super.onInit();
 		if (this.executor != null) {
 			if (!(this.executor instanceof ErrorHandlingTaskExecutor)) {
 				if (this.errorHandler == null) {

spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java

@@ -24,6 +24,7 @@ import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 
+import org.springframework.integration.channel.management.QueueChannelManagement;
 import org.springframework.integration.core.MessageSelector;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
@@ -40,7 +41,8 @@ import org.springframework.util.Assert;
  * @author Gary Russell
  * @author Artem Bilan
  */
-public class QueueChannel extends AbstractPollableChannel implements QueueChannelOperations {
+public class QueueChannel extends AbstractPollableChannel implements QueueChannelOperations,
+		QueueChannelManagement {
 
 	private final Queue<Message<?>> queue;
 
@@ -76,7 +78,6 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne
 		this(new LinkedBlockingQueue<Message<?>>());
 	}
 
-
 	@Override
 	protected boolean doSend(Message<?> message, long timeout) {
 		Assert.notNull(message, "'message' must not be null");

spring-integration-core/src/main/java/org/springframework/integration/channel/management/ChannelReceiveMetrics.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2002-2015 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.channel.management;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.springframework.jmx.export.annotation.ManagedOperation;
+
+/**
+ * @author Dave Syer
+ * @author Gary Russell
+ * @since 2.0
+ */
+public class ChannelReceiveMetrics extends ChannelSendMetrics {
+
+	private final AtomicLong receiveCount = new AtomicLong();
+
+	private final AtomicLong receiveErrorCount = new AtomicLong();
+
+
+	public ChannelReceiveMetrics(String name) {
+		super(name);
+	}
+
+	public void afterReceive() {
+		if (logger.isTraceEnabled()) {
+			logger.trace("Recording receive on channel(" + getName() + ") ");
+		}
+		this.receiveCount.incrementAndGet();
+	}
+
+	public void afterError() {
+		this.receiveErrorCount.incrementAndGet();
+	}
+
+	@Override
+	@ManagedOperation
+	public synchronized void reset() {
+		super.reset();
+		this.receiveErrorCount.set(0);
+		this.receiveCount.set(0);
+	}
+
+	public int getReceiveCount() {
+		return (int) this.receiveCount.get();
+	}
+
+	public long getReceiveCountLong() {
+		return this.receiveCount.get();
+	}
+
+	public int getReceiveErrorCount() {
+		return (int) this.receiveErrorCount.get();
+	}
+
+	public long getReceiveErrorCountLong() {
+		return this.receiveErrorCount.get();
+	}
+
+	@Override
+	public String toString() {
+		return String.format("MessageChannelMonitor: [name=%s, sends=%d, receives=%d]",
+				getName(), getSendCount(), this.receiveCount.get());
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/channel/management/ChannelSendMetrics.java

@@ -0,0 +1,187 @@
+/*
+ * Copyright 2009-2015 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.channel.management;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.integration.support.management.ExponentialMovingAverage;
+import org.springframework.integration.support.management.ExponentialMovingAverageRate;
+import org.springframework.integration.support.management.ExponentialMovingAverageRatio;
+import org.springframework.integration.support.management.Statistics;
+import org.springframework.jmx.export.annotation.ManagedResource;
+import org.springframework.util.StopWatch;
+
+/**
+ * Registers all message channels, and accumulates statistics about their performance. The statistics are then published
+ * locally for other components to consume and publish remotely.
+ *
+ * @author Dave Syer
+ * @author Helena Edelson
+ * @author Gary Russell
+ * @since 2.0
+ */
+@ManagedResource
+public class ChannelSendMetrics {
+
+	protected final Log logger = LogFactory.getLog(getClass());
+
+	public static final long ONE_SECOND_SECONDS = 1;
+
+	public static final long ONE_MINUTE_SECONDS = 60;
+
+	public static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10;
+
+	private final ExponentialMovingAverage sendDuration = new ExponentialMovingAverage(
+			DEFAULT_MOVING_AVERAGE_WINDOW);
+
+	private final ExponentialMovingAverageRate sendErrorRate = new ExponentialMovingAverageRate(
+			ONE_SECOND_SECONDS, ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW);
+
+	private final ExponentialMovingAverageRatio sendSuccessRatio = new ExponentialMovingAverageRatio(
+			ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW);
+
+	private final ExponentialMovingAverageRate sendRate = new ExponentialMovingAverageRate(
+			ONE_SECOND_SECONDS, ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW);
+
+	private final AtomicLong sendCount = new AtomicLong();
+
+	private final AtomicLong sendErrorCount = new AtomicLong();
+
+	private final String name;
+
+
+	public ChannelSendMetrics(String name) {
+		this.name = name;
+	}
+
+
+	public void destroy() {
+		if (logger.isDebugEnabled()) {
+			logger.debug(sendDuration);
+		}
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public StopWatch beforeSend() {
+		if (logger.isTraceEnabled()) {
+			logger.trace("Recording send on channel(" + this.name + ")");
+		}
+		final StopWatch timer = new StopWatch(this.name + ".send:execution");
+		timer.start();
+
+		this.sendCount.incrementAndGet();
+		this.sendRate.increment();
+
+		return timer;
+	}
+
+	public void afterSend(StopWatch timer, boolean result) {
+		if (timer != null) {
+			timer.stop();
+			if (result) {
+				sendSuccessRatio.success();
+				sendDuration.append(timer.getTotalTimeMillis());
+			}
+			else {
+				sendSuccessRatio.failure();
+				sendErrorCount.incrementAndGet();
+				sendErrorRate.increment();
+			}
+			if (logger.isTraceEnabled()) {
+				logger.trace(timer);
+			}
+		}
+	}
+
+	public synchronized void reset() {
+		sendDuration.reset();
+		sendErrorRate.reset();
+		sendSuccessRatio.reset();
+		sendRate.reset();
+		sendCount.set(0);
+		sendErrorCount.set(0);
+	}
+
+	public int getSendCount() {
+		return (int) sendCount.get();
+	}
+
+	public long getSendCountLong() {
+		return sendCount.get();
+	}
+
+	public int getSendErrorCount() {
+		return (int) sendErrorCount.get();
+	}
+
+	public long getSendErrorCountLong() {
+		return sendErrorCount.get();
+	}
+
+	public double getTimeSinceLastSend() {
+		return sendRate.getTimeSinceLastMeasurement();
+	}
+
+	public double getMeanSendRate() {
+		return sendRate.getMean();
+	}
+
+	public double getMeanErrorRate() {
+		return sendErrorRate.getMean();
+	}
+
+	public double getMeanErrorRatio() {
+		return 1 - sendSuccessRatio.getMean();
+	}
+
+	public double getMeanSendDuration() {
+		return sendDuration.getMean();
+	}
+
+	public double getMinSendDuration() {
+		return sendDuration.getMin();
+	}
+
+	public double getMaxSendDuration() {
+		return sendDuration.getMax();
+	}
+
+	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() {
+		return String.format("MessageChannelMonitor: [name=%s, sends=%d]", name, sendCount.get());
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/channel/management/MessageChannelMetrics.java

@@ -0,0 +1,133 @@
+/*
+ * Copyright 2002-2015 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.channel.management;
+
+import org.springframework.integration.support.management.Statistics;
+import org.springframework.jmx.export.annotation.ManagedAttribute;
+import org.springframework.jmx.export.annotation.ManagedMetric;
+import org.springframework.jmx.export.annotation.ManagedOperation;
+import org.springframework.jmx.support.MetricType;
+
+/**
+ * Interface for all message channel monitors containing accessors for various useful metrics that are generic for all
+ * channel types.
+ *
+ * @author Dave Syer
+ * @author Gary Russell
+ * @since 2.0
+ */
+public interface MessageChannelMetrics {
+
+	@ManagedOperation
+	void reset();
+
+	@ManagedOperation
+	void enableStats(boolean statsEnabled);
+
+	@ManagedAttribute
+	boolean isStatsEnabled();
+
+	/**
+	 * @return the number of successful sends
+	 */
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Send Count")
+	int getSendCount();
+
+	/**
+	 * @return the number of successful sends
+	 * @since 3.0
+	 */
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Send Count")
+	long getSendCountLong();
+
+	/**
+	 * @return the number of failed sends (either throwing an exception or rejected by the channel)
+	 */
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Send Error Count")
+	int getSendErrorCount();
+
+	/**
+	 * @return the number of failed sends (either throwing an exception or rejected by the channel)
+	 * @since 3.0
+	 */
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Send Error Count")
+	long getSendErrorCountLong();
+
+	/**
+	 * @return the time in seconds since the last send
+	 */
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Time Since Last Send in Seconds")
+	double getTimeSinceLastSend();
+
+	/**
+	 * @return the mean send rate (per second)
+	 */
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Rate per Second")
+	double getMeanSendRate();
+
+	/**
+	 * @return the mean error rate (per second).  Errors comprise all failed sends.
+	 */
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Error Rate per Second")
+	double getMeanErrorRate();
+
+	/**
+	 * @return the mean ratio of failed to successful sends in approximately the last minute
+	 */
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Mean Channel Error Ratio per Minute")
+	double getMeanErrorRatio();
+
+	/**
+	 * @return the mean send duration (milliseconds)
+	 */
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Mean Duration in Milliseconds")
+	double getMeanSendDuration();
+
+	/**
+	 * @return the minimum send duration (milliseconds) since startup
+	 */
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Min Duration in Milliseconds")
+	double getMinSendDuration();
+
+	/**
+	 * @return the maximum send duration (milliseconds) since startup
+	 */
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Max Duration in Milliseconds")
+	double getMaxSendDuration();
+
+	/**
+	 * @return the standard deviation send duration (milliseconds)
+	 */
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Standard Deviation Duration in Milliseconds")
+	double getStandardDeviationSendDuration();
+
+	/**
+	 * @return summary statistics about the send duration (milliseconds)
+	 */
+	Statistics getSendDuration();
+
+	/**
+	 * @return summary statistics about the send rates (per second)
+	 */
+	Statistics getSendRate();
+
+	/**
+	 * @return summary statistics about the error rates (per second)
+	 */
+	Statistics getErrorRate();
+
+}

spring-integration-core/src/main/java/org/springframework/integration/channel/management/PollableChannelManagement.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright 2015 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.channel.management;
+
+import org.springframework.jmx.export.annotation.ManagedMetric;
+import org.springframework.jmx.support.MetricType;
+
+/**
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public interface PollableChannelManagement extends MessageChannelMetrics {
+
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Count")
+	int getReceiveCount();
+
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Count")
+	long getReceiveCountLong();
+
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Error Count")
+	int getReceiveErrorCount();
+
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Error Count")
+	long getReceiveErrorCountLong();
+
+}

spring-integration-core/src/main/java/org/springframework/integration/channel/management/QueueChannelManagement.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2015 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.channel.management;
+
+import org.springframework.jmx.export.annotation.ManagedMetric;
+import org.springframework.jmx.support.MetricType;
+
+/**
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public interface QueueChannelManagement extends PollableChannelManagement {
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "QueueChannel Queue Size")
+	int getQueueSize();
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "QueueChannel Remaining Capacity")
+	int getRemainingCapacity();
+
+}

spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java

@@ -0,0 +1,140 @@
+/*
+ * Copyright 2009-2015 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.support.management;
+
+
+
+/**
+ * Cumulative statistics for a series of real numbers with higher weight given to recent data but without storing any
+ * history. Clients call {@link #append(double)} every time there is a new measurement, and then can collect summary
+ * statistics from the convenience getters (e.g. {@link #getStatistics()}). Older values are given exponentially smaller
+ * weight, with a decay factor determined by a "window" size chosen by the caller. The result is a good approximation to
+ * the statistics of the series but with more weight given to recent measurements, so if the statistics change over time
+ * those trends can be approximately reflected.
+ *
+ * @author Dave Syer
+ * @since 2.0
+ */
+public class ExponentialMovingAverage {
+
+	private volatile long count;
+
+	private volatile double weight;
+
+	private volatile double sum;
+
+	private volatile double sumSquares;
+
+	private volatile double min;
+
+	private volatile double max;
+
+	private final double decay;
+
+
+	/**
+	 * Create a moving average accumulator with decay lapse window provided. Measurements older than this will have
+	 * smaller weight than <code>1/e</code>.
+	 *
+	 * @param window the exponential lapse window (number of measurements)
+	 */
+	public ExponentialMovingAverage(int window) {
+		this.decay = 1 - 1. / window;
+	}
+
+
+	public synchronized void reset() {
+		weight = 0;
+		sum = 0;
+		sumSquares = 0;
+		count = 0;
+		min = 0;
+		max = 0;
+	}
+
+	/**
+	 * Add a new measurement to the series.
+	 *
+	 * @param value the measurement to append
+	 */
+	public synchronized void append(double value) {
+		if (value > max || count == 0) {
+			max = value;
+		}
+		if (value < min || count == 0) {
+			min = value;
+		}
+		sum = decay * sum + value;
+		sumSquares = decay * sumSquares + value * value;
+		weight = decay * weight + 1;
+		count++;//NOSONAR - false positive, we're synchronized
+	}
+
+	/**
+	 * @return the number of measurements recorded
+	 */
+	public int getCount() {
+		return (int) count;
+	}
+
+	/**
+	 * @return the number of measurements recorded
+	 */
+	public long getCountLong() {
+		return count;
+	}
+
+	/**
+	 * @return the mean value
+	 */
+	public double getMean() {
+		return weight > 0 ? sum / weight : 0.;
+	}
+
+	/**
+	 * @return the approximate standard deviation
+	 */
+	public double getStandardDeviation() {
+		double mean = getMean();
+		double var = weight > 0 ? sumSquares / weight - mean * mean : 0.;
+		return var > 0 ? Math.sqrt(var) : 0;
+	}
+
+	/**
+	 * @return the maximum value recorded (not weighted)
+	 */
+	public double getMax() {
+		return max;
+	}
+
+	/**
+	 * @return the minimum value recorded (not weighted)
+	 */
+	public double getMin() {
+		return min;
+	}
+
+	/**
+	 * @return summary statistics (count, mean, standard deviation etc.)
+	 */
+	public Statistics getStatistics() {
+		return new Statistics(count, min, max, getMean(), getStandardDeviation());
+	}
+
+	@Override
+	public String toString() {
+		return getStatistics().toString();
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java

@@ -0,0 +1,162 @@
+/*
+ * Copyright 2009-2015 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.support.management;
+
+
+
+/**
+ * Cumulative statistics for an event rate with higher weight given to recent data but without storing any history.
+ * Clients call {@link #increment()} when a new event occurs, and then use convenience methods (e.g. {@link #getMean()})
+ * to retrieve estimates of the rate of event arrivals and the statistics of the series. Older values are given
+ * exponentially smaller weight, with a decay factor determined by a duration chosen by the client. The rate measurement
+ * weights decay in two dimensions:
+ * <ul>
+ * <li>in time according to the lapse period supplied: <code>weight = exp((t0-t)/T)</code> where <code>t0</code> is the
+ * last measurement time, <code>t</code> is the current time and <code>T</code> is the lapse period)</li>
+ * <li>per measurement according to the lapse window supplied: <code>weight = exp(-i/L)</code> where <code>L</code> is
+ * the lapse window and <code>i</code> is the sequence number of the measurement.</li>
+ * </ul>
+ *
+ * @author Dave Syer
+ * @author Gary Russell
+ *
+ */
+public class ExponentialMovingAverageRate {
+
+	private final ExponentialMovingAverage rates;
+
+	private volatile double weight;
+
+	private volatile double sum;
+
+	private volatile double min;
+
+	private volatile double max;
+
+	private volatile long t0 = System.currentTimeMillis();
+
+	private final double lapse;
+
+	private final double period;
+
+
+	/**
+	 * @param period the period to base the rate measurement (in seconds)
+	 * @param lapsePeriod the exponential lapse rate for the rate average (in seconds)
+	 * @param window the exponential lapse window (number of measurements)
+	 */
+	public ExponentialMovingAverageRate(double period, double lapsePeriod, int window) {
+		rates = new ExponentialMovingAverage(window);
+		this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to milliseconds
+		this.period = period * 1000; // convert to milliseconds
+	}
+
+
+	public synchronized void reset() {
+		min = 0;
+		max = 0;
+		weight = 0;
+		sum = 0;
+		t0 = System.currentTimeMillis();
+		rates.reset();
+	}
+
+	/**
+	 * Add a new event to the series.
+	 */
+	public synchronized void increment() {
+		long t = System.currentTimeMillis();
+		double value = t > t0 ? (t - t0) / period : 0;
+		if (value > max || getCount() == 0) {
+			max = value;
+		}
+		if (value < min || getCount() == 0) {
+			min = value;
+		}
+		double alpha = Math.exp((t0 - t) * lapse);
+		t0 = t;
+		sum = alpha * sum + value;
+		weight = alpha * weight + 1;
+		rates.append(sum > 0 ? weight / sum : 0);
+	}
+
+	/**
+	 * @return the number of measurements recorded
+	 */
+	public int getCount() {
+		return rates.getCount();
+	}
+
+	/**
+	 * @return the number of measurements recorded
+	 * @since 3.0
+	 */
+	public long getCountLong() {
+		return rates.getCountLong();
+	}
+
+	/**
+	 * @return the time in seconds since the last measurement
+	 */
+	public double getTimeSinceLastMeasurement() {
+		return (System.currentTimeMillis() - t0) / 1000.;
+	}
+
+	/**
+	 * @return the mean value
+	 */
+	public double getMean() {
+		long count = rates.getCountLong();
+		if (count == 0) {
+			return 0;
+		}
+		long t = System.currentTimeMillis();
+		double value = t > t0 ? (t - t0) / period : 0;
+		return count / (count / rates.getMean() + value);
+	}
+
+	/**
+	 * @return the approximate standard deviation
+	 */
+	public double getStandardDeviation() {
+		return rates.getStandardDeviation();
+	}
+
+	/**
+	 * @return the maximum value recorded (not weighted)
+	 */
+	public double getMax() {
+		return min > 0 ? 1 / min : 0;
+	}
+
+	/**
+	 * @return the minimum value recorded (not weighted)
+	 */
+	public double getMin() {
+		return max > 0 ? 1 / max : 0;
+	}
+
+	/**
+	 * @return summary statistics (count, mean, standard deviation etc.)
+	 */
+	public Statistics getStatistics() {
+		return new Statistics(getCount(), min, max, getMean(), getStandardDeviation());
+	}
+
+	@Override
+	public String toString() {
+		return String.format("[%s, timeSinceLast=%f]", getStatistics(), getTimeSinceLastMeasurement());
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java

@@ -0,0 +1,154 @@
+/*
+ * Copyright 2009-2015 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.support.management;
+
+
+
+/**
+ * Cumulative statistics for success ratio with higher weight given to recent data but without storing any history.
+ * Clients call {@link #success()} or {@link #failure()} when an event occurs, and the ratio of success to total events
+ * is accumulated. Older values are given exponentially smaller weight, with a decay factor determined by a duration
+ * chosen by the client. The rate measurement weights decay in two dimensions:
+ * <ul>
+ * <li>in time according to the lapse period supplied: <code>weight = exp((t0-t)/T)</code> where <code>t0</code> is the
+ * last measurement time, <code>t</code> is the current time and <code>T</code> is the lapse period)</li>
+ * <li>per measurement according to the lapse window supplied: <code>weight = exp(-i/L)</code> where <code>L</code> is
+ * the lapse window and <code>i</code> is the sequence number of the measurement.</li>
+ * </ul>
+ *
+ * @author Dave Syer
+ * @since 2.0
+ */
+public class ExponentialMovingAverageRatio {
+
+	private volatile double weight;
+
+	private volatile double sum;
+
+	private volatile long t0 = System.currentTimeMillis();
+
+	private final double lapse;
+
+	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 ExponentialMovingAverageRatio(double lapsePeriod, int window) {
+		this.cumulative = new ExponentialMovingAverage(window);
+		this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to millisecs
+	}
+
+
+	/**
+	 * Add a new event with successful outcome.
+	 */
+	public void success() {
+		append(1);
+	}
+
+	/**
+	 * Add a new event with failed outcome.
+	 */
+	public void failure() {
+		append(0);
+	}
+
+	public synchronized void reset() {
+		weight = 0;
+		sum = 0;
+		t0 = System.currentTimeMillis();
+		cumulative.reset();
+	}
+
+	private synchronized void append(int value) {
+		long t = System.currentTimeMillis();
+		double alpha = Math.exp((t0 - t) * lapse);
+		t0 = t;
+		sum = alpha * sum + value;
+		weight = alpha * weight + 1;
+		cumulative.append(sum / weight);
+	}
+
+	/**
+	 * @return the number of measurements recorded
+	 */
+	public int getCount() {
+		return cumulative.getCount();
+	}
+
+	/**
+	 * @return the number of measurements recorded
+	 */
+	public long getCountLong() {
+		return cumulative.getCountLong();
+	}
+
+	/**
+	 * @return the time in seconds since the last measurement
+	 */
+	public double getTimeSinceLastMeasurement() {
+		return (System.currentTimeMillis() - t0) / 1000.;
+	}
+
+	/**
+	 * @return the mean success rate
+	 */
+	public double getMean() {
+		long count = cumulative.getCountLong();
+		if (count == 0) {
+			// Optimistic to start: success rate is 100%
+			return 1;
+		}
+		long t = System.currentTimeMillis();
+		double alpha = Math.exp((t0 - t) * lapse);
+		return alpha * cumulative.getMean() + 1 - alpha;
+	}
+
+	/**
+	 * @return the approximate standard deviation of the success rate measurements
+	 */
+	public double getStandardDeviation() {
+		return cumulative.getStandardDeviation();
+	}
+
+	/**
+	 * @return the maximum value recorded of the exponential weighted average (per measurement) success rate
+	 */
+	public double getMax() {
+		return cumulative.getMax();
+	}
+
+	/**
+	 * @return the minimum value recorded of the exponential weighted average (per measurement) success rate
+	 */
+	public double getMin() {
+		return cumulative.getMin();
+	}
+
+	/**
+	 * @return summary statistics (count, mean, standard deviation etc.)
+	 */
+	public Statistics getStatistics() {
+		return new Statistics(getCount(), getMin(), getMax(), getMean(), getStandardDeviation());
+	}
+
+	@Override
+	public String toString() {
+		return String.format("[%s, timeSinceLast=%f]", getStatistics(), getTimeSinceLastMeasurement());
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/support/management/Statistics.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2002-2015 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.support.management;
+
+/**
+ * @author Dave Syer
+ * @since 2.0
+ */
+public class Statistics {
+
+	private final long count;
+
+	private final double min;
+
+	private final double max;
+
+	private final double mean;
+
+	private final double standardDeviation;
+
+
+	public Statistics(long 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 (int) this.count;
+	}
+
+	public long getCountLong() {
+		return this.count;
+	}
+
+	public double getMin() {
+		return this.min;
+	}
+
+	public double getMax() {
+		return this.max;
+	}
+
+	public double getMean() {
+		return this.mean;
+	}
+
+	public double getStandardDeviation() {
+		return this.standardDeviation;
+	}
+
+	@Override
+	public String toString() {
+		return String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f]",
+				this.count, this.min, this.max, getMean(), getStandardDeviation());
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageRateTests.java

@@ -0,0 +1,129 @@
+/*
+ * Copyright 2002-2015 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.support.management;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import org.springframework.integration.test.util.TestUtils;
+
+/**
+ * @author Dave Syer
+ * @author Gary Russell
+ *
+ */
+public class ExponentialMovingAverageRateTests {
+
+	private final static Log logger = LogFactory.getLog(ExponentialMovingAverageRateTests.class);
+
+	private final ExponentialMovingAverageRate history = new ExponentialMovingAverageRate(1., 10., 10);
+
+	@Test
+	public void testWindow() {
+		ExponentialMovingAverageRate rate = new ExponentialMovingAverageRate(1., 10., 20);
+		double decay = TestUtils.getPropertyValue(rate, "rates.decay", Double.class);
+		assertEquals(95, (int) (decay * 100.));
+	}
+
+	@Test
+	public void testGetCount() {
+		assertEquals(0, history.getCount());
+		history.increment();
+		assertEquals(1, history.getCount());
+	}
+
+	@Test
+	public void testGetTimeSinceLastMeasurement() throws Exception {
+		history.increment();
+		Thread.sleep(20L);
+		assertTrue(history.getTimeSinceLastMeasurement() > 0);
+	}
+
+	@Test
+	public void testGetEarlyMean() throws Exception {
+		long t0 = System.currentTimeMillis();
+		assertEquals(0, history.getMean(), 0.01);
+		Thread.sleep(20L);
+		history.increment();
+		long elapsed = System.currentTimeMillis() - t0;
+		if (elapsed < 30L) {
+			assertTrue(history.getMean() > 10);
+		}
+		else {
+			logger.warn("Test took too long to verify mean");
+		}
+	}
+
+	@Test
+	public void testGetMean() throws Exception {
+		long t0 = System.currentTimeMillis();
+		assertEquals(0, history.getMean(), 0.01);
+		Thread.sleep(20L);
+		history.increment();
+		Thread.sleep(20L);
+		history.increment();
+		double before = history.getMean();
+		long elapsed = System.currentTimeMillis() - t0;
+		if (elapsed < 50L) {
+			assertTrue(before > 10);
+			Thread.sleep(20L);
+			elapsed = System.currentTimeMillis() - t0;
+			if (elapsed < 80L) {
+				assertTrue(history.getMean() < before);
+			}
+			else {
+				logger.warn("Test took too long to verify mean");
+			}
+		}
+		else {
+			logger.warn("Test took too long to verify mean");
+		}
+	}
+
+	@Test
+	@Ignore
+	public void testGetStandardDeviation() throws Exception {
+		assertEquals(0, history.getStandardDeviation(), 0.01);
+		Thread.sleep(20L);
+		history.increment();
+		Thread.sleep(22L);
+		history.increment();
+		Thread.sleep(18L);
+		// System.err.println(history);
+		assertTrue("Standard deviation should be non-zero: " + history, history.getStandardDeviation() > 0);
+	}
+
+	@Test
+	@Ignore
+	public void testReset() throws Exception {
+		assertEquals(0, history.getStandardDeviation(), 0.01);
+		history.increment();
+		Thread.sleep(30L);
+		history.increment();
+		assertFalse(0==history.getStandardDeviation());
+		history.reset();
+		assertEquals(0, history.getStandardDeviation(), 0.01);
+		assertEquals(0, history.getCount());
+		assertEquals(0, history.getTimeSinceLastMeasurement(), 0.01);
+		assertEquals(0, history.getMean(), 0.01);
+		assertEquals(0, history.getMin(), 0.01);
+		assertEquals(0, history.getMax(), 0.01);
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageRatioTests.java

@@ -0,0 +1,133 @@
+/*
+ * Copyright 2002-2015 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.support.management;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+
+/**
+ * @author Dave Syer
+ *
+ */
+public class ExponentialMovingAverageRatioTests {
+
+	private final ExponentialMovingAverageRatio history = new ExponentialMovingAverageRatio(
+			0.5, 10);
+
+	@Test
+	public void testGetCount() {
+		assertEquals(0, history.getCount());
+		history.success();
+		assertEquals(1, history.getCount());
+	}
+
+	@Test
+	public void testGetTimeSinceLastMeasurement() throws Exception {
+		history.success();
+		Thread.sleep(20L);
+		assertTrue(history.getTimeSinceLastMeasurement() > 0);
+	}
+
+	@Test
+	public void testGetEarlyMean() throws Exception {
+		assertEquals(1, history.getMean(), 0.01);
+		history.success();
+		assertEquals(1, history.getMean(), 0.01);
+	}
+
+	@Test
+	public void testGetEarlyFailure() throws Exception {
+		assertEquals(1, history.getMean(), 0.01);
+		history.failure();
+		assertEquals(0, history.getMean(), 0.01);
+	}
+
+	@Test
+	public void testDecayedMean() throws Exception {
+		history.failure();
+		Thread.sleep(200L);
+		assertEquals(average(0, Math.exp(-0.4)), history.getMean(), 0.01);
+	}
+
+	@Test
+	public void testGetMean() throws Exception {
+		assertEquals(1, history.getMean(), 0.01);
+		history.success();
+		assertEquals(1, history.getMean(), 0.01);
+		history.success();
+		assertEquals(1, history.getMean(), 0.01);
+		history.success();
+		assertEquals(1, history.getMean(), 0.01);
+	}
+
+	@Test
+	public void testGetMeanFailuresHighRate() throws Exception {
+		assertEquals(1, history.getMean(), 0.01);
+		history.success();
+		assertEquals(average(1), history.getMean(), 0.01);
+		history.failure();
+		assertEquals(average(1, 0.5), history.getMean(), 0.1);
+		history.success();
+		assertEquals(average(1, 0.5, 0.67), history.getMean(), 0.1);
+	}
+
+	@Test
+	public void testGetMeanFailuresLowRate() throws Exception {
+		assertEquals(1, history.getMean(), 0.01);
+		history.failure();
+		assertEquals(average(0), history.getMean(), 0.01);
+		history.failure();
+		assertEquals(average(0, 0), history.getMean(), 0.01);
+		history.success();
+		assertEquals(average(0, 0, 0.33), history.getMean(), 0.1);
+	}
+
+	@Test
+	public void testGetStandardDeviation() throws Exception {
+		assertEquals(0, history.getStandardDeviation(), 0.01);
+		history.success();
+		assertEquals(0, history.getStandardDeviation(), 1);
+	}
+
+	@Test
+	public void testReset() throws Exception {
+		assertEquals(0, history.getStandardDeviation(), 0.01);
+		history.success();
+		history.failure();
+		assertFalse(0==history.getStandardDeviation());
+		history.reset();
+		assertEquals(0, history.getStandardDeviation(), 0.01);
+		assertEquals(0, history.getCount());
+		assertEquals(0, history.getTimeSinceLastMeasurement(), 0.01);
+		assertEquals(1, history.getMean(), 0.01);
+		assertEquals(0, history.getMin(), 0.01);
+		assertEquals(0, history.getMax(), 0.01);
+	}
+
+	private double average(double... values) {
+		int count = 0;
+		double sum = 0;
+		for (double d : values) {
+			sum += d;
+			count++;
+		}
+		return sum / count;
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2002-2015 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.support.management;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+import org.junit.Test;
+
+/**
+ * @author Dave Syer
+ * @author Artem Bilan
+ *
+ */
+public class ExponentialMovingAverageTests {
+
+	private final ExponentialMovingAverage history = new ExponentialMovingAverage(10);
+
+	@Test
+	public void testGetCount() {
+		assertEquals(0, history.getCount());
+		history.append(1);
+		assertEquals(1, history.getCount());
+	}
+
+	@Test
+	public void testGetMean() throws Exception {
+		assertEquals(0, history.getMean(), 0.01);
+		history.append(1);
+		history.append(1);
+		assertEquals(1, history.getMean(), 0.01);
+	}
+
+	@Test
+	public void testGetStandardDeviation() throws Exception {
+		assertEquals(0, history.getStandardDeviation(), 0.01);
+		history.append(1);
+		history.append(1);
+		assertEquals(0, history.getStandardDeviation(), 0.01);
+	}
+
+	@Test
+	public void testReset() throws Exception {
+		assertEquals(0, history.getStandardDeviation(), 0.01);
+		history.append(1);
+		history.append(2);
+		assertFalse(0==history.getStandardDeviation());
+		history.reset();
+		assertEquals(0, history.getStandardDeviation(), 0.01);
+		// INT-2165
+		assertEquals(String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f]", 0, 0d, 0d, 0d, 0d), history.toString());
+	}
+
+}