diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningMessageProducer.java b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningMessageProducer.java index 99680fe493..ddb4d30020 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningMessageProducer.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningMessageProducer.java @@ -104,8 +104,7 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport @Override public String getComponentType() { - // TODO: provide header: ("transport", "jmx"); - return "notification-listener"; + return "jmx:notification-listening-channel-adapter"; } /** diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/OperationInvokingMessageHandler.java b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/OperationInvokingMessageHandler.java index adf9c02b7d..799f510d05 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/OperationInvokingMessageHandler.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/OperationInvokingMessageHandler.java @@ -208,7 +208,6 @@ public class OperationInvokingMessageHandler extends AbstractReplyProducingMessa return map; } - @SuppressWarnings("rawtypes") private Map createParameterMapFromList(List parameters) { Map map = new HashMap(); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/OperationInvokingOutboundGatewayParser.java b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/OperationInvokingOutboundGatewayParser.java index 086d9e1825..7c78355656 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/OperationInvokingOutboundGatewayParser.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/OperationInvokingOutboundGatewayParser.java @@ -13,13 +13,15 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.jmx.config; +import org.w3c.dom.Element; + import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.AbstractConsumerEndpointParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; -import org.w3c.dom.Element; /** * @author Oleg Zhurakousky @@ -33,10 +35,9 @@ public class OperationInvokingOutboundGatewayParser extends AbstractConsumerEndp } @Override - protected BeanDefinitionBuilder parseHandler(Element element, - ParserContext parserContext) { + protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition( - "org.springframework.integration.jmx.OperationInvokingMessageHandler"); + "org.springframework.integration.jmx.OperationInvokingMessageHandler"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "server"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "object-name"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "operation-name"); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DirectChannelMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DirectChannelMetrics.java index 8c4f780dd1..520fb81b47 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DirectChannelMetrics.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DirectChannelMetrics.java @@ -10,6 +10,7 @@ * 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; @@ -18,6 +19,7 @@ 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.ManagedResource; @@ -29,6 +31,7 @@ import org.springframework.util.StopWatch; * * @author Dave Syer * @author Helena Edelson + * @since 2.0 */ @ManagedResource public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMetrics { @@ -41,6 +44,7 @@ public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMe public static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10; + private ExponentialMovingAverage sendDuration = new ExponentialMovingAverage( DEFAULT_MOVING_AVERAGE_WINDOW); @@ -59,10 +63,12 @@ public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMe private final String name; + public DirectChannelMetrics(String name) { this.name = name; } + public void destroy() { if (logger.isDebugEnabled()) { logger.debug(sendDuration); @@ -87,15 +93,11 @@ public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMe return invocation.proceed(); } - private Object monitorSend(MethodInvocation invocation, MessageChannel channel, Message message) - throws Throwable { - + 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(); @@ -108,13 +110,13 @@ public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMe if ((Boolean)result) { sendSuccessRatio.success(); sendDuration.append(timer.getTotalTimeMillis()); - } else { + } + else { sendSuccessRatio.failure(); sendErrorCount.incrementAndGet(); sendErrorRate.increment(); } return result; - } catch (Throwable e) { sendSuccessRatio.failure(); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverage.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverage.java index 70fd1e725f..62a770c448 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverage.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverage.java @@ -10,6 +10,7 @@ * 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; /** @@ -21,7 +22,7 @@ package org.springframework.integration.monitor; * those trends can be approximately reflected. * * @author Dave Syer - * + * @since 2.0 */ public class ExponentialMovingAverage { @@ -39,6 +40,7 @@ public class ExponentialMovingAverage { private final double decay; + /** * Create a moving average accumulator with decay lapse window provided. Measurements older than this will have * smaller weight than 1/e. @@ -49,6 +51,7 @@ public class ExponentialMovingAverage { this.decay = 1 - 1. / window; } + public synchronized void reset() { weight = 0; sum = 0; @@ -64,10 +67,12 @@ public class ExponentialMovingAverage { * @param value the measurement to append */ public synchronized void append(double value) { - if (value > max || count == 0) + if (value > max || count == 0) { max = value; - if (value < min || count == 0) + } + if (value < min || count == 0) { min = value; + } sum = decay * sum + value; sumSquares = decay * sumSquares + value * value; weight = decay * weight + 1; diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRate.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRate.java index 3a4fabfd90..6e672e3624 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRate.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRate.java @@ -10,6 +10,7 @@ * 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; /** @@ -46,6 +47,7 @@ public class ExponentialMovingAverageRate { private final double period; + /** * @param period the period to base the rate measurement (in seconds) * @param lapsePeriod the exponential lapse rate for the rate average (in seconds) @@ -53,10 +55,11 @@ public class ExponentialMovingAverageRate { */ public ExponentialMovingAverageRate(double period, double lapsePeriod, int window) { rates = new ExponentialMovingAverage(10); - this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to millisecs - this.period = period * 1000; // convert to millisecs + this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to milliseconds + this.period = period * 1000; // convert to milliseconds } + public synchronized void reset() { min = 0; max = 0; @@ -70,7 +73,6 @@ 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; if (value > max || getCount() == 0) { @@ -84,7 +86,6 @@ public class ExponentialMovingAverageRate { sum = alpha * sum + value; weight = alpha * weight + 1; rates.append(sum > 0 ? weight / sum : 0); - } /** diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRatio.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRatio.java index f44efd6655..b257d23f73 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRatio.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRatio.java @@ -10,6 +10,7 @@ * 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; /** @@ -24,7 +25,7 @@ package org.springframework.integration.monitor; * the lapse window and i is the sequence number of the measurement. * * @author Dave Syer - * + * @since 2.0 */ public class ExponentialMovingAverageRatio { @@ -38,6 +39,7 @@ public class ExponentialMovingAverageRatio { private final ExponentialMovingAverage cumulative; + /** * @param lapsePeriod the exponential lapse rate for the rate average (in seconds) * @param window the exponential lapse window (number of measurements) @@ -47,6 +49,7 @@ public class ExponentialMovingAverageRatio { this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to millisecs } + /** * Add a new event with successful outcome. */ @@ -69,14 +72,12 @@ public class ExponentialMovingAverageRatio { } private synchronized void append(int value) { - long t = System.currentTimeMillis(); double alpha = Math.exp((t0 - t) * lapse); t0 = t; sum = alpha * sum + value; weight = alpha * weight + 1; cumulative.append(sum / weight); - } /** diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java index 8110ec54df..e91d080869 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java @@ -1,5 +1,5 @@ /* - * Copyright 2009-2010 the original author or authors. + * 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 @@ -10,6 +10,7 @@ * 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.lang.reflect.Field; @@ -82,13 +83,14 @@ import org.springframework.util.ReflectionUtils; * @author Oleg Zhurakousky */ @ManagedResource -public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostProcessor, BeanFactoryAware, - BeanClassLoaderAware, SmartLifecycle { +public class IntegrationMBeanExporter extends MBeanExporter + implements BeanPostProcessor, BeanFactoryAware, BeanClassLoaderAware, SmartLifecycle { private static final Log logger = LogFactory.getLog(IntegrationMBeanExporter.class); public static final String DEFAULT_DOMAIN = "org.springframework.integration"; + private final AnnotationJmxAttributeSource attributeSource = new AnnotationJmxAttributeSource(); private ListableBeanFactory beanFactory; @@ -123,6 +125,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP private final Map objectNameStaticProperties = new HashMap(); + public IntegrationMBeanExporter() { super(); // Shouldn't be necessary, but to be on the safe side... @@ -131,6 +134,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP setAssembler(new MetadataMBeanInfoAssembler(attributeSource)); } + @Override public void setBeanClassLoader(ClassLoader classLoader) { this.beanClassLoader = classLoader; @@ -163,7 +167,6 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { - if (bean instanceof Advised) { for (Advisor advisor : ((Advised) bean).getAdvisors()) { Advice advice = advisor.getAdvice(); @@ -194,19 +197,19 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP Object target = extractTarget(bean); if (target instanceof QueueChannel) { monitor = new QueueChannelMetrics((QueueChannel) target, beanName); - } else { + } + else { monitor = new PollableChannelMetrics(beanName); } - } else { + } + else { monitor = new DirectChannelMetrics(beanName); } Object advised = applyChannelInterceptor(bean, monitor, beanClassLoader); channels.add(monitor); return advised; } - return bean; - } public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { @@ -215,7 +218,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP @Override protected void registerBeans() { - // Completely disable sup class registration to avoid duplicates + // Completely disable super class registration to avoid duplicates } public final boolean isAutoStartup() { @@ -230,7 +233,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP this.lifecycleLock.lock(); try { return this.running; - } finally { + } + finally { this.lifecycleLock.unlock(); } } @@ -245,7 +249,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP logger.info("started " + this); } } - } finally { + } + finally { this.lifecycleLock.unlock(); } } @@ -260,7 +265,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP logger.info("stopped " + this); } } - } finally { + } + finally { this.lifecycleLock.unlock(); } } @@ -270,7 +276,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP try { this.stop(); callback.run(); - } finally { + } + finally { this.lifecycleLock.unlock(); } } @@ -450,7 +457,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } try { return extractTarget(advised.getTargetSource().getTarget()); - } catch (Exception e) { + } + catch (Exception e) { logger.error("Could not extract target", e); return null; } @@ -462,7 +470,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP if (bean instanceof Advised) { ((Advised) bean).addAdvisor(advisor); return bean; - } else { + } + else { ProxyFactory proxyFactory = new ProxyFactory(bean); proxyFactory.addAdvisor(advisor); return proxyFactory.getProxy(beanClassLoader); @@ -522,7 +531,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP Object field = null; try { field = extractTarget(getField(endpoint, "handler")); - } catch (Exception e) { + } + catch (Exception e) { logger.trace("Could not get handler from bean = " + beanName); } if (field == monitor.getMessageHandler()) { @@ -541,7 +551,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP if (targetSource != null) { try { target = targetSource.getTarget(); - } catch (Exception e) { + } + catch (Exception e) { logger.debug("Could not get handler from bean = " + name); } } @@ -659,7 +670,6 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP monitor.setName(name); return result; - } private static Object getField(Object target, String name) { diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMetrics.java index c50c6ba78c..cc0e8123d2 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMetrics.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMetrics.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.monitor; import org.springframework.context.Lifecycle; @@ -25,9 +26,7 @@ import org.springframework.jmx.export.annotation.ManagedResource; * 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 { @@ -36,68 +35,70 @@ public class LifecycleMessageHandlerMetrics implements MessageHandlerMetrics, Li private final MessageHandlerMetrics delegate; + public LifecycleMessageHandlerMetrics(Lifecycle lifecycle, MessageHandlerMetrics delegate) { this.lifecycle = lifecycle; this.delegate = delegate; } + @ManagedAttribute public boolean isRunning() { - return lifecycle.isRunning(); + return this.lifecycle.isRunning(); } @ManagedOperation public void start() { - lifecycle.start(); + this.lifecycle.start(); } @ManagedOperation public void stop() { - lifecycle.stop(); + this.lifecycle.stop(); } public void reset() { - delegate.reset(); + this.delegate.reset(); } public int getErrorCount() { - return delegate.getErrorCount(); + return this.delegate.getErrorCount(); } public int getHandleCount() { - return delegate.getHandleCount(); + return this.delegate.getHandleCount(); } public double getMaxDuration() { - return delegate.getMaxDuration(); + return this.delegate.getMaxDuration(); } public double getMeanDuration() { - return delegate.getMeanDuration(); + return this.delegate.getMeanDuration(); } public double getMinDuration() { - return delegate.getMinDuration(); + return this.delegate.getMinDuration(); } public double getStandardDeviationDuration() { - return delegate.getStandardDeviationDuration(); + return this.delegate.getStandardDeviationDuration(); } public Statistics getDuration() { - return delegate.getDuration(); + return this.delegate.getDuration(); } public String getName() { - return delegate.getName(); + return this.delegate.getName(); } public String getSource() { - return delegate.getSource(); + return this.delegate.getSource(); } public int getActiveCount() { - return delegate.getActiveCount(); + return this.delegate.getActiveCount(); } } diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageSourceMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageSourceMetrics.java index 4762d187fd..fe51f498b8 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageSourceMetrics.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageSourceMetrics.java @@ -13,6 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.monitor; import org.springframework.context.Lifecycle; @@ -22,12 +23,10 @@ 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. + * be used to start and stop polling endpoints, for instance, in a live system. * * @author Dave Syer - * * @since 2.0 - * */ @ManagedResource public class LifecycleMessageSourceMetrics implements MessageSourceMetrics, Lifecycle { @@ -36,37 +35,39 @@ public class LifecycleMessageSourceMetrics implements MessageSourceMetrics, Life private final MessageSourceMetrics delegate; + public LifecycleMessageSourceMetrics(Lifecycle lifecycle, MessageSourceMetrics delegate) { this.lifecycle = lifecycle; this.delegate = delegate; } + @ManagedOperation public void reset() { - delegate.reset(); + this.delegate.reset(); } @ManagedAttribute public boolean isRunning() { - return lifecycle.isRunning(); + return this.lifecycle.isRunning(); } @ManagedOperation public void start() { - lifecycle.start(); + this.lifecycle.start(); } @ManagedOperation public void stop() { - lifecycle.stop(); + this.lifecycle.stop(); } public String getName() { - return delegate.getName(); + return this.delegate.getName(); } public String getSource() { - return delegate.getSource(); + return this.delegate.getSource(); } /** @@ -74,7 +75,7 @@ public class LifecycleMessageSourceMetrics implements MessageSourceMetrics, Life * @see org.springframework.integration.monitor.MessageSourceMetrics#getMessageCount() */ public int getMessageCount() { - return delegate.getMessageCount(); + return this.delegate.getMessageCount(); } } diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMetrics.java index 703893f09c..92c5058997 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMetrics.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMetrics.java @@ -13,6 +13,7 @@ * 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; @@ -24,9 +25,7 @@ import org.springframework.jmx.support.MetricType; * channel types. * * @author Dave Syer - * * @since 2.0 - * */ public interface MessageChannelMetrics { @@ -108,4 +107,4 @@ public interface MessageChannelMetrics { */ Statistics getErrorRate(); -} \ No newline at end of file +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMetrics.java index b62bc95232..903f200dfe 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMetrics.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMetrics.java @@ -13,6 +13,7 @@ * 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; @@ -21,7 +22,6 @@ import org.springframework.jmx.support.MetricType; /** * @author Dave Syer - * * @since 2.0 */ public interface MessageHandlerMetrics { @@ -42,7 +42,7 @@ public interface MessageHandlerMetrics { int getErrorCount(); /** - * @return the maximum handler duration (milliseconds) + * @return the mean handler duration (milliseconds) */ @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Mean Duration in Milliseconds") double getMeanDuration(); @@ -54,11 +54,14 @@ public interface MessageHandlerMetrics { double getMinDuration(); /** - * @return the standard deviation handler duration (milliseconds) + * @return the maximum handler duration (milliseconds) */ @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Max Duration in Milliseconds") double getMaxDuration(); + /** + * @return the standard deviation handler duration (milliseconds) + */ @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Standard Deviation Duration in Milliseconds") double getStandardDeviationDuration(); @@ -74,4 +77,4 @@ public interface MessageHandlerMetrics { String getSource(); -} \ No newline at end of file +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageSourceMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageSourceMetrics.java index b1984b37e4..7fda97d47f 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageSourceMetrics.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageSourceMetrics.java @@ -19,7 +19,6 @@ import org.springframework.jmx.support.MetricType; /** * @author Dave Syer - * * @since 2.0 */ public interface MessageSourceMetrics { diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMetrics.java index 63fc62df91..2386945c52 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMetrics.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMetrics.java @@ -13,11 +13,13 @@ * 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.export.annotation.ManagedOperation; @@ -25,9 +27,7 @@ import org.springframework.jmx.support.MetricType; /** * @author Dave Syer - * * @since 2.0 - * */ public class PollableChannelMetrics extends DirectChannelMetrics { @@ -35,6 +35,7 @@ public class PollableChannelMetrics extends DirectChannelMetrics { private final AtomicInteger receiveErrorCount = new AtomicInteger(); + /** * @param name */ @@ -42,6 +43,7 @@ public class PollableChannelMetrics extends DirectChannelMetrics { super(name); } + @Override protected Object doInvoke(MethodInvocation invocation, String method, MessageChannel channel) throws Throwable { if ("receive".equals(method)) { @@ -56,14 +58,13 @@ public class PollableChannelMetrics extends DirectChannelMetrics { } try { Object object = invocation.proceed(); - if (object!=null) { - receiveCount.incrementAndGet(); + if (object != null) { + this.receiveCount.incrementAndGet(); } return object; - } catch (Throwable e) { - receiveErrorCount.incrementAndGet(); + this.receiveErrorCount.incrementAndGet(); throw e; } } @@ -71,24 +72,24 @@ public class PollableChannelMetrics extends DirectChannelMetrics { @ManagedOperation public synchronized void reset() { super.reset(); - receiveErrorCount.set(0); - receiveCount.set(0); + this.receiveErrorCount.set(0); + this.receiveCount.set(0); } @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Count") public int getReceiveCount() { - return receiveCount.get(); + return this.receiveCount.get(); } @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Error Count") public int getReceiveErrorCount() { - return receiveErrorCount.get(); + return this.receiveErrorCount.get(); } @Override public String toString() { - return String.format("MessageChannelMonitor: [name=%s, sends=%d, receives=%d]", getName(), getSendCount(), - receiveCount.get()); + return String.format("MessageChannelMonitor: [name=%s, sends=%d, receives=%d]", + getName(), getSendCount(), this.receiveCount.get()); } } diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/QueueChannelMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/QueueChannelMetrics.java index d3764b7f9d..5c7e844a41 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/QueueChannelMetrics.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/QueueChannelMetrics.java @@ -13,6 +13,7 @@ * 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; @@ -21,15 +22,15 @@ import org.springframework.jmx.support.MetricType; /** * @author Dave Syer - * * @since 2.0 - * */ public class QueueChannelMetrics extends PollableChannelMetrics { private final QueueChannel channel; + /** + * @param channel * @param name */ public QueueChannelMetrics(QueueChannel channel, String name) { @@ -37,6 +38,7 @@ public class QueueChannelMetrics extends PollableChannelMetrics { this.channel = channel; } + @ManagedMetric(metricType = MetricType.GAUGE, displayName = "QueueChannel Queue Size") public int getQueueSize() { return channel.getQueueSize(); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMetrics.java index 9dc052f9db..89e0918d1b 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMetrics.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMetrics.java @@ -13,6 +13,7 @@ * 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; @@ -21,19 +22,16 @@ 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.MessagingException; import org.springframework.integration.core.MessageHandler; import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.util.StopWatch; /** * @author Dave Syer - * * @since 2.0 - * */ @ManagedResource public class SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHandlerMetrics { @@ -42,6 +40,7 @@ public class SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHa private static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10; + private final MessageHandler handler; private final AtomicInteger activeCount = new AtomicInteger(); @@ -50,23 +49,24 @@ public class SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHa private final AtomicInteger errorCount = new AtomicInteger(); - private final ExponentialMovingAverage duration = new ExponentialMovingAverage( - DEFAULT_MOVING_AVERAGE_WINDOW); + private final ExponentialMovingAverage duration = new ExponentialMovingAverage(DEFAULT_MOVING_AVERAGE_WINDOW); - private String name; + private volatile String name; + + private volatile String source; - private String source; public SimpleMessageHandlerMetrics(MessageHandler handler) { this.handler = handler; } + public void setName(String name) { this.name = name; } public String getName() { - return name; + return this.name; } public void setSource(String source) { @@ -78,7 +78,7 @@ public class SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHa } public MessageHandler getMessageHandler() { - return handler; + return this.handler; } public Object invoke(MethodInvocation invocation) throws Throwable { @@ -91,77 +91,77 @@ public class SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHa return invocation.proceed(); } - private void handleMessage(Message message) throws MessageRejectedException, MessageHandlingException, - MessageDeliveryException { + private void handleMessage(Message message) throws MessagingException { if (logger.isTraceEnabled()) { - logger.trace("messageHandler(" + handler + ") message(" + message + ") :"); + logger.trace("messageHandler(" + this.handler + ") message(" + message + ") :"); } - String name = this.name; if (name == null) { - name = handler.toString(); + name = this.handler.toString(); } StopWatch timer = new StopWatch(name + ".handle:execution"); - try { timer.start(); - handleCount.incrementAndGet(); - activeCount.incrementAndGet(); + this.handleCount.incrementAndGet(); + this.activeCount.incrementAndGet(); - handler.handleMessage(message); + this.handler.handleMessage(message); timer.stop(); - duration.append(timer.getTotalTimeMillis()); - } catch (RuntimeException e) { - errorCount.incrementAndGet(); + this.duration.append(timer.getTotalTimeMillis()); + } + catch (RuntimeException e) { + this.errorCount.incrementAndGet(); throw e; - } catch (Error e) { - errorCount.incrementAndGet(); + } + catch (Error e) { + this.errorCount.incrementAndGet(); throw e; - } finally { - activeCount.decrementAndGet(); + } + finally { + this.activeCount.decrementAndGet(); } } public synchronized void reset() { - duration.reset(); - errorCount.set(0); - handleCount.set(0); + this.duration.reset(); + this.errorCount.set(0); + this.handleCount.set(0); } public int getHandleCount() { if (logger.isTraceEnabled()) { logger.trace("Getting Handle Count:" + this); } - return handleCount.get(); + return this.handleCount.get(); } public int getErrorCount() { - return errorCount.get(); + return this.errorCount.get(); } public double getMeanDuration() { - return duration.getMean(); + return this.duration.getMean(); } public double getMinDuration() { - return duration.getMin(); + return this.duration.getMin(); } public double getMaxDuration() { - return duration.getMax(); + return this.duration.getMax(); } public double getStandardDeviationDuration() { - return duration.getStandardDeviation(); + return this.duration.getStandardDeviation(); } - + public int getActiveCount() { - return activeCount.get(); + return this.activeCount.get(); } - + public Statistics getDuration() { - return duration.getStatistics(); + return this.duration.getStatistics(); } @Override @@ -169,4 +169,4 @@ public class SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHa return String.format("MessageHandlerMonitor: [name=%s, source=%s, duration=%s]", name, source, duration); } -} \ No newline at end of file +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageSourceMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageSourceMetrics.java index 0edda2aa6e..3241f5fbd3 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageSourceMetrics.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageSourceMetrics.java @@ -21,7 +21,6 @@ import org.springframework.integration.core.MessageSource; /** * @author Dave Syer - * * @since 2.0 */ public class SimpleMessageSourceMetrics implements MethodInterceptor, MessageSourceMetrics { @@ -30,21 +29,22 @@ public class SimpleMessageSourceMetrics implements MethodInterceptor, MessageSou private final MessageSource messageSource; - private String source; + private volatile String source; + + private volatile String name; - private String name; public SimpleMessageSourceMetrics(MessageSource messageSource) { this.messageSource = messageSource; } - + public void setName(String name) { this.name = name; } public String getName() { - return name; + return this.name; } public void setSource(String source) { @@ -56,22 +56,22 @@ public class SimpleMessageSourceMetrics implements MethodInterceptor, MessageSou } public MessageSource getMessageSource() { - return messageSource; + return this.messageSource; } public void reset() { - messageCount.set(0); + this.messageCount.set(0); } public int getMessageCount() { - return messageCount.get(); + return this.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(); + this.messageCount.incrementAndGet(); } return result; } diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/Statistics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/Statistics.java index 19dcd86003..3bd15bf867 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/Statistics.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/Statistics.java @@ -13,25 +13,26 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration.monitor; /** * @author Dave Syer - * * @since 2.0 - * */ public class Statistics { - + private final int count; + private final double min; + private final double max; + private final double mean; + private final double standardDeviation; - /** - * - */ + public Statistics(int count, double min, double max, double mean, double standardDeviation) { this.count = count; this.min = min; @@ -40,6 +41,7 @@ public class Statistics { this.standardDeviation = standardDeviation; } + public int getCount() { return count; } @@ -62,8 +64,8 @@ public class Statistics { @Override public String toString() { - return String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f]", count, min, max, getMean(), - getStandardDeviation()); + return String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f]", + count, min, max, getMean(), getStandardDeviation()); } }