INT-3653: JMX: Add Pluggable Metrics Factory

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

Also switch to using `System.nanoTime()` for deltas.

INT-3653: Polishing; PR Comments

INT-3653: Polishing

* Fix `NullChannel` for the generic type
* Add `Assert.notNull` for metrics setters and for the result of `MetricsFactory`
* Fix compile and JavaDocs warnings

spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java

@@ -24,7 +24,6 @@ import org.springframework.amqp.core.AmqpTemplate;
 import org.springframework.amqp.core.Queue;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
-import org.springframework.integration.channel.management.ChannelReceiveMetrics;
 import org.springframework.integration.channel.management.PollableChannelManagement;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
@@ -58,16 +57,6 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable
 	}
 
 
-	@Override
-	protected void initMetrics() {
-		setChannelMetrics(new ChannelReceiveMetrics(this.getComponentName()));
-	}
-
-	@Override
-	protected ChannelReceiveMetrics getMetrics() {
-		return (ChannelReceiveMetrics) super.getMetrics();
-	}
-
 	/**
 	 * Provide an explicitly configured queue name. If this is not provided, then a Queue will be created
 	 * implicitly with the channelName as its name. The implicit creation will require that either an AmqpAdmin
@@ -136,6 +125,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable
 		ChannelInterceptorList interceptorList = getInterceptors();
 		Deque<ChannelInterceptor> interceptorStack = null;
 		boolean counted = false;
+		boolean countsEnabled = isCountsEnabled();
 		try {
 			if (interceptorList.getInterceptors().size() > 0) {
 				interceptorStack = new ArrayDeque<ChannelInterceptor>();
@@ -148,7 +138,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable
 			if (object == null) {
 				return null;
 			}
-			if (isCountsEnabled()) {
+			if (countsEnabled) {
 				getMetrics().afterReceive();
 				counted = true;
 			}
@@ -166,7 +156,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable
 			return message;
 		}
 		catch (RuntimeException e) {
-			if (isCountsEnabled() && !counted) {
+			if (countsEnabled && !counted) {
 				getMetrics().afterError();
 			}
 			if (interceptorStack != null) {

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

@@ -26,14 +26,17 @@ 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.AbstractMessageChannelMetrics;
+import org.springframework.integration.channel.management.DefaultMessageChannelMetrics;
 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.ConfigurableMetricsAware;
 import org.springframework.integration.support.management.IntegrationManagedResource;
+import org.springframework.integration.support.management.MetricsContext;
 import org.springframework.integration.support.management.Statistics;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -57,7 +60,8 @@ import org.springframework.util.StringUtils;
  */
 @IntegrationManagedResource
 public abstract class AbstractMessageChannel extends IntegrationObjectSupport
-		implements MessageChannel, TrackableComponent, ChannelInterceptorAware, MessageChannelMetrics {
+		implements MessageChannel, TrackableComponent, ChannelInterceptorAware, MessageChannelMetrics,
+		ConfigurableMetricsAware<AbstractMessageChannelMetrics> {
 
 	private final ChannelInterceptorList interceptors = new ChannelInterceptorList();
 
@@ -75,7 +79,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 
 	private volatile boolean statsEnabled;
 
-	private volatile ChannelSendMetrics channelMetrics;
+	private volatile AbstractMessageChannelMetrics channelMetrics = new DefaultMessageChannelMetrics();
 
 	@Override
 	public String getComponentType() {
@@ -106,9 +110,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 			this.countsEnabled = true;
 		}
 		this.statsEnabled = statsEnabled;
-		if (this.channelMetrics != null) {
-			this.channelMetrics.setFullStatsEnabled(statsEnabled);
-		}
+		this.channelMetrics.setFullStatsEnabled(statsEnabled);
 	}
 
 	@Override
@@ -116,6 +118,16 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 		return this.statsEnabled;
 	}
 
+	protected AbstractMessageChannelMetrics getMetrics() {
+		return this.channelMetrics;
+	}
+
+	@Override
+	public void configureMetrics(AbstractMessageChannelMetrics metrics) {
+		Assert.notNull(metrics, "'metrics' must not be null");
+		this.channelMetrics = metrics;
+	}
+
 	/**
 	 * Specify the Message payload datatype(s) supported by this channel. If a
 	 * payload type does not match directly, but the 'conversionService' is
@@ -325,16 +337,9 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 				}
 			}
 		}
-		initMetrics();
-	}
-
-	protected void initMetrics() {
-		setChannelMetrics(new ChannelSendMetrics(getComponentName()));
-	}
-
-	protected void setChannelMetrics(ChannelSendMetrics channelMetrics) {
-		this.channelMetrics = channelMetrics;
-		this.channelMetrics.setFullStatsEnabled(this.statsEnabled);
+		if (this.statsEnabled) {
+			this.channelMetrics.setFullStatsEnabled(true);
+		}
 	}
 
 	/**
@@ -393,39 +398,42 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
 
 		Deque<ChannelInterceptor> interceptorStack = null;
 		boolean sent = false;
-		boolean statsProcessed = false;
-		long start = 0;
+		boolean metricsProcessed = false;
+		MetricsContext metrics = null;
+		boolean countsEnabled = this.countsEnabled;
+		ChannelInterceptorList interceptors = this.interceptors;
+		AbstractMessageChannelMetrics channelMetrics = this.channelMetrics;
 		try {
 			if (this.datatypes.length > 0) {
 				message = this.convertPayloadIfNecessary(message);
 			}
-			if (this.interceptors.getInterceptors().size() > 0) {
+			if (interceptors.getInterceptors().size() > 0) {
 				interceptorStack = new ArrayDeque<ChannelInterceptor>();
-				message = this.interceptors.preSend(message, this, interceptorStack);
+				message = interceptors.preSend(message, this, interceptorStack);
 				if (message == null) {
 					return false;
 				}
 			}
-			if (this.countsEnabled) {
-				start = this.channelMetrics.beforeSend();
+			if (countsEnabled) {
+				metrics = channelMetrics.beforeSend();
 			}
 			sent = this.doSend(message, timeout);
-			if (this.countsEnabled) {
-				this.channelMetrics.afterSend(start, sent);
-				statsProcessed = true;
+			if (countsEnabled) {
+				channelMetrics.afterSend(metrics, sent);
+				metricsProcessed = true;
 			}
-			this.interceptors.postSend(message, this, sent);
+			interceptors.postSend(message, this, sent);
 			if (interceptorStack != null) {
-				this.interceptors.afterSendCompletion(message, this, sent, null, interceptorStack);
+				interceptors.afterSendCompletion(message, this, sent, null, interceptorStack);
 			}
 			return sent;
 		}
 		catch (Exception e) {
-			if (this.countsEnabled && !statsProcessed) {
-				this.channelMetrics.afterSend(start, false);
+			if (countsEnabled && !metricsProcessed) {
+				channelMetrics.afterSend(metrics, false);
 			}
 			if (interceptorStack != null) {
-				this.interceptors.afterSendCompletion(message, this, sent, e, interceptorStack);
+				interceptors.afterSendCompletion(message, this, sent, e, interceptorStack);
 			}
 			if (e instanceof MessagingException) {
 				throw (MessagingException) e;
@@ -435,10 +443,6 @@ 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) {

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

@@ -19,7 +19,6 @@ 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;
@@ -36,16 +35,6 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
 		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();
@@ -96,6 +85,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
 		ChannelInterceptorList interceptorList = getInterceptors();
 		Deque<ChannelInterceptor> interceptorStack = null;
 		boolean counted = false;
+		boolean countsEnabled = isCountsEnabled();
 		try {
 			if (interceptorList.getInterceptors().size() > 0) {
 				interceptorStack = new ArrayDeque<ChannelInterceptor>();
@@ -105,7 +95,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
 				}
 			}
 			Message<?> message = this.doReceive(timeout);
-			if (isCountsEnabled()) {
+			if (countsEnabled) {
 				getMetrics().afterReceive();
 				counted = true;
 			}
@@ -116,7 +106,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
 			return message;
 		}
 		catch (RuntimeException e) {
-			if (isCountsEnabled() && !counted) {
+			if (countsEnabled && !counted) {
 				getMetrics().afterError();
 			}
 			if (interceptorStack != null) {

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

@@ -20,13 +20,16 @@ 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.AbstractMessageChannelMetrics;
+import org.springframework.integration.channel.management.DefaultMessageChannelMetrics;
 import org.springframework.integration.channel.management.MessageChannelMetrics;
 import org.springframework.integration.support.context.NamedComponent;
+import org.springframework.integration.support.management.ConfigurableMetricsAware;
 import org.springframework.integration.support.management.IntegrationManagedResource;
 import org.springframework.integration.support.management.Statistics;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.PollableChannel;
+import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
 /**
@@ -39,11 +42,12 @@ import org.springframework.util.StringUtils;
  * @author Gary Russell
  */
 @IntegrationManagedResource
-public class NullChannel implements PollableChannel, MessageChannelMetrics, BeanNameAware, NamedComponent {
+public class NullChannel implements PollableChannel, MessageChannelMetrics,
+		ConfigurableMetricsAware<AbstractMessageChannelMetrics>, BeanNameAware, NamedComponent {
 
 	private final Log logger = LogFactory.getLog(this.getClass());
 
-	private volatile ChannelSendMetrics channelMetrics = new ChannelSendMetrics("nullChannel");
+	private volatile AbstractMessageChannelMetrics channelMetrics = new DefaultMessageChannelMetrics("nullChannel");
 
 	private volatile boolean countsEnabled;
 
@@ -54,7 +58,7 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics, Bean
 	@Override
 	public void setBeanName(String beanName) {
 		this.beanName = beanName;
-		this.channelMetrics = new ChannelSendMetrics(getComponentName());
+		this.channelMetrics = new DefaultMessageChannelMetrics(getComponentName());
 	}
 
 	@Override
@@ -67,6 +71,12 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics, Bean
 		return "channel";
 	}
 
+	@Override
+	public void configureMetrics(AbstractMessageChannelMetrics metrics) {
+		Assert.notNull(metrics, "'metrics' must not be null");
+		this.channelMetrics = metrics;
+	}
+
 	@Override
 	public void reset() {
 		this.channelMetrics.reset();

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

@@ -0,0 +1,120 @@
+/*
+ * 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.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.integration.support.management.ConfigurableMetrics;
+import org.springframework.integration.support.management.MetricsContext;
+import org.springframework.integration.support.management.Statistics;
+
+/**
+ * Abstract base class for channel metrics implementations.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public abstract class AbstractMessageChannelMetrics implements ConfigurableMetrics {
+
+	protected final Log logger = LogFactory.getLog(getClass());
+
+	protected final String name;
+
+	private volatile boolean fullStatsEnabled;
+
+	public AbstractMessageChannelMetrics(String name) {
+		this.name = name;
+	}
+
+	/**
+	 * When false, simple counts are maintained; when true complete statistics
+	 * are maintained.
+	 * @param fullStatsEnabled true for complete statistics.
+	 */
+	public void setFullStatsEnabled(boolean fullStatsEnabled) {
+		this.fullStatsEnabled = fullStatsEnabled;
+	}
+
+	protected boolean isFullStatsEnabled() {
+		return fullStatsEnabled;
+	}
+
+	/**
+	 * Begin a send event.
+	 * @return the context to be used in a subsequent {@link #afterSend(MetricsContext, boolean)}
+	 * call.
+	 */
+	public abstract MetricsContext beforeSend();
+
+	/**
+	 * End a send event. Note that implementations typically will not validate that the
+	 * context is of the correct type and not null; callers should take care to ensure
+	 * the context is the object returned by the previous {@link #beforeSend()} call.
+	 * @param context the context.
+	 * @param result true for success, false otherwise.
+	 */
+	public abstract void afterSend(MetricsContext context, boolean result);
+
+	/**
+	 * Reset all counters/statistics.
+	 */
+	public abstract void reset();
+
+	public abstract int getSendCount();
+
+	public abstract long getSendCountLong();
+
+	public abstract int getSendErrorCount();
+
+	public abstract long getSendErrorCountLong();
+
+	public abstract double getTimeSinceLastSend();
+
+	public abstract double getMeanSendRate();
+
+	public abstract double getMeanErrorRate();
+
+	public abstract double getMeanErrorRatio();
+
+	public abstract double getMeanSendDuration();
+
+	public abstract double getMinSendDuration();
+
+	public abstract double getMaxSendDuration();
+
+	public abstract double getStandardDeviationSendDuration();
+
+	public abstract Statistics getSendDuration();
+
+	public abstract Statistics getSendRate();
+
+	public abstract Statistics getErrorRate();
+
+	public abstract void afterReceive();
+
+	public abstract void afterError();
+
+	public abstract int getReceiveCount();
+
+	public abstract long getReceiveCountLong();
+
+	public abstract int getReceiveErrorCount();
+
+	public abstract long getReceiveErrorCountLong();
+
+}

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

@@ -1,80 +0,0 @@
-/*
- * 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/DefaultMessageChannelMetrics.java

@@ -21,6 +21,7 @@ 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.MetricsContext;
 import org.springframework.integration.support.management.Statistics;
 
 /**
@@ -32,7 +33,7 @@ import org.springframework.integration.support.management.Statistics;
  * @author Gary Russell
  * @since 2.0
  */
-public class ChannelSendMetrics {
+public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics {
 
 	protected final Log logger = LogFactory.getLog(getClass());
 
@@ -58,17 +59,16 @@ public class ChannelSendMetrics {
 
 	private final AtomicLong sendErrorCount = new AtomicLong();
 
-	private final String name;
+	protected final AtomicLong receiveCount = new AtomicLong();
 
-	private volatile boolean fullStatsEnabled;
+	protected final AtomicLong receiveErrorCount = new AtomicLong();
 
-
-	public ChannelSendMetrics(String name) {
-		this.name = name;
+	public DefaultMessageChannelMetrics() {
+		super(null);
 	}
 
-	public void setFullStatsEnabled(boolean fullStatsEnabled) {
-		this.fullStatsEnabled = fullStatsEnabled;
+	public DefaultMessageChannelMetrics(String name) {
+		super(name);
 	}
 
 	public void destroy() {
@@ -77,118 +77,176 @@ public class ChannelSendMetrics {
 		}
 	}
 
-	public String getName() {
-		return name;
-	}
-
-	public long beforeSend() {
+	@Override
+	public MetricsContext beforeSend() {
 		if (logger.isTraceEnabled()) {
 			logger.trace("Recording send on channel(" + this.name + ")");
 		}
 
 		long start = 0;
-		if (this.fullStatsEnabled) {
-			start = System.currentTimeMillis();
+		if (isFullStatsEnabled()) {
+			start = System.nanoTime();
 			this.sendRate.increment();
 		}
 		this.sendCount.incrementAndGet();
-
-		return start;
+		return new DefaultChannelMetricsContext(start);
 	}
 
-	public void afterSend(long start, boolean result) {
-		if (start > 0) {
-			long now = System.currentTimeMillis();
-			long elapsed = now - start;
-			if (result && this.fullStatsEnabled) {
-				sendSuccessRatio.success(now);
-				sendDuration.append(elapsed);
-			}
-			else {
-				if (this.fullStatsEnabled) {
-	 				sendSuccessRatio.failure(now);
-					sendErrorCount.incrementAndGet();
-				}
-				sendErrorRate.increment();
-			}
-			if (logger.isTraceEnabled()) {
-				logger.trace("Elapsed: " + this.name + ": " + elapsed);
+	@Override
+	public void afterSend(MetricsContext context, boolean result) {
+		if (result && isFullStatsEnabled()) {
+			long now = System.nanoTime();
+			this.sendSuccessRatio.success(now);
+			this.sendDuration.appendNanos(now - ((DefaultChannelMetricsContext) context).start);
+		}
+		else {
+			if (isFullStatsEnabled()) {
+				this.sendSuccessRatio.failure(System.nanoTime());
+				this.sendErrorRate.increment();
 			}
+			this.sendErrorCount.incrementAndGet();
+		}
+		if (logger.isTraceEnabled()) {
+			logger.trace("Elapsed: " + this.name
+					+ ": " + ((System.nanoTime() - ((DefaultChannelMetricsContext) context).start) / 1000000.) + "ms");
 		}
 	}
 
+	@Override
 	public synchronized void reset() {
-		sendDuration.reset();
-		sendErrorRate.reset();
-		sendSuccessRatio.reset();
-		sendRate.reset();
-		sendCount.set(0);
-		sendErrorCount.set(0);
+		this.sendDuration.reset();
+		this.sendErrorRate.reset();
+		this.sendSuccessRatio.reset();
+		this.sendRate.reset();
+		this.sendCount.set(0);
+		this.sendErrorCount.set(0);
+		this.receiveErrorCount.set(0);
+		this.receiveCount.set(0);
 	}
 
+	@Override
 	public int getSendCount() {
-		return (int) sendCount.get();
+		return (int) this.sendCount.get();
 	}
 
+	@Override
 	public long getSendCountLong() {
-		return sendCount.get();
+		return this.sendCount.get();
 	}
 
+	@Override
 	public int getSendErrorCount() {
-		return (int) sendErrorCount.get();
+		return (int) this.sendErrorCount.get();
 	}
 
+	@Override
 	public long getSendErrorCountLong() {
-		return sendErrorCount.get();
+		return this.sendErrorCount.get();
 	}
 
+	@Override
 	public double getTimeSinceLastSend() {
-		return sendRate.getTimeSinceLastMeasurement();
+		return this.sendRate.getTimeSinceLastMeasurement();
 	}
 
+	@Override
 	public double getMeanSendRate() {
-		return sendRate.getMean();
+		return this.sendRate.getMean();
 	}
 
+	@Override
 	public double getMeanErrorRate() {
-		return sendErrorRate.getMean();
+		return this.sendErrorRate.getMean();
 	}
 
+	@Override
 	public double getMeanErrorRatio() {
-		return 1 - sendSuccessRatio.getMean();
+		return 1 - this.sendSuccessRatio.getMean();
 	}
 
+	@Override
 	public double getMeanSendDuration() {
-		return sendDuration.getMean();
+		return this.sendDuration.getMean();
 	}
 
+	@Override
 	public double getMinSendDuration() {
-		return sendDuration.getMin();
+		return this.sendDuration.getMin();
 	}
 
+	@Override
 	public double getMaxSendDuration() {
-		return sendDuration.getMax();
+		return this.sendDuration.getMax();
 	}
 
+	@Override
 	public double getStandardDeviationSendDuration() {
-		return sendDuration.getStandardDeviation();
+		return this.sendDuration.getStandardDeviation();
 	}
 
+	@Override
 	public Statistics getSendDuration() {
-		return sendDuration.getStatistics();
+		return this.sendDuration.getStatistics();
 	}
 
+	@Override
 	public Statistics getSendRate() {
-		return sendRate.getStatistics();
+		return this.sendRate.getStatistics();
 	}
 
+	@Override
 	public Statistics getErrorRate() {
-		return sendErrorRate.getStatistics();
+		return this.sendErrorRate.getStatistics();
+	}
+
+	@Override
+	public void afterReceive() {
+		if (logger.isTraceEnabled()) {
+			logger.trace("Recording receive on channel(" + this.name + ") ");
+		}
+		this.receiveCount.incrementAndGet();
+	}
+
+	@Override
+	public void afterError() {
+		this.receiveErrorCount.incrementAndGet();
+	}
+
+	@Override
+	public int getReceiveCount() {
+		return (int) this.receiveCount.get();
+	}
+
+	@Override
+	public long getReceiveCountLong() {
+		return this.receiveCount.get();
+	}
+
+	@Override
+	public int getReceiveErrorCount() {
+		return (int) this.receiveErrorCount.get();
+	}
+
+	@Override
+	public long getReceiveErrorCountLong() {
+		return this.receiveErrorCount.get();
 	}
 
 	@Override
 	public String toString() {
-		return String.format("MessageChannelMonitor: [name=%s, sends=%d]", name, sendCount.get());
+		return String.format("MessageChannelMonitor: [name=%s, sends=%d"
+				+ (this.receiveCount.get() == 0 ? "" : this.receiveCount.get())
+				+ "]", name, sendCount.get());
+	}
+
+	private static class DefaultChannelMetricsContext implements MetricsContext {
+
+		private final long start;
+
+		public DefaultChannelMetricsContext(long start) {
+			this.start = start;
+		}
+
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java

@@ -19,11 +19,14 @@ package org.springframework.integration.handler;
 import org.springframework.core.Ordered;
 import org.springframework.integration.context.IntegrationObjectSupport;
 import org.springframework.integration.context.Orderable;
+import org.springframework.integration.handler.management.AbstractMessageHandlerMetrics;
+import org.springframework.integration.handler.management.DefaultMessageHandlerMetrics;
 import org.springframework.integration.handler.management.MessageHandlerMetrics;
-import org.springframework.integration.handler.management.SimpleMessageHandlerMetrics;
 import org.springframework.integration.history.MessageHistory;
 import org.springframework.integration.history.TrackableComponent;
+import org.springframework.integration.support.management.ConfigurableMetricsAware;
 import org.springframework.integration.support.management.IntegrationManagedResource;
+import org.springframework.integration.support.management.MetricsContext;
 import org.springframework.integration.support.management.Statistics;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandler;
@@ -43,13 +46,13 @@ import org.springframework.util.Assert;
  */
 @IntegrationManagedResource
 public abstract class AbstractMessageHandler extends IntegrationObjectSupport implements MessageHandler,
-		MessageHandlerMetrics, TrackableComponent, Orderable {
+		MessageHandlerMetrics, ConfigurableMetricsAware<AbstractMessageHandlerMetrics>, TrackableComponent, Orderable {
 
 	private volatile boolean shouldTrack = false;
 
 	private volatile int order = Ordered.LOWEST_PRECEDENCE;
 
-	private final SimpleMessageHandlerMetrics handlerMetrics = new SimpleMessageHandlerMetrics();
+	private volatile AbstractMessageHandlerMetrics handlerMetrics = new DefaultMessageHandlerMetrics();
 
 	private volatile boolean statsEnabled;
 
@@ -79,6 +82,19 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
 		this.shouldTrack = shouldTrack;
 	}
 
+	@Override
+	public void configureMetrics(AbstractMessageHandlerMetrics metrics) {
+		Assert.notNull(metrics, "'metrics' must not be null");
+		this.handlerMetrics = metrics;
+	}
+
+	@Override
+	protected void onInit() throws Exception {
+		if (this.statsEnabled) {
+			this.handlerMetrics.setFullStatsEnabled(true);
+		}
+	}
+
 	@Override
 	public final void handleMessage(Message<?> message) {
 		Assert.notNull(message, "Message must not be null");
@@ -86,19 +102,25 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
 		if (this.logger.isDebugEnabled()) {
 			this.logger.debug(this + " received message: " + message);
 		}
-		long start = 0;
+		MetricsContext start = null;
+		boolean countsEnabled = this.countsEnabled;
+		AbstractMessageHandlerMetrics handlerMetrics = this.handlerMetrics;
 		try {
 			if (message != null && this.shouldTrack) {
 				message = MessageHistory.write(message, this, this.getMessageBuilderFactory());
 			}
-			if (this.countsEnabled) {
-				start = this.handlerMetrics.beforeHandle(message);
+			if (countsEnabled) {
+				start = handlerMetrics.beforeHandle(message);
 			}
 			this.handleMessageInternal(message);
-			this.handlerMetrics.afterHandle(start, true);
+			if (countsEnabled) {
+				handlerMetrics.afterHandle(start, true);
+			}
 		}
 		catch (Exception e) {
-			this.handlerMetrics.afterHandle(start, false);
+			if (countsEnabled) {
+				handlerMetrics.afterHandle(start, false);
+			}
 			if (e instanceof MessagingException) {
 				throw (MessagingException) e;
 			}

spring-integration-core/src/main/java/org/springframework/integration/handler/management/AbstractMessageHandlerMetrics.java

@@ -0,0 +1,96 @@
+/*
+ * 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.handler.management;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.integration.support.management.ConfigurableMetrics;
+import org.springframework.integration.support.management.MetricsContext;
+import org.springframework.integration.support.management.Statistics;
+import org.springframework.messaging.Message;
+
+/**
+ * Abstract base class for handler metrics implementations.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public abstract class AbstractMessageHandlerMetrics implements ConfigurableMetrics {
+
+	protected final Log logger = LogFactory.getLog(getClass());
+
+	protected final String name;
+
+	private volatile boolean fullStatsEnabled;
+
+	public AbstractMessageHandlerMetrics(String name) {
+		this.name = name;
+	}
+
+	/**
+	 * When false, simple counts are maintained; when true complete statistics
+	 * are maintained.
+	 * @param fullStatsEnabled true for complete statistics.
+	 */
+	public void setFullStatsEnabled(boolean fullStatsEnabled) {
+		this.fullStatsEnabled = fullStatsEnabled;
+	}
+
+	protected boolean isFullStatsEnabled() {
+		return fullStatsEnabled;
+	}
+
+	/**
+	 * Begin a handle event.
+	 * @param message the message to handle.
+	 * @return the context.
+	 */
+	public abstract MetricsContext beforeHandle(Message<?> message);
+
+	/**
+	 * End a handle event
+	 * @param context the context.
+	 * @param success true for success, false otherwise.
+	 */
+	public abstract void afterHandle(MetricsContext context, boolean success);
+
+	public abstract void reset();
+
+	public abstract long getHandleCountLong();
+
+	public abstract int getHandleCount();
+
+	public abstract int getErrorCount();
+
+	public abstract long getErrorCountLong();
+
+	public abstract double getMeanDuration();
+
+	public abstract double getMinDuration();
+
+	public abstract double getMaxDuration();
+
+	public abstract double getStandardDeviationDuration();
+
+	public abstract int getActiveCount();
+
+	public abstract long getActiveCountLong();
+
+	public abstract Statistics getDuration();
+
+}

spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java

@@ -18,10 +18,8 @@ package org.springframework.integration.handler.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.MetricsContext;
 import org.springframework.integration.support.management.Statistics;
 import org.springframework.messaging.Message;
 
@@ -30,9 +28,7 @@ import org.springframework.messaging.Message;
  * @author Gary Russell
  * @since 2.0
  */
-public class SimpleMessageHandlerMetrics {
-
-	private static final Log logger = LogFactory.getLog(SimpleMessageHandlerMetrics.class);
+public class DefaultMessageHandlerMetrics extends AbstractMessageHandlerMetrics {
 
 	private static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10;
 
@@ -45,47 +41,48 @@ public class SimpleMessageHandlerMetrics {
 
 	private final ExponentialMovingAverage duration = new ExponentialMovingAverage(DEFAULT_MOVING_AVERAGE_WINDOW);
 
-	private volatile String name;
-
-	private volatile boolean fullStatsEnabled;
-
-	public void setName(String name) {
-		this.name = name;
+	public DefaultMessageHandlerMetrics() {
+		super(null);
 	}
 
-	public void setFullStatsEnabled(boolean fullStatsEnabled) {
-		this.fullStatsEnabled = fullStatsEnabled;
+	public DefaultMessageHandlerMetrics(String name) {
+		super(name);
 	}
 
-	public long beforeHandle(Message<?> message) throws Exception {
+
+	@Override
+	public MetricsContext beforeHandle(Message<?> message) {
 		if (logger.isTraceEnabled()) {
 			logger.trace("messageHandler(" + this.name + ") message(" + message + ") :");
 		}
 		long start = 0;
-		if (this.fullStatsEnabled) {
-			start = System.currentTimeMillis();
+		if (isFullStatsEnabled()) {
+			start = System.nanoTime();
 		}
 		this.handleCount.incrementAndGet();
 		this.activeCount.incrementAndGet();
-		return start;
+		return new DefaultHandlerMetricsContext(start);
 	}
 
-	public void afterHandle(long start, boolean success) {
+	@Override
+	public void afterHandle(MetricsContext context, boolean success) {
 		this.activeCount.decrementAndGet();
-		if (this.fullStatsEnabled && success) {
-			this.duration.append(System.currentTimeMillis() - start);
+		if (isFullStatsEnabled() && success) {
+			this.duration.appendNanos(System.nanoTime() - ((DefaultHandlerMetricsContext) context).start);
 		}
 		else if (!success) {
 			this.errorCount.incrementAndGet();
 		}
 	}
 
+	@Override
 	public synchronized void reset() {
 		this.duration.reset();
 		this.errorCount.set(0);
 		this.handleCount.set(0);
 	}
 
+	@Override
 	public long getHandleCountLong() {
 		if (logger.isTraceEnabled()) {
 			logger.trace("Getting Handle Count:" + this);
@@ -93,44 +90,64 @@ public class SimpleMessageHandlerMetrics {
 		return this.handleCount.get();
 	}
 
+	@Override
 	public int getHandleCount() {
 		return (int) getHandleCountLong();
 	}
 
+	@Override
 	public int getErrorCount() {
 		return (int) this.errorCount.get();
 	}
 
+	@Override
 	public long getErrorCountLong() {
 		return this.errorCount.get();
 	}
 
+	@Override
 	public double getMeanDuration() {
 		return this.duration.getMean();
 	}
 
+	@Override
 	public double getMinDuration() {
 		return this.duration.getMin();
 	}
 
+	@Override
 	public double getMaxDuration() {
 		return this.duration.getMax();
 	}
 
+	@Override
 	public double getStandardDeviationDuration() {
 		return this.duration.getStandardDeviation();
 	}
 
+	@Override
 	public int getActiveCount() {
 		return (int) this.activeCount.get();
 	}
 
+	@Override
 	public long getActiveCountLong() {
 		return this.activeCount.get();
 	}
 
+	@Override
 	public Statistics getDuration() {
 		return this.duration.getStatistics();
 	}
 
+	private static class DefaultHandlerMetricsContext implements MetricsContext {
+
+		private final long start;
+
+		public DefaultHandlerMetricsContext(long start) {
+			this.start = start;
+		}
+
+	}
+
 }

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

@@ -0,0 +1,27 @@
+/*
+ * 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.support.management;
+
+/**
+ * Marker interface for metrics.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public interface ConfigurableMetrics {
+
+}

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

@@ -0,0 +1,29 @@
+/*
+ * 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.support.management;
+
+/**
+ * Classes implementing this interface can accept a {@link ConfigurableMetrics}.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public interface ConfigurableMetricsAware<M extends ConfigurableMetrics> {
+
+	void configureMetrics(M metrics);
+
+}

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

@@ -66,9 +66,18 @@ public class ExponentialMovingAverage {
 	/**
 	 * Add a new measurement to the series.
 	 *
-	 * @param value the measurement to append
+	 * @param value the measurement to append (milliseconds)
 	 */
 	public synchronized void append(double value) {
+		appendNanos(value * 1000000);
+	}
+
+	/**
+	 * Add a new measurement to the series.
+	 *
+	 * @param value the measurement to append (nanoseconds)
+	 */
+	public synchronized void appendNanos(double value) {
 		if (value > max || count == 0) {
 			max = value;
 		}
@@ -76,7 +85,8 @@ public class ExponentialMovingAverage {
 			min = value;
 		}
 		sum = decay * sum + value;
-		sumSquares = decay * sumSquares + value * value;
+		double valueMillis = value / 1000000.;
+		sumSquares = decay * sumSquares + valueMillis * valueMillis;
 		weight = decay * weight + 1;
 		count++;//NOSONAR - false positive, we're synchronized
 	}
@@ -96,9 +106,16 @@ public class ExponentialMovingAverage {
 	}
 
 	/**
-	 * @return the mean value
+	 * @return the mean value (milliseconds)
 	 */
 	public double getMean() {
+		return weight > 0 ? sum / weight / 1000000. : 0.;
+	}
+
+	/**
+	 * @return the mean value (nanoseconds)
+	 */
+	public double getMeanNanos() {
 		return weight > 0 ? sum / weight : 0.;
 	}
 
@@ -112,24 +129,24 @@ public class ExponentialMovingAverage {
 	}
 
 	/**
-	 * @return the maximum value recorded (not weighted)
+	 * @return the maximum value recorded (not weighted, milliseconds).
 	 */
 	public double getMax() {
-		return max;
+		return max / 1000000.;
 	}
 
 	/**
-	 * @return the minimum value recorded (not weighted)
+	 * @return the minimum value recorded (not weighted, milliseconds).
 	 */
 	public double getMin() {
-		return min;
+		return min / 1000000.;
 	}
 
 	/**
 	 * @return summary statistics (count, mean, standard deviation etc.)
 	 */
 	public Statistics getStatistics() {
-		return new Statistics(count, min, max, getMean(), getStandardDeviation());
+		return new Statistics(count, getMin(), getMax(), getMean(), getStandardDeviation());
 	}
 
 	@Override

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

@@ -44,7 +44,7 @@ public class ExponentialMovingAverageRate {
 
 	private volatile double max;
 
-	private volatile long t0 = System.currentTimeMillis();
+	private volatile long t0 = System.nanoTime();
 
 	private final double lapse;
 
@@ -58,8 +58,8 @@ public class ExponentialMovingAverageRate {
 	 */
 	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
+		this.lapse = lapsePeriod > 0 ? 0.000000001 / lapsePeriod : 0; // convert to nanoseconds
+		this.period = period * 1000000000; // convert to nanoseconds
 	}
 
 
@@ -68,7 +68,7 @@ public class ExponentialMovingAverageRate {
 		max = 0;
 		weight = 0;
 		sum = 0;
-		t0 = System.currentTimeMillis();
+		t0 = System.nanoTime();
 		rates.reset();
 	}
 
@@ -76,15 +76,16 @@ public class ExponentialMovingAverageRate {
 	 * Add a new event to the series.
 	 */
 	public synchronized void increment() {
-		long t = System.currentTimeMillis();
-		double value = t > t0 ? (t - t0) / period : 0;
+		long t = System.nanoTime();
+		double delta = t - t0;
+		double value = delta > 0 ? delta / period : 0;
 		if (value > max || getCount() == 0) {
 			max = value;
 		}
 		if (value < min || getCount() == 0) {
 			min = value;
 		}
-		double alpha = Math.exp((t0 - t) * lapse);
+		double alpha = Math.exp(-delta * lapse);
 		t0 = t;
 		sum = alpha * sum + value;
 		weight = alpha * weight + 1;
@@ -110,7 +111,7 @@ public class ExponentialMovingAverageRate {
 	 * @return the time in seconds since the last measurement
 	 */
 	public double getTimeSinceLastMeasurement() {
-		return (System.currentTimeMillis() - t0) / 1000.;
+		return (System.nanoTime() - t0) / 1000000000.;
 	}
 
 	/**
@@ -121,9 +122,9 @@ public class ExponentialMovingAverageRate {
 		if (count == 0) {
 			return 0;
 		}
-		long t = System.currentTimeMillis();
-		double value = t > t0 ? (t - t0) / period : 0;
-		return count / (count / rates.getMean() + value);
+		long delta = System.nanoTime() - t0;
+		double value = delta > 0 ? delta / period : 0;
+		return count / (count / rates.getMeanNanos() + value);
 	}
 
 	/**

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

@@ -36,7 +36,7 @@ public class ExponentialMovingAverageRatio {
 
 	private volatile double sum;
 
-	private volatile long t0 = System.currentTimeMillis();
+	private volatile long t0 = System.nanoTime();
 
 	private final double lapse;
 
@@ -57,12 +57,12 @@ public class ExponentialMovingAverageRatio {
 	 * Add a new event with successful outcome.
 	 */
 	public void success() {
-		append(1, System.currentTimeMillis());
+		append(1, System.nanoTime());
 	}
 
 	/**
 	 * Add a new event with successful outcome.
-	 * @param t The current timestamp.
+	 * @param t The current timestamp (System.nanoTime()).
 	 */
 	public void success(long t) {
 		append(1, t);
@@ -72,12 +72,12 @@ public class ExponentialMovingAverageRatio {
 	 * Add a new event with failed outcome.
 	 */
 	public void failure() {
-		append(0, System.currentTimeMillis());
+		append(0, System.nanoTime());
 	}
 
 	/**
 	 * Add a new event with failed outcome.
-	 * @param t the current timestamp.
+	 * @param t the current timestamp (System.nanoTime()).
 	 */
 	public void failure(long t) {
 		append(0, t);
@@ -86,16 +86,16 @@ public class ExponentialMovingAverageRatio {
 	public synchronized void reset() {
 		weight = 0;
 		sum = 0;
-		t0 = System.currentTimeMillis();
+		t0 = System.nanoTime();
 		cumulative.reset();
 	}
 
 	private synchronized void append(int value, long t) {
-		double alpha = Math.exp((t0 - t) * lapse);
+		double alpha = Math.exp((t0 - t) / 1000000. * lapse);
 		t0 = t;
 		sum = alpha * sum + value;
 		weight = alpha * weight + 1;
-		cumulative.append(sum / weight);
+		cumulative.appendNanos(sum / weight);
 	}
 
 	/**
@@ -116,7 +116,7 @@ public class ExponentialMovingAverageRatio {
 	 * @return the time in seconds since the last measurement
 	 */
 	public double getTimeSinceLastMeasurement() {
-		return (System.currentTimeMillis() - t0) / 1000.;
+		return (System.nanoTime() - t0) / 1000000000.;
 	}
 
 	/**
@@ -128,9 +128,9 @@ public class ExponentialMovingAverageRatio {
 			// 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;
+		long t = System.nanoTime();
+		double alpha = Math.exp((t0 - t) / 1000000. * lapse);
+		return alpha * cumulative.getMeanNanos() + 1 - alpha;
 	}
 
 	/**

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

@@ -0,0 +1,28 @@
+/*
+ * 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.support.management;
+
+/**
+ * Interface representing an opaque object containing state between initiating an
+ * event and concluding it.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public interface MetricsContext {
+
+}

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

@@ -1,4 +1,4 @@
 /**
- * Provides classes related to management.
+ * Provides classes related to management support.
  */
 package org.springframework.integration.support.management;

spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java

@@ -19,7 +19,6 @@ package org.springframework.integration.jms;
 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.jms.core.JmsTemplate;
 import org.springframework.messaging.Message;
@@ -46,16 +45,6 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh
 		this.messageSelector = messageSelector;
 	}
 
-	@Override
-	protected void initMetrics() {
-		setChannelMetrics(new ChannelReceiveMetrics(this.getComponentName()));
-	}
-
-	@Override
-	protected ChannelReceiveMetrics getMetrics() {
-		return (ChannelReceiveMetrics) super.getMetrics();
-	}
-
 	@Override
 	public int getReceiveCount() {
 		return getMetrics().getReceiveCount();
@@ -81,6 +70,7 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh
 		ChannelInterceptorList interceptorList = getInterceptors();
 		Deque<ChannelInterceptor> interceptorStack = null;
 		boolean counted = false;
+		boolean countsEnabled = isCountsEnabled();
 		try {
 			if (interceptorList.getInterceptors().size() > 0) {
 				interceptorStack = new ArrayDeque<ChannelInterceptor>();
@@ -100,7 +90,7 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh
 			if (object == null) {
 				return null;
 			}
-			if (isCountsEnabled()) {
+			if (countsEnabled) {
 				getMetrics().afterReceive();
 				counted = true;
 			}
@@ -118,7 +108,7 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh
 			return message;
 		}
 		catch (RuntimeException e) {
-			if (isCountsEnabled() && !counted) {
+			if (countsEnabled && !counted) {
 				getMetrics().afterError();
 			}
 			if (interceptorStack != null) {

spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java

@@ -60,6 +60,7 @@ public class MBeanExporterParser extends AbstractSingleBeanDefinitionParser {
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "managed-components", "componentNamePatterns");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "counts-enabled", "enabledCountsPatterns");
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "stats-enabled", "enabledStatsPatterns");
+		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "metrics-factory");
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "object-naming-strategy", "namingStrategy");
 
 		builder.addPropertyValue("server", mbeanServer);

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

@@ -0,0 +1,44 @@
+/*
+ * 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.monitor;
+
+import org.springframework.integration.channel.management.AbstractMessageChannelMetrics;
+import org.springframework.integration.channel.management.DefaultMessageChannelMetrics;
+import org.springframework.integration.handler.management.AbstractMessageHandlerMetrics;
+import org.springframework.integration.handler.management.DefaultMessageHandlerMetrics;
+
+
+
+/**
+ * Default implementation.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public class DefaultMetricsFactory implements MetricsFactory {
+
+	@Override
+	public AbstractMessageChannelMetrics createChannelMetrics(String name) {
+		return new DefaultMessageChannelMetrics(name);
+	}
+
+	@Override
+	public AbstractMessageHandlerMetrics createHandlerMetrics(String name) {
+		return new DefaultMessageHandlerMetrics(name);
+	}
+
+}

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

@@ -34,11 +34,8 @@ import javax.management.modelmbean.ModelMBean;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
-import org.springframework.aop.PointcutAdvisor;
 import org.springframework.aop.TargetSource;
 import org.springframework.aop.framework.Advised;
-import org.springframework.aop.framework.ProxyFactory;
-import org.springframework.aop.support.AopUtils;
 import org.springframework.beans.BeansException;
 import org.springframework.beans.annotation.AnnotationBeanUtils;
 import org.springframework.beans.factory.BeanFactory;
@@ -52,6 +49,7 @@ import org.springframework.context.Lifecycle;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.core.annotation.AnnotationUtils;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.channel.management.AbstractMessageChannelMetrics;
 import org.springframework.integration.channel.management.MessageChannelMetrics;
 import org.springframework.integration.channel.management.PollableChannelManagement;
 import org.springframework.integration.context.IntegrationContextUtils;
@@ -59,12 +57,14 @@ import org.springframework.integration.context.OrderlyShutdownCapable;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.endpoint.AbstractEndpoint;
 import org.springframework.integration.endpoint.management.MessageSourceMetrics;
-import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.handler.AbstractMessageProducingHandler;
+import org.springframework.integration.handler.management.AbstractMessageHandlerMetrics;
 import org.springframework.integration.handler.management.MessageHandlerMetrics;
 import org.springframework.integration.history.MessageHistoryConfigurer;
 import org.springframework.integration.history.TrackableComponent;
 import org.springframework.integration.router.MappingMessageRouterManagement;
 import org.springframework.integration.support.context.NamedComponent;
+import org.springframework.integration.support.management.ConfigurableMetricsAware;
 import org.springframework.integration.support.management.IntegrationManagedResource;
 import org.springframework.integration.support.management.Statistics;
 import org.springframework.jmx.export.MBeanExporter;
@@ -188,6 +188,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 
 	private StringValueResolver embeddedValueResolver;
 
+	private MetricsFactory metricsFactory = new DefaultMetricsFactory();
+
 
 	public IntegrationMBeanExporter() {
 		super();
@@ -294,6 +296,15 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 		this.embeddedValueResolver = resolver;
 	}
 
+	/**
+	 * Set a metrics factory.
+	 * @param metricsFactory the factory.
+	 * @since 4.2
+	 */
+	public void setMetricsFactory(MetricsFactory metricsFactory) {
+		this.metricsFactory = metricsFactory;
+	}
+
 	@Override
 	public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
 
@@ -333,7 +344,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 
 		if (bean instanceof MessageProducer && bean instanceof Lifecycle) {
 			Lifecycle target = (Lifecycle) extractTarget(bean);
-			if (!(target instanceof AbstractReplyProducingMessageHandler)) { // TODO: change to AMPMH
+			if (!(target instanceof AbstractMessageProducingHandler)) {
 				this.inboundLifecycleMessageProducers.add(target);
 			}
 		}
@@ -749,6 +760,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 		}
 	}
 
+	@SuppressWarnings("unchecked")
 	private void registerChannels() {
 		for (MessageChannelMetrics monitor : channels) {
 			String name = ((NamedComponent) monitor).getComponentName();
@@ -763,6 +775,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 				if (name != null) {
 					channelsByName.put(name, monitor);
 				}
+				AbstractMessageChannelMetrics metrics = this.metricsFactory.createChannelMetrics(name);
+				Assert.state(metrics != null, "'metrics' must not be null");
 				Boolean enabled = smartMatch(this.enabledCountsPatterns, name);
 				if (enabled != null) {
 					monitor.enableCounts(enabled);
@@ -770,12 +784,17 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 				enabled = smartMatch(this.enabledStatsPatterns, name);
 				if (enabled != null) {
 					monitor.enableStats(enabled);
+					metrics.setFullStatsEnabled(enabled);
+				}
+				if (monitor instanceof ConfigurableMetricsAware) {
+					((ConfigurableMetricsAware<AbstractMessageChannelMetrics>) monitor).configureMetrics(metrics);
 				}
 				registerBeanNameOrInstance(monitor, beanKey);
 			}
 		}
 	}
 
+	@SuppressWarnings("unchecked")
 	private void registerHandlers() {
 		for (MessageHandlerMetrics handler : handlers) {
 			MessageHandlerMetrics monitor = enhanceHandlerMonitor(handler);
@@ -790,6 +809,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 				if (name != null) {
 					handlersByName.put(name, monitor);
 				}
+				AbstractMessageHandlerMetrics metrics = this.metricsFactory.createHandlerMetrics(name);
+				Assert.state(metrics != null, "'metrics' must not be null");
 				Boolean enabled = smartMatch(this.enabledCountsPatterns, name);
 				if (enabled != null) {
 					monitor.enableCounts(enabled);
@@ -797,6 +818,10 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 				enabled = smartMatch(this.enabledStatsPatterns, name);
 				if (enabled != null) {
 					monitor.enableStats(enabled);
+					metrics.setFullStatsEnabled(enabled);
+				}
+				if (monitor instanceof ConfigurableMetricsAware) {
+					((ConfigurableMetricsAware<AbstractMessageHandlerMetrics>) monitor).configureMetrics(metrics);
 				}
 				registerBeanNameOrInstance(monitor, beanKey);
 			}
@@ -915,27 +940,6 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
 		}
 	}
 
-	private Object applyAdvice(Object bean, PointcutAdvisor advisor, ClassLoader beanClassLoader) {
-		Class<?> targetClass = AopUtils.getTargetClass(bean);
-		if (AopUtils.canApply(advisor.getPointcut(), targetClass)) {
-			if (bean instanceof Advised) {
-				((Advised) bean).addAdvisor(advisor);
-				return bean;
-			}
-			else {
-				ProxyFactory proxyFactory = new ProxyFactory(bean);
-				proxyFactory.addAdvisor(advisor);
-				/**
-				 * N.B. it's not a good idea to use proxyFactory.setProxyTargetClass(true) here because it forces all
-				 * the integration components to be cglib proxyable (i.e. have a default constructor etc.), which they
-				 * are not in general (usually for good reason).
-				 */
-				return proxyFactory.getProxy(beanClassLoader);
-			}
-		}
-		return bean;
-	}
-
 	private String getChannelBeanKey(String channel) {
 		String name = "" + channel;
 		if (name.startsWith("org.springframework.integration")) {

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

@@ -0,0 +1,45 @@
+/*
+ * 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.monitor;
+
+import org.springframework.integration.channel.management.AbstractMessageChannelMetrics;
+import org.springframework.integration.handler.management.AbstractMessageHandlerMetrics;
+
+/**
+ * Factories implementing this interface provide metric objects for message channels and
+ * message handlers.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public interface MetricsFactory {
+
+	/**
+	 * Factory method to create an {@link AbstractMessageChannelMetrics}.
+	 * @param name the name.
+	 * @return the metrics.
+	 */
+	AbstractMessageChannelMetrics createChannelMetrics(String name);
+
+	/**
+	 * Factory method to create an {@link AbstractMessageHandlerMetrics}.
+	 * @param name the name.
+	 * @return the metrics.
+	 */
+	AbstractMessageHandlerMetrics createHandlerMetrics(String name);
+
+}

spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-4.2.xsd

@@ -221,6 +221,19 @@
 							</xsd:documentation>
 						</xsd:annotation>
 					</xsd:attribute>
+					<xsd:attribute name="metrics-factory" use="optional">
+						<xsd:annotation>
+							<xsd:appinfo>
+								<tool:annotation kind="ref">
+									<tool:expected-type type="org.springframework.integration.monitor.MetricsFactory" />
+								</tool:annotation>
+							</xsd:appinfo>
+							<xsd:documentation>
+								A MetricsFactory responsible for creating objects that maintain metrics for message
+								channels and message handlers.
+							</xsd:documentation>
+						</xsd:annotation>
+					</xsd:attribute>
 					<xsd:attribute name="object-naming-strategy" use="optional">
 						<xsd:annotation>
 							<xsd:appinfo>

spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests-context.xml

@@ -21,7 +21,8 @@
 					  object-name-static-properties="appProperties"
 					  object-naming-strategy="keyNamer"
 					  counts-enabled="foo, !baz, ba*"
-					  stats-enabled="fiz, buz" />
+					  stats-enabled="fiz, buz" 
+					  metrics-factory="mf" />
 					  
 	<util:properties id="appProperties">
 		<prop key="foo">foo</prop>
@@ -42,4 +43,6 @@
 
 	<si:channel id="buz" />
 
+	<bean id="mf" class="org.springframework.integration.monitor.DefaultMetricsFactory" />
+
 </beans>

spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests.java

@@ -33,6 +33,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationContext;
 import org.springframework.integration.channel.management.MessageChannelMetrics;
 import org.springframework.integration.monitor.IntegrationMBeanExporter;
+import org.springframework.integration.monitor.MetricsFactory;
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.ContextConfiguration;
@@ -82,6 +83,8 @@ public class MBeanExporterParserTests {
 		metrics = context.getBean("buz", MessageChannelMetrics.class);
 		assertTrue(metrics.isCountsEnabled());
 		assertTrue(metrics.isStatsEnabled());
+		MetricsFactory factory = context.getBean(MetricsFactory.class);
+		assertSame(factory, TestUtils.getPropertyValue(exporter, "metricsFactory"));
 		exporter.destroy();
 	}