INT-1503: *Monitor->*Metrics

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

@@ -0,0 +1,200 @@
+/*
+ * Copyright 2009-2010 the original author or authors.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+package org.springframework.integration.monitor;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
+import org.springframework.jmx.export.annotation.ManagedMetric;
+import org.springframework.jmx.export.annotation.ManagedResource;
+import org.springframework.jmx.support.MetricType;
+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
+ */
+@ManagedResource
+public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMetrics {
+
+	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 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 AtomicInteger sendCount = new AtomicInteger();
+
+	private final AtomicInteger sendErrorCount = new AtomicInteger();
+
+	private final String name;
+
+	public DirectChannelMetrics(String name) {
+		this.name = name;
+	}
+
+	public void destroy() {
+		if (logger.isDebugEnabled()) {
+			logger.debug(sendDuration);
+		}
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public Object invoke(MethodInvocation invocation) throws Throwable {
+		String method = invocation.getMethod().getName();
+		MessageChannel channel = (MessageChannel) invocation.getThis();
+		return doInvoke(invocation, method, channel);
+	}
+
+	protected Object doInvoke(MethodInvocation invocation, String method, MessageChannel channel) throws Throwable {
+		if ("send".equals(method)) {
+			Message<?> message = (Message<?>) invocation.getArguments()[0];
+			return monitorSend(invocation, channel, message);
+		}
+		return invocation.proceed();
+	}
+
+	private Object monitorSend(MethodInvocation invocation, MessageChannel channel, Message<?> message)
+			throws Throwable {
+
+		if (logger.isTraceEnabled()) {
+			logger.trace("Recording send on channel(" + channel + ") : message(" + message + ")");
+		}
+
+		final StopWatch timer = new StopWatch(channel + ".send:execution");
+
+		try {
+			timer.start();
+
+			sendCount.incrementAndGet();
+			sendRate.increment();
+
+			Object result = invocation.proceed();
+
+			timer.stop();
+			if ((Boolean)result) {
+				sendSuccessRatio.success();
+				sendDuration.append(timer.getTotalTimeSeconds());
+			} else {
+				sendSuccessRatio.failure();
+				sendErrorCount.incrementAndGet();
+				sendErrorRate.increment();
+			}
+			return result;
+
+		}
+		catch (Throwable e) {
+			sendSuccessRatio.failure();
+			sendErrorCount.incrementAndGet();
+			sendErrorRate.increment();
+			throw e;
+		}
+		finally {
+			if (logger.isTraceEnabled()) {
+				logger.trace(timer);
+			}
+		}
+	}
+
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Sends")
+	public int getSendCount() {
+		return sendCount.get();
+	}
+
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Send Errors")
+	public int getSendErrorCount() {
+		return sendErrorCount.get();
+	}
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Time Since Last Send in Seconds")
+	public double getTimeSinceLastSend() {
+		return sendRate.getTimeSinceLastMeasurement();
+	}
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Rate per Second")
+	public double getMeanSendRate() {
+		return sendRate.getMean();
+	}
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Error Rate per Second")
+	public double getMeanErrorRate() {
+		return sendErrorRate.getMean();
+	}
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Mean Channel Error Ratio per Minute")
+	public double getMeanErrorRatio() {
+		return 1 - sendSuccessRatio.getMean();
+	}
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Mean Duration")
+	public double getMeanSendDuration() {
+		return sendDuration.getMean();
+	}
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Min Duration")
+	public double getMinSendDuration() {
+		return sendDuration.getMin();
+	}
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Max Duration")
+	public double getMaxSendDuration() {
+		return sendDuration.getMax();
+	}
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Standard Deviation Duration")
+	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-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMetrics.java

@@ -0,0 +1,99 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.monitor;
+
+import org.springframework.context.Lifecycle;
+import org.springframework.jmx.export.annotation.ManagedAttribute;
+import org.springframework.jmx.export.annotation.ManagedOperation;
+import org.springframework.jmx.export.annotation.ManagedResource;
+
+/**
+ * A {@link MessageHandlerMetrics} that exposes in addition the {@link Lifecycle} interface. The lifecycle methods can
+ * be used to stop and start polling endpoints, for instance, in a live system.
+ * 
+ * @author Dave Syer
+ * 
+ * @since 2.0
+ * 
+ */
+@ManagedResource
+public class LifecycleMessageHandlerMetrics implements MessageHandlerMetrics, Lifecycle {
+
+	private final Lifecycle lifecycle;
+
+	private final MessageHandlerMetrics delegate;
+
+	public LifecycleMessageHandlerMetrics(Lifecycle lifecycle, MessageHandlerMetrics delegate) {
+		this.lifecycle = lifecycle;
+		this.delegate = delegate;
+	}
+
+	@ManagedAttribute
+	public boolean isRunning() {
+		return lifecycle.isRunning();
+	}
+
+	@ManagedOperation
+	public void start() {
+		lifecycle.start();
+	}
+
+	@ManagedOperation
+	public void stop() {
+		lifecycle.stop();
+	}
+
+	public int getErrorCount() {
+		return delegate.getErrorCount();
+	}
+
+	public int getHandleCount() {
+		return delegate.getHandleCount();
+	}
+
+	public double getMaxDuration() {
+		return delegate.getMaxDuration();
+	}
+
+	public double getMeanDuration() {
+		return delegate.getMeanDuration();
+	}
+
+	public double getMinDuration() {
+		return delegate.getMinDuration();
+	}
+
+	public double getStandardDeviationDuration() {
+		return delegate.getStandardDeviationDuration();
+	}
+
+	public Statistics getDuration() {
+		return delegate.getDuration();
+	}
+
+	public String getName() {
+		return delegate.getName();
+	}
+
+	public String getSource() {
+		return delegate.getSource();
+	}
+
+	public int getActiveCount() {
+		return delegate.getActiveCount();
+	}
+
+}

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

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.monitor;
+
+import org.springframework.context.Lifecycle;
+import org.springframework.jmx.export.annotation.ManagedAttribute;
+import org.springframework.jmx.export.annotation.ManagedOperation;
+import org.springframework.jmx.export.annotation.ManagedResource;
+
+/**
+ * A {@link MessageSourceMetrics} that exposes in addition the {@link Lifecycle} interface. The lifecycle methods can
+ * be used to stop and start polling endpoints, for instance, in a live system.
+ * 
+ * @author Dave Syer
+ * 
+ * @since 2.0
+ * 
+ */
+@ManagedResource
+public class LifecycleMessageSourceMetrics implements MessageSourceMetrics, Lifecycle {
+
+	private final Lifecycle lifecycle;
+
+	private final MessageSourceMetrics delegate;
+
+	public LifecycleMessageSourceMetrics(Lifecycle lifecycle, MessageSourceMetrics delegate) {
+		this.lifecycle = lifecycle;
+		this.delegate = delegate;
+	}
+
+	@ManagedAttribute
+	public boolean isRunning() {
+		return lifecycle.isRunning();
+	}
+
+	@ManagedOperation
+	public void start() {
+		lifecycle.start();
+	}
+
+	@ManagedOperation
+	public void stop() {
+		lifecycle.stop();
+	}
+
+	public String getName() {
+		return delegate.getName();
+	}
+
+	public String getSource() {
+		return delegate.getSource();
+	}
+
+	/**
+	 * @return
+	 * @see org.springframework.integration.monitor.MessageSourceMetrics#getMessageCount()
+	 */
+	public int getMessageCount() {
+		return delegate.getMessageCount();
+	}
+
+}

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

@@ -0,0 +1,107 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.monitor;
+
+import org.springframework.jmx.export.annotation.ManagedMetric;
+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
+ * 
+ * @since 2.0
+ * 
+ */
+public interface MessageChannelMetrics {
+
+	/**
+	 * @return the number of successful sends
+	 */
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Sends")
+	int getSendCount();
+
+	/**
+	 * @return the number of failed sends (either throwing an exception or rejected by the channel)
+	 */
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Send Errors")
+	int getSendErrorCount();
+
+	/**
+	 * @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")
+	double getMeanSendDuration();
+
+	/**
+	 * @return the minimum send duration (milliseconds) since startup
+	 */
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Min Duration")
+	double getMinSendDuration();
+
+	/**
+	 * @return the maximum send duration (milliseconds) since startup
+	 */
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Max Duration")
+	double getMaxSendDuration();
+
+	/**
+	 * @return the standard deviation send duration (milliseconds)
+	 */
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Standard Deviation Duration")
+	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-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMetrics.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.monitor;
+
+import org.springframework.jmx.export.annotation.ManagedMetric;
+import org.springframework.jmx.support.MetricType;
+
+/**
+ * @author Dave Syer
+ *
+ * @since 2.0
+ */
+public interface MessageHandlerMetrics {
+
+	/**
+	 * @return the number of successful handler calls
+	 */
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "Handler Execution Count", description = "rate=1h")
+	int getHandleCount();
+
+	/**
+	 * @return the number of failed handler calls
+	 */
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "Handler Error Count", description = "rate=1h")
+	int getErrorCount();
+
+	/**
+	 * @return the maximum handler duration (milliseconds)
+	 */
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Mean Duration")
+	double getMeanDuration();
+
+	/**
+	 * @return the minimum handler duration (milliseconds)
+	 */
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Min Duration")
+	double getMinDuration();
+
+	/**
+	 * @return the standard deviation handler duration (milliseconds)
+	 */
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Max Duration")
+	double getMaxDuration();
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Standard Deviation Duration")
+	double getStandardDeviationDuration();
+	
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Active Status")
+	int getActiveCount();
+	
+	/**
+	 * @return summary statistics about the handler duration (milliseconds)
+	 */
+	Statistics getDuration();
+
+	String getName();
+
+	String getSource();
+
+}

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

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package org.springframework.integration.monitor;
+
+import org.springframework.jmx.export.annotation.ManagedMetric;
+import org.springframework.jmx.support.MetricType;
+
+/**
+ * @author Dave Syer
+ *
+ * @since 2.0
+ */
+public interface MessageSourceMetrics {
+
+	/**
+	 * @return the number of successful handler calls
+	 */
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "Message Source Message Count", description = "rate=1h")
+	int getMessageCount();
+
+	String getName();
+
+	String getSource();
+
+}

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

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.monitor;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.aopalliance.intercept.MethodInvocation;
+import org.springframework.integration.MessageChannel;
+import org.springframework.jmx.export.annotation.ManagedMetric;
+import org.springframework.jmx.support.MetricType;
+
+/**
+ * @author Dave Syer
+ * 
+ * @since 2.0
+ * 
+ */
+public class PollableChannelMetrics extends DirectChannelMetrics {
+
+	private final AtomicInteger receiveCount = new AtomicInteger();
+
+	private final AtomicInteger receiveErrorCount = new AtomicInteger();
+
+	/**
+	 * @param name
+	 */
+	public PollableChannelMetrics(String name) {
+		super(name);
+	}
+
+	@Override
+	protected Object doInvoke(MethodInvocation invocation, String method, MessageChannel channel) throws Throwable {
+		if ("receive".equals(method)) {
+			return monitorReceive(invocation, channel);
+		}
+		return super.doInvoke(invocation, method, channel);
+	}
+
+	private Object monitorReceive(MethodInvocation invocation, MessageChannel channel) throws Throwable {
+		if (logger.isTraceEnabled()) {
+			logger.trace("Recording receive on channel(" + channel + ") ");
+		}
+		try {
+			Object object = invocation.proceed();
+			if (object!=null) {
+				receiveCount.incrementAndGet();
+			}
+			return object;
+
+		}
+		catch (Throwable e) {
+			receiveErrorCount.incrementAndGet();
+			throw e;
+		}
+	}
+
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receives")
+	public int getReceiveCount() {
+		return receiveCount.get();
+	}
+
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Errors")
+	public int getReceiveErrorCount() {
+		return receiveErrorCount.get();
+	}
+
+	@Override
+	public String toString() {
+		return String.format("MessageChannelMonitor: [name=%s, sends=%d, receives=%d]", getName(), getSendCount(),
+				receiveCount.get());
+	}
+
+}

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

@@ -0,0 +1,50 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.monitor;
+
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.jmx.export.annotation.ManagedMetric;
+import org.springframework.jmx.support.MetricType;
+
+/**
+ * @author Dave Syer
+ * 
+ * @since 2.0
+ * 
+ */
+public class QueueChannelMetrics extends PollableChannelMetrics {
+
+	private final QueueChannel channel;
+
+	/**
+	 * @param name
+	 */
+	public QueueChannelMetrics(QueueChannel channel, String name) {
+		super(name);
+		this.channel = channel;
+	}
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "QueueChannel Queue Size")
+	public int getQueueSize() {
+		return channel.getQueueSize();
+	}
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "QueueChannel Remaining Capacity")
+	public int getRemainingCapacity() {
+		return channel.getRemainingCapacity();
+	}
+
+}

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

@@ -0,0 +1,175 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.monitor;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageDeliveryException;
+import org.springframework.integration.MessageHandlingException;
+import org.springframework.integration.MessageRejectedException;
+import org.springframework.integration.core.MessageHandler;
+import org.springframework.jmx.export.annotation.ManagedMetric;
+import org.springframework.jmx.export.annotation.ManagedResource;
+import org.springframework.jmx.support.MetricType;
+import org.springframework.util.StopWatch;
+
+/**
+ * @author Dave Syer
+ * 
+ * @since 2.0
+ *
+ */
+@ManagedResource
+public class SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHandlerMetrics {
+
+	private static final Log logger = LogFactory.getLog(SimpleMessageHandlerMetrics.class);
+
+	private static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10;
+
+	private final MessageHandler handler;
+
+	private final AtomicInteger activeCount = new AtomicInteger();
+
+	private final AtomicInteger handleCount = new AtomicInteger();
+
+	private final AtomicInteger errorCount = new AtomicInteger();
+
+	private final ExponentialMovingAverage duration = new ExponentialMovingAverage(
+			DEFAULT_MOVING_AVERAGE_WINDOW);
+
+	private String name;
+
+	private String source;
+
+	public SimpleMessageHandlerMetrics(MessageHandler handler) {
+		this.handler = handler;
+	}
+
+	public void setName(String name) {
+		this.name = name;
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public void setSource(String source) {
+		this.source = source;
+	}
+
+	public String getSource() {
+		return this.source;
+	}
+
+	public MessageHandler getMessageHandler() {
+		return handler;
+	}
+
+	public Object invoke(MethodInvocation invocation) throws Throwable {
+		String method = invocation.getMethod().getName();
+		if ("handleMessage".equals(method)) {
+			Message<?> message = (Message<?>) invocation.getArguments()[0];
+			handleMessage(message);
+			return null;
+		}
+		return invocation.proceed();
+	}
+
+	private void handleMessage(Message<?> message) throws MessageRejectedException, MessageHandlingException,
+			MessageDeliveryException {
+		if (logger.isTraceEnabled()) {
+			logger.trace("messageHandler(" + handler + ") message(" + message + ") :");
+		}
+
+		String name = this.name;
+		if (name == null) {
+			name = handler.toString();
+		}
+		StopWatch timer = new StopWatch(name + ".handle:execution");
+
+		try {
+			timer.start();
+			handleCount.incrementAndGet();
+			activeCount.incrementAndGet();
+
+			handler.handleMessage(message);
+
+			timer.stop();
+			duration.append(timer.getTotalTimeSeconds());
+		} catch (RuntimeException e) {
+			errorCount.incrementAndGet();
+			throw e;
+		} catch (Error e) {
+			errorCount.incrementAndGet();
+			throw e;
+		} finally {
+			activeCount.decrementAndGet();
+		}
+	}
+
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "Handler Execution Count", description = "rate=1h")
+	public int getHandleCount() {
+		if (logger.isTraceEnabled()) {
+			logger.trace("Getting Handle Count:" + this);
+		}
+		return handleCount.get();
+	}
+
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "Handler Error Count", description = "rate=1h")
+	public int getErrorCount() {
+		return errorCount.get();
+	}
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Mean Duration")
+	public double getMeanDuration() {
+		return duration.getMean();
+	}
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Min Duration")
+	public double getMinDuration() {
+		return duration.getMin();
+	}
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Max Duration")
+	public double getMaxDuration() {
+		return duration.getMax();
+	}
+
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Standard Deviation Duration")
+	public double getStandardDeviationDuration() {
+		return duration.getStandardDeviation();
+	}
+	
+	@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Active Count")
+	public int getActiveCount() {
+		return activeCount.get();
+	}
+	
+	public Statistics getDuration() {
+		return duration.getStatistics();
+	}
+
+	@Override
+	public String toString() {
+		return String.format("MessageHandlerMonitor: [name=%s, source=%s, duration=%s]", name, source, duration);
+	}
+
+}

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

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.monitor;
+
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.core.MessageHandler;
+import org.springframework.integration.core.MessageProducer;
+import org.springframework.jmx.export.annotation.ManagedResource;
+
+/**
+ * @author Oleg Zhurakousky
+ * @since 2.0
+ *
+ */
+@ManagedResource
+public class SimpleMessageProducingHandlerMetrics extends SimpleMessageHandlerMetrics implements MessageProducer {
+	
+	public SimpleMessageProducingHandlerMetrics(MessageHandler handler) {
+		super(handler);
+	}
+
+	public void setOutputChannel(MessageChannel outputChannel) {
+		((MessageProducer)this.getMessageHandler()).setOutputChannel(outputChannel);
+	}
+}

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

@@ -0,0 +1,80 @@
+/*
+ * Copyright 2002-2010 the original author or authors.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ * 
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package org.springframework.integration.monitor;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+import org.springframework.integration.core.MessageSource;
+
+/**
+ * @author Dave Syer
+ *
+ * @since 2.0
+ */
+public class SimpleMessageSourceMetrics implements MethodInterceptor, MessageSourceMetrics {
+
+	private final AtomicInteger messageCount = new AtomicInteger();
+
+	private final MessageSource<?> messageSource;
+
+	private String source;
+
+	private String name;
+
+	public SimpleMessageSourceMetrics(MessageSource<?> messageSource) {
+		this.messageSource = messageSource;	
+	}
+	
+
+	public void setName(String name) {
+		this.name = name;
+	}
+
+	public String getName() {
+		return name;
+	}
+
+	public void setSource(String source) {
+		this.source = source;
+	}
+
+	public String getSource() {
+		return this.source;
+	}
+
+	public MessageSource<?> getMessageSource() {
+		return messageSource;
+	}
+
+	public int getMessageCount() {
+		return messageCount.get();
+	}
+
+	public Object invoke(MethodInvocation invocation) throws Throwable {
+		String method = invocation.getMethod().getName();
+		Object result = invocation.proceed();
+		if ("receive".equals(method) && result!=null) {
+			messageCount.incrementAndGet();
+		}
+		return result;
+	}
+
+	@Override
+	public String toString() {
+		return String.format("MessageSourceMonitor: [name=%s, source=%s, count=%d]", name, source, messageCount.get());
+	}
+
+}