diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index eacbe63662..6f25a34de6 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -19,11 +19,9 @@ package org.springframework.integration.channel; import java.util.Collections; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.core.OrderComparator; import org.springframework.core.convert.ConversionService; import org.springframework.integration.Message; @@ -52,10 +50,6 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport im private volatile boolean shouldTrack = false; - private final AtomicLong sendSuccessCount = new AtomicLong(); - - private final AtomicLong sendErrorCount = new AtomicLong(); - private volatile Class[] datatypes = new Class[] { Object.class }; private final ChannelInterceptorList interceptors = new ChannelInterceptorList(); @@ -69,24 +63,6 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport im this.shouldTrack = shouldTrack; } - /** - * Return the current count of Messages that have been sent - * to this channel successfully. - */ - public long getSendSuccessCount() { - return this.sendSuccessCount.get(); - } - - /** - * Return the current count of errors that have occurred while - * attempting to send a Message to this channel. This value is - * incremented whenever an Exception is thrown from one of the - * send() methods. - */ - public long getSendErrorCount() { - return this.sendErrorCount.get(); - } - /** * Specify the Message payload datatype(s) supported by this channel. If a * payload type does not match directly, but the 'conversionService' is @@ -179,14 +155,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport im } try { boolean sent = this.doSend(message, timeout); - if (sent) { - this.sendSuccessCount.incrementAndGet(); - } this.interceptors.postSend(message, this, sent); return sent; } catch (Exception e) { - this.sendErrorCount.incrementAndGet(); if (e instanceof MessagingException) { throw (MessagingException) e; } diff --git a/spring-integration-jmx/pom.xml b/spring-integration-jmx/pom.xml index 20b3d6cecb..7f760d1d3c 100644 --- a/spring-integration-jmx/pom.xml +++ b/spring-integration-jmx/pom.xml @@ -29,5 +29,13 @@ org.springframework spring-test + + org.aspectj + aspectjrt + + + org.aspectj + aspectjweaver + diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/control/ControlBus.java b/spring-integration-jmx/src/main/java/org/springframework/integration/control/ControlBus.java index d84d35c5a2..a1b4d80c23 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/control/ControlBus.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/control/ControlBus.java @@ -16,43 +16,22 @@ package org.springframework.integration.control; -import java.lang.reflect.Method; -import java.util.Arrays; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; - import javax.management.MBeanServer; -import javax.management.MalformedObjectNameException; -import javax.management.ObjectName; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; -import org.springframework.beans.factory.BeanFactoryUtils; -import org.springframework.beans.factory.BeanInitializationException; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.ListableBeanFactory; -import org.springframework.context.Lifecycle; import org.springframework.integration.Message; -import org.springframework.integration.MessageChannel; import org.springframework.integration.MessageHeaders; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.core.PollableChannel; import org.springframework.integration.core.SubscribableChannel; -import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.jmx.JmxHeaders; import org.springframework.integration.jmx.OperationInvokingMessageHandler; +import org.springframework.integration.monitor.ObjectNameLocator; import org.springframework.integration.support.MessageBuilder; -import org.springframework.jmx.export.MBeanExporter; -import org.springframework.jmx.export.assembler.AbstractConfigurableMBeanInfoAssembler; -import org.springframework.jmx.export.assembler.MBeanInfoAssembler; -import org.springframework.jmx.support.ObjectNameManager; import org.springframework.util.Assert; -import org.springframework.util.ClassUtils; -import org.springframework.util.ObjectUtils; /** * JMX-based Control Bus implementation. Exports all channel and endpoint beans from a given BeanFactory as MBeans. @@ -62,57 +41,23 @@ import org.springframework.util.ObjectUtils; */ public class ControlBus implements BeanFactoryAware, InitializingBean { - public static final String DEFAULT_DOMAIN = "org.springframework.integration"; - public static final String TARGET_BEAN_NAME = JmxHeaders.PREFIX + "_controlBus_targetBeanName"; private volatile SubscribableChannel operationChannel; - private final MBeanExporter exporter; - - private final String domain; - - private final Map exportedBeanObjectNameMap = new HashMap(); - - private final Map objectNameStaticProperties = new HashMap(); - private volatile ListableBeanFactory beanFactory; - private final Set> managedTypes = new HashSet>(Arrays.asList(new Class[] { - MessageChannel.class, AbstractEndpoint.class })); + private final ObjectNameLocator exporter; + + private final MBeanServer server; + /** - * Static properties that will be added to all object names. - * - * @param objectNameStaticProperties the objectNameStaticProperties to set + * Create a {@link ControlBus}. */ - public void setObjectNameStaticProperties(Map objectNameStaticProperties) { - this.objectNameStaticProperties.putAll(objectNameStaticProperties); - } - - /** - * Create a {@link ControlBus} that will register channels and endpoints as MBeans with the given MBeanServer using - * the default domain name. - * @see #DEFAULT_DOMAIN - */ - public ControlBus(MBeanServer server) { - this(server, null); - } - - /** - * Create a {@link ControlBus} that will register channels and endpoints as MBeans with the given MBeanServer using - * the specified domain name. - */ - public ControlBus(MBeanServer server, String domain) { - Assert.notNull(server, "MBeanServer must not be null."); - this.domain = (domain != null) ? domain : DEFAULT_DOMAIN; - Assert.isTrue(!ObjectUtils.containsElement(server.getDomains(), this.domain), "Domain [" + this.domain - + "] is already in use within this MBeanServer."); - MBeanExporter exporter = new MBeanExporter(); - exporter.setServer(server); - exporter.setAutodetect(false); - exporter.setAssembler(new ControlBusMBeanInfoAssembler()); - this.exporter = exporter; + public ControlBus(ObjectNameLocator locator, MBeanServer server) { + this.exporter = locator; + this.server = server; } public void setOperationChannel(SubscribableChannel operationChannel) { @@ -133,28 +78,12 @@ public class ControlBus implements BeanFactoryAware, InitializingBean { Assert.isTrue(beanFactory instanceof ListableBeanFactory, "A ListableBeanFactory is required."); this.beanFactory = (ListableBeanFactory) beanFactory; } + public void afterPropertiesSet() throws Exception { - this.exporter.afterPropertiesSet(); - for (Class type : this.managedTypes) { - Map beans = BeanFactoryUtils.beansOfTypeIncludingAncestors(this.beanFactory, type); - for (Map.Entry entry : beans.entrySet()) { - Object bean = entry.getValue(); - String beanName = entry.getKey(); - Class beanType = bean.getClass(); - try { - ObjectName objectName = this.generateObjectName(beanName, beanType); - this.exporter.registerManagedResource(bean, objectName); - this.exportedBeanObjectNameMap.put(beanName, objectName); - } - catch (MalformedObjectNameException e) { - throw new BeanInitializationException("Failed to generate JMX ObjectName.", e); - } - } - } OperationInvokingMessageHandler handler = new ControlBusOperationInvokingMessageHandler(); handler.setBeanFactory(this.beanFactory); - handler.setServer(this.exporter.getServer()); + handler.setServer(this.server); handler.afterPropertiesSet(); if (this.operationChannel == null) { this.operationChannel = new DirectChannel(); @@ -162,89 +91,13 @@ public class ControlBus implements BeanFactoryAware, InitializingBean { this.operationChannel.subscribe(handler); } - protected ObjectName generateObjectName(String beanName, Class beanType) throws MalformedObjectNameException { - - String name = beanName.startsWith("org.springframework.integration") ? "anonymous,generated="+beanName : beanName; - StringBuilder sb = new StringBuilder(this.domain + ":name=" + name + ","); - if (MessageChannel.class.isAssignableFrom(beanType)) { - sb.append("type=channel"); - } - else if (AbstractEndpoint.class.isAssignableFrom(beanType)) { - sb.append("type=endpoint"); - } - else { - sb.append("type=" + ClassUtils.getShortNameAsProperty(beanType)); - } - for (String key : objectNameStaticProperties.keySet()) { - sb.append("," + key + "=" + objectNameStaticProperties.get(key)); - } - return ObjectNameManager.getInstance(sb.toString()); - - } - - /** - * An {@link MBeanInfoAssembler} implementation for channels and endpoints. - */ - private static class ControlBusMBeanInfoAssembler extends AbstractConfigurableMBeanInfoAssembler { - - @Override - protected boolean includeOperation(Method method, String beanKey) { - Class declaringClass = method.getDeclaringClass(); - return this.shouldInclude(method, declaringClass); - } - - @Override - protected boolean includeReadAttribute(Method method, String beanKey) { - Class declaringClass = method.getDeclaringClass(); - return this.shouldInclude(method, declaringClass); - } - - @Override - protected boolean includeWriteAttribute(Method method, String beanKey) { - Class declaringClass = method.getDeclaringClass(); - return this.shouldInclude(method, declaringClass); - } - - private boolean shouldInclude(Method method, Class declaringClass) { - if (MessageChannel.class.isAssignableFrom(declaringClass) - || AbstractEndpoint.class.isAssignableFrom(declaringClass)) { - Class managementInterface = this.getManagementInterface(declaringClass); - if (managementInterface != null) { - for (Method interfaceMethod : managementInterface.getMethods()) { - if (interfaceMethod.getName().equals(method.getName()) - && Arrays.equals(interfaceMethod.getParameterTypes(), method.getParameterTypes())) { - return true; - } - } - } - } - return false; - } - - private Class getManagementInterface(Class type) { - if (AbstractEndpoint.class.isAssignableFrom(type)) { - return Lifecycle.class; - } - if (QueueChannel.class.isAssignableFrom(type)) { - return QueueChannelInfo.class; - } - if (PollableChannel.class.isAssignableFrom(type)) { - return PollableChannelInfo.class; - } - if (MessageChannel.class.isAssignableFrom(type)) { - return MessageChannelInfo.class; - } - return null; - } - } - private class ControlBusOperationInvokingMessageHandler extends OperationInvokingMessageHandler { @Override protected Object handleRequestMessage(Message requestMessage) { String beanName = requestMessage.getHeaders().get(TARGET_BEAN_NAME, String.class); Assert.notNull(beanName, "The ControlBus.TARGET_BEAN_NAME is required."); - ObjectName objectName = exportedBeanObjectNameMap.get(beanName); + String objectName = exporter.getObjectName(beanName); Assert.notNull(objectName, "ControlBus has not exported an MBean for '" + beanName + "'"); requestMessage = MessageBuilder.fromMessage(requestMessage).setHeader(JmxHeaders.OBJECT_NAME, objectName) .setHeader(TARGET_BEAN_NAME, null).build(); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/ControlBusParser.java b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/ControlBusParser.java index c658ee6172..9da9311ab9 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/ControlBusParser.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/ControlBusParser.java @@ -18,13 +18,14 @@ package org.springframework.integration.jmx.config; import javax.management.MBeanServerFactory; -import org.w3c.dom.Element; - +import org.springframework.beans.BeanMetadataElement; +import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; import org.springframework.util.StringUtils; +import org.w3c.dom.Element; /** * @author Mark Fisher @@ -39,20 +40,30 @@ public class ControlBusParser extends AbstractSimpleBeanDefinitionParser { @Override protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { - Object source = parserContext.extractSource(element); - builder.getRawBeanDefinition().setSource(source); - String mbeanServer = element.getAttribute("mbean-server"); - if (StringUtils.hasText(mbeanServer)) { - builder.addConstructorArgReference(mbeanServer); - } - else { - builder.addConstructorArgValue(MBeanServerFactory.createMBeanServer()); - } - String domain = element.getAttribute("domain"); - if (StringUtils.hasText(domain)) { - builder.addConstructorArgValue(domain); - } + builder.getRawBeanDefinition().setSource(parserContext.extractSource(element)); + Object mbeanServer = getMBeanServer(element, parserContext); + builder.addConstructorArgValue(getMBeanExporter(element, parserContext, mbeanServer)); + builder.addConstructorArgValue(mbeanServer); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "operation-channel"); } + private Object getMBeanServer(Element element, ParserContext parserContext) { + String mbeanServer = element.getAttribute("mbean-server"); + if (StringUtils.hasText(mbeanServer)) { + return new RuntimeBeanReference(mbeanServer); + } + else { + return MBeanServerFactory.createMBeanServer(); + } + } + + private BeanMetadataElement getMBeanExporter(Element element, ParserContext parserContext, Object mbeanServer) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder + .genericBeanDefinition("org.springframework.integration.monitor.IntegrationMBeanExporter"); + builder.getRawBeanDefinition().setSource(parserContext.extractSource(element)); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "domain"); + builder.addPropertyValue("server", mbeanServer); + return builder.getBeanDefinition(); + } + } diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageCumulativeHistory.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageCumulativeHistory.java new file mode 100644 index 0000000000..a6e117ec9b --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageCumulativeHistory.java @@ -0,0 +1,85 @@ +/* + * 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; + +/** + * Cumulative statistics for a series of real numbers with higher weight given to recent data but without storing any + * history. Older values are given exponentially smaller weight, with a decay factor determined by a "window" size + * chosen by the client. + * + * @author Dave Syer + * + */ +public class ExponentialMovingAverageCumulativeHistory { + + private int count; + + private double weight; + + private double sum; + + private double sumSquares; + + private double min; + + private double max; + + private final double decay; + + /** + * @param window the exponential lapse window (number of measurements) + */ + public ExponentialMovingAverageCumulativeHistory(int window) { + this.decay = 1 - 1. / window; + } + + public void append(double value) { + if (value > max || count == 0) + max = value; + if (value < min || count == 0) + min = value; + sum = decay * sum + value; + sumSquares = decay * sumSquares + value * value; + weight = decay * weight + 1; + count++; + } + + public int getCount() { + return count; + } + + public double getMean() { + return weight > 0 ? sum / weight : 0.; + } + + public double getStandardDeviation() { + double mean = getMean(); + double var = weight > 0 ? sumSquares / weight - mean * mean : 0.; + return var > 0 ? Math.sqrt(var) : 0; + } + + public double getMax() { + return max; + } + + public double getMin() { + return min; + } + + @Override + public String toString() { + return String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f]", count, min, max, getMean(), + getStandardDeviation()); + } + +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRateCumulativeHistory.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRateCumulativeHistory.java new file mode 100644 index 0000000000..99bf8b41e5 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRateCumulativeHistory.java @@ -0,0 +1,108 @@ +/* + * 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; + +/** + * Cumulative statistics for rate with higher weight given to recent data but without storing any history. Older values + * are given exponentially smaller weight, with a decay factor determined by a duration chosen by the client. + * + * @author Dave Syer + * + */ +public class ExponentialMovingAverageRateCumulativeHistory { + + private final ExponentialMovingAverageCumulativeHistory rates; + + private double weight; + + private double sum; + + private double min; + + private double max; + + private volatile long t0 = System.currentTimeMillis(); + + private final double lapse; + + private final double period; + + /** + * @param period the period to base the rate measurement (in seconds) + * @param lapsePeriod the exponential lapse rate for the rate average (in seconds) + * @param window the exponential lapse window (number of measurements) + */ + public ExponentialMovingAverageRateCumulativeHistory(double period, double lapsePeriod, int window) { + rates = new ExponentialMovingAverageCumulativeHistory(10); + this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to millisecs + this.period = period * 1000; // convert to millisecs + } + + public void increment() { + + long t = System.currentTimeMillis(); + double value = t > t0 ? (t - t0) / period : 0; + if (value > max || getCount() == 0) { + max = value; + } + if (value < min || getCount() == 0) { + min = value; + } + double alpha = Math.exp((t0 - t) * lapse); + t0 = t; + sum = alpha * sum + value; + weight = alpha * weight + 1; + rates.append(sum > 0 ? weight / sum : 0); + + } + + public int getCount() { + return rates.getCount(); + } + + /** + * @return the time in seconds since the last measurement + */ + public double getTimeSinceLastMeasurement() { + return (System.currentTimeMillis() - t0) / 1000.; + } + + public double getMean() { + int count = rates.getCount(); + if (count==0) { + return 0; + } + long t = System.currentTimeMillis(); + double value = t > t0 ? (t - t0) / period : 0; + return count / (count / rates.getMean() + value); + } + + public double getStandardDeviation() { + return rates.getStandardDeviation(); + } + + public double getMax() { + return min > 0 ? 1 / min : 0; + } + + public double getMin() { + return max > 0 ? 1 / max : 0; + } + + @Override + public String toString() { + return String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f, timeSinceLast=%f]", getCount(), getMin(), + getMax(), getMean(), getStandardDeviation(), getTimeSinceLastMeasurement()); + } + +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRatioCumulativeHistory.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRatioCumulativeHistory.java new file mode 100644 index 0000000000..20b3db4135 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRatioCumulativeHistory.java @@ -0,0 +1,102 @@ +/* + * 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; + +/** + * Cumulative statistics for success rate (ratio) with higher weight given to recent data but without storing any + * history. Older values are given exponentially smaller weight, with a decay factor determined by a duration chosen by + * the client. + * + * @author Dave Syer + * + */ +public class ExponentialMovingAverageRatioCumulativeHistory { + + private double weight; + + private double sum; + + private volatile long t0 = System.currentTimeMillis(); + + private final double lapse; + + private final ExponentialMovingAverageCumulativeHistory cumulative; + + /** + * @param lapsePeriod the exponential lapse rate for the rate average (in seconds) + * @param window the exponential lapse window (number of measurements) + */ + public ExponentialMovingAverageRatioCumulativeHistory(double lapsePeriod, int window) { + this.cumulative = new ExponentialMovingAverageCumulativeHistory(window); + this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to millisecs + } + + public void success() { + append(1); + } + + public void failure() { + append(0); + } + + private 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); + + } + + public int getCount() { + return cumulative.getCount(); + } + + /** + * @return the time in seconds since the last measurement + */ + public double getTimeSinceLastMeasurement() { + return (System.currentTimeMillis() - t0) / 1000.; + } + + public double getMean() { + int count = cumulative.getCount(); + if (count == 0) { + return 0; + } + long t = System.currentTimeMillis(); + double alpha = Math.exp((t0 - t) * lapse); + return alpha * cumulative.getMean() + 1 - alpha; + } + + public double getStandardDeviation() { + return cumulative.getStandardDeviation(); + } + + public double getMax() { + return cumulative.getMax(); + } + + public double getMin() { + return cumulative.getMin(); + } + + @Override + public String toString() { + return String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f, timeSinceLast=%f]", getCount(), getMin(), + getMax(), getMean(), getStandardDeviation(), getTimeSinceLastMeasurement()); + } + +} 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 new file mode 100644 index 0000000000..0e527a37f2 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java @@ -0,0 +1,544 @@ +/* + * 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.lang.reflect.Field; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.ReentrantLock; + +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.aop.support.NameMatchMethodPointcutAdvisor; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanClassLoaderAware; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.context.Lifecycle; +import org.springframework.context.SmartLifecycle; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.MessageHandler; +import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.jmx.export.MBeanExporter; +import org.springframework.jmx.export.annotation.AnnotationJmxAttributeSource; +import org.springframework.jmx.export.annotation.ManagedAttribute; +import org.springframework.jmx.export.annotation.ManagedMetric; +import org.springframework.jmx.export.annotation.ManagedOperation; +import org.springframework.jmx.export.annotation.ManagedResource; +import org.springframework.jmx.export.assembler.MetadataMBeanInfoAssembler; +import org.springframework.jmx.export.naming.MetadataNamingStrategy; +import org.springframework.jmx.support.MetricType; +import org.springframework.util.Assert; +import org.springframework.util.ReflectionUtils; + +/** + * MBean exporter for Spring Integration components in an existing application. + * + * @author Dave Syer + * @author Helena Edelson + */ +@ManagedResource +public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostProcessor, BeanFactoryAware, + BeanClassLoaderAware, SmartLifecycle, ObjectNameLocator { + + private static final Log logger = LogFactory.getLog(IntegrationMBeanExporter.class); + + public static final String DEFAULT_DOMAIN = "spring.application"; + + private Set channelKeys = new HashSet(); + + private Set handlerKeys = new HashSet(); + + private final AnnotationJmxAttributeSource attributeSource = new AnnotationJmxAttributeSource(); + + private ListableBeanFactory beanFactory; + + private Map anonymousCounters = new HashMap(); + + private Set handlers = new HashSet(); + + private Set channels = new HashSet(); + + private Map channelsByName = new HashMap(); + + private Map handlersByName = new HashMap(); + + private Map objectNamesByName = new HashMap(); + + private ClassLoader beanClassLoader; + + private volatile boolean autoStartup = true; + + private volatile int phase = 0; + + private volatile boolean running; + + private final ReentrantLock lifecycleLock = new ReentrantLock(); + + private String domain = DEFAULT_DOMAIN; + + private final Map objectNameStaticProperties = new HashMap(); + + public IntegrationMBeanExporter() { + super(); + // Shouldn't be necessary, but to be on the safe side... + setAutodetect(false); + setNamingStrategy(new MetadataNamingStrategy(attributeSource)); + setAssembler(new MetadataMBeanInfoAssembler(attributeSource)); + } + + @Override + public void setBeanClassLoader(ClassLoader classLoader) { + this.beanClassLoader = classLoader; + super.setBeanClassLoader(classLoader); + } + + /** + * Static properties that will be added to all object names. + * + * @param objectNameStaticProperties the objectNameStaticProperties to set + */ + public void setObjectNameStaticProperties(Map objectNameStaticProperties) { + this.objectNameStaticProperties.putAll(objectNameStaticProperties); + } + + /** + * The JMX domain to use for MBeans registered. Defaults to spring.application (which is useful in + * SpringSource HQ). + * + * @param domain the domain name to set + */ + public void setDomain(String domain) { + this.domain = domain; + } + + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + super.setBeanFactory(beanFactory); + Assert.isTrue(beanFactory instanceof ListableBeanFactory, "A ListableBeanFactory is required."); + this.beanFactory = (ListableBeanFactory) beanFactory; + } + + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + if (bean instanceof MessageHandler) { + SimpleMessageHandlerMonitor monitor = new SimpleMessageHandlerMonitor((MessageHandler) bean); + handlers.add(monitor); + return monitor; + } + if (bean instanceof MessageChannel) { + SimpleMessageChannelMonitor monitor; + if (bean instanceof PollableChannel) { + Object target = extractTarget(bean); + if (target instanceof QueueChannel) { + monitor = new QueueChannelMonitor((QueueChannel) target, beanName); + } + else { + monitor = new PollableChannelMonitor(beanName); + } + } + else { + monitor = new SimpleMessageChannelMonitor(beanName); + } + Object advised = applyChannelInterceptor(bean, monitor, beanClassLoader); + channels.add(monitor); + return advised; + } + return bean; + } + + public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { + return bean; + } + + @Override + protected void registerBeans() { + // Completely disable sup class registration to avoid duplicates + } + + public final boolean isAutoStartup() { + return this.autoStartup; + } + + public final int getPhase() { + return this.phase; + } + + public final boolean isRunning() { + this.lifecycleLock.lock(); + try { + return this.running; + } + finally { + this.lifecycleLock.unlock(); + } + } + + public final void start() { + this.lifecycleLock.lock(); + try { + if (!this.running) { + this.doStart(); + this.running = true; + if (logger.isInfoEnabled()) { + logger.info("started " + this); + } + } + } + finally { + this.lifecycleLock.unlock(); + } + } + + public final void stop() { + this.lifecycleLock.lock(); + try { + if (this.running) { + this.doStop(); + this.running = false; + if (logger.isInfoEnabled()) { + logger.info("stopped " + this); + } + } + } + finally { + this.lifecycleLock.unlock(); + } + } + + public final void stop(Runnable callback) { + this.lifecycleLock.lock(); + try { + this.stop(); + callback.run(); + } + finally { + this.lifecycleLock.unlock(); + } + } + + protected void doStop() { + } + + protected void doStart() { + registerChannels(); + registerHandlers(); + logger.info("Summary on start: " + objectNamesByName); + } + + @Override + public void destroy() { + super.destroy(); + for (MessageChannelMonitor monitor : channels) { + logger.info("Summary on shutdown: " + monitor); + } + for (MessageHandlerMonitor monitor : handlers) { + logger.info("Summary on shutdown: " + monitor); + } + } + + @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Channel Count") + public double getChannelCount() { + return channelKeys.size(); + } + + @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageHandler Handler Count") + public double getHandlerCount() { + return handlerKeys.size(); + } + + @ManagedAttribute + public Collection getHandlerNames() { + return handlersByName.keySet(); + } + + @ManagedAttribute + public Collection getChannelNames() { + return channelsByName.keySet(); + } + + @ManagedOperation(description = "Get the JMX object name (as a String) for the specified Spring bean name") + public String getObjectName(String beanName) { + return objectNamesByName.get(beanName); + } + + @ManagedAttribute(description = "Get a map from Spring bean names to JMX object names registered by this exporter") + public Map getObjectNames() { + return Collections.unmodifiableMap(objectNamesByName); + } + + public double getHandlerMeanDuration(String name) { + if (handlersByName.containsKey(name)) { + return handlersByName.get(name).getMeanDuration(); + } + logger.debug("No handler found for (" + name + ")"); + return -1; + } + + public long getChannelSendCount(String name) { + if (channelsByName.containsKey(name)) { + return channelsByName.get(name).getSendCount(); + } + logger.debug("No channel found for (" + name + ")"); + return -1; + } + + public long getChannelSendErrorCount(String name) { + if (channelsByName.containsKey(name)) { + return channelsByName.get(name).getSendErrorCount(); + } + logger.debug("No channel found for (" + name + ")"); + return -1; + } + + public long getChannelReceiveCount(String name) { + if (channelsByName.containsKey(name)) { + if (channelsByName.get(name) instanceof PollableChannelMonitor) { + return ((PollableChannelMonitor) channelsByName.get(name)).getReceiveCount(); + } + } + logger.debug("No channel found for (" + name + ")"); + return -1; + } + + public double getChannelSendRate(String name) { + if (channelsByName.containsKey(name)) { + return channelsByName.get(name).getSendRate(); + } + logger.debug("No channel found for (" + name + ")"); + return -1; + } + + public double getChannelErrorRate(String name) { + if (channelsByName.containsKey(name)) { + return channelsByName.get(name).getErrorRate(); + } + logger.debug("No channel found for (" + name + ")"); + return -1; + } + + public double getChannelMeanSendDuration(String name) { + if (channelsByName.containsKey(name)) { + return channelsByName.get(name).getMeanSendDuration(); + } + logger.debug("No channel found for (" + name + ")"); + return -1; + } + + private void registerChannels() { + for (SimpleMessageChannelMonitor monitor : channels) { + String name = monitor.getName(); + // Only register once... + if (!channelsByName.containsKey(name)) { + String beanKey = getChannelBeanKey(name); + logger.info("Registering MessageChannel " + name); + if (name != null) { + channelsByName.put(name, monitor); + objectNamesByName.put(name, beanKey); + } + registerBeanNameOrInstance(monitor, beanKey); + } + } + } + + private void registerHandlers() { + for (SimpleMessageHandlerMonitor source : handlers) { + MessageHandlerMonitor monitor = enhanceMonitor(source); + String name = monitor.getName(); + // Only register once... + if (!handlersByName.containsKey(name)) { + String beanKey = getHandlerBeanKey(monitor); + if (name != null) { + handlersByName.put(name, monitor); + objectNamesByName.put(name, beanKey); + } + registerBeanNameOrInstance(monitor, beanKey); + } + } + } + + private Object applyChannelInterceptor(Object bean, SimpleMessageChannelMonitor interceptor, + ClassLoader beanClassLoader) { + NameMatchMethodPointcutAdvisor channelsAdvice = new NameMatchMethodPointcutAdvisor(interceptor); + channelsAdvice.addMethodName("send"); + channelsAdvice.addMethodName("receive"); + return applyAdvice(bean, channelsAdvice, beanClassLoader); + } + + private Object extractTarget(Object bean) { + if (!(bean instanceof Advised)) { + return bean; + } + Advised advised = (Advised) bean; + if (advised.getTargetSource() == null) { + return null; + } + try { + return extractTarget(advised.getTargetSource().getTarget()); + } + catch (Exception e) { + logger.error("Could not extract target", e); + return null; + } + } + + 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); + return proxyFactory.getProxy(beanClassLoader); + } + } + return bean; + } + + private String getChannelBeanKey(String channel) { + String name = "" + channel; + if (name.startsWith("org.springframework.integration")) { + name = name + ",source=anonymous"; + } + return String.format(domain + ":type=MessageChannel,name=%s" + getStaticNames(), name); + } + + private String getHandlerBeanKey(MessageHandlerMonitor handler) { + // This ordering of keys seems to work with default settings of JConsole + return String.format(domain + ":type=MessageHandler,name=%s,bean=%s" + getStaticNames(), handler.getName(), + handler.getSource()); + } + + private String getStaticNames() { + if (objectNameStaticProperties.isEmpty()) { + return ""; + } + StringBuilder builder = new StringBuilder(); + for (String key : objectNameStaticProperties.keySet()) { + builder.append("," + key + "=" + objectNameStaticProperties.get(key)); + } + return builder.toString(); + } + + private MessageHandlerMonitor enhanceMonitor(SimpleMessageHandlerMonitor monitor) { + + MessageHandlerMonitor result = monitor; + + if (monitor.getName() != null && monitor.getSource() != null) { + return monitor; + } + + // Assignment algorithm and bean id, with bean id pulled reflectively out of enclosing endpoint if possible + String[] names = beanFactory.getBeanNamesForType(AbstractEndpoint.class); + + String name = null; + String source = "endpoint"; + Object endpoint = null; + + for (String beanName : names) { + endpoint = beanFactory.getBean(beanName); + Object field = null; + try { + field = getField(endpoint, "handler"); + } + catch (Exception e) { + logger.debug("Could not get handler from bean = " + beanName); + } + if (field == monitor) { + name = beanName; + break; + } + } + if (name != null && endpoint != null && name.startsWith("_org.springframework.integration")) { + name = name.substring("_org.springframework.integration".length() + 1); + source = "internal"; + } + if (name != null && endpoint != null && name.startsWith("org.springframework.integration")) { + Object target = endpoint; + if (endpoint instanceof Advised) { + TargetSource targetSource = ((Advised) endpoint).getTargetSource(); + if (targetSource != null) { + try { + target = targetSource.getTarget(); + } + catch (Exception e) { + logger.debug("Could not get handler from bean = " + name); + } + } + } + Object field = getField(target, "inputChannel"); + if (field != null) { + if (!anonymousCounters.containsKey(field)) { + anonymousCounters.put(field, new AtomicLong()); + } + AtomicLong count = anonymousCounters.get(field); + long total = count.incrementAndGet(); + String suffix = ""; + /* + * Short hack to makes sure object names are unique if more than one endpoint has the same input channel + */ + if (total > 1) { + suffix = "#" + total; + } + name = field + suffix; + source = "anonymous"; + } + } + + if (endpoint instanceof Lifecycle) { + // Wrap the monitor in a lifecycle so it exposes the start/stop operations + result = new LifecycleMessageHandlerMonitor((Lifecycle) endpoint, monitor); + } + + if (name == null) { + name = monitor.getMessageHandler().toString(); + source = "handler"; + } + + monitor.setSource(source); + monitor.setName(name); + + return result; + + } + + private static Object getField(Object target, String name) { + Assert.notNull(target, "Target object must not be null"); + Field field = ReflectionUtils.findField(target.getClass(), name); + if (field == null) { + throw new IllegalArgumentException("Could not find field [" + name + "] on target [" + target + "]"); + } + + if (logger.isDebugEnabled()) { + logger.debug("Getting field [" + name + "] from target [" + target + "]"); + } + ReflectionUtils.makeAccessible(field); + return ReflectionUtils.getField(field, target); + } + +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMonitor.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMonitor.java new file mode 100644 index 0000000000..c0e3fba937 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMonitor.java @@ -0,0 +1,87 @@ +/* + * 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; + +/** + * @author Dave Syer + * + * @since 2.0 + * + */ +@ManagedResource +public class LifecycleMessageHandlerMonitor implements MessageHandlerMonitor { + + private final Lifecycle lifecycle; + private final MessageHandlerMonitor delegate; + + public LifecycleMessageHandlerMonitor(Lifecycle lifecycle, MessageHandlerMonitor 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 String getName() { + return delegate.getName(); + } + + public String getSource() { + return delegate.getSource(); + } + +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMonitor.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMonitor.java new file mode 100644 index 0000000000..54866688ec --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMonitor.java @@ -0,0 +1,57 @@ +/* + * 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 dsyer + * + */ +public interface MessageChannelMonitor { + + @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Sends") + int getSendCount(); + + @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Send Errors") + int getSendErrorCount(); + + @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Time Since Last Send in Seconds") + double getTimeSinceLastSend(); + + @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Rate per Second") + double getSendRate(); + + @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Error Rate per Second") + double getErrorRate(); + + @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Mean Channel Error Ratio per Minute") + double getErrorRatio(); + + @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Mean Duration") + double getMeanSendDuration(); + + @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Min Duration") + double getMinSendDuration(); + + @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Max Duration") + double getMaxSendDuration(); + + @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Send Standard Deviation Duration") + double getStandardDeviationSendDuration(); + +} \ No newline at end of file diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMonitor.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMonitor.java new file mode 100644 index 0000000000..afce109a8b --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMonitor.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.jmx.export.annotation.ManagedMetric; +import org.springframework.jmx.support.MetricType; + +/** + * @author Dave Syer + * + * @since 2.0 + */ +public interface MessageHandlerMonitor { + + @ManagedMetric(metricType = MetricType.COUNTER, displayName = "Handler Execution Count", description = "rate=1h") + int getHandleCount(); + + @ManagedMetric(metricType = MetricType.COUNTER, displayName = "Handler Error Count", description = "rate=1h") + int getErrorCount(); + + @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Mean Duration") + double getMeanDuration(); + + @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Min Duration") + double getMinDuration(); + + @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Max Duration") + double getMaxDuration(); + + @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Standard Deviation Duration") + double getStandardDeviationDuration(); + + String getName(); + + String getSource(); + +} \ No newline at end of file diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ObjectNameLocator.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ObjectNameLocator.java new file mode 100644 index 0000000000..648e727f27 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ObjectNameLocator.java @@ -0,0 +1,32 @@ +/* + * 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.Map; + +/** + * @author Dave Syer + * + * @since 2.0 + * + */ +public interface ObjectNameLocator { + + String getObjectName(String beanName); + + Map getObjectNames(); + +} \ No newline at end of file diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMonitor.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMonitor.java new file mode 100644 index 0000000000..6fc6c8c89e --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMonitor.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.AtomicLong; + +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 PollableChannelMonitor extends SimpleMessageChannelMonitor { + + private final AtomicLong receiveCount = new AtomicLong(); + + private final AtomicLong receiveErrorCount = new AtomicLong(); + + /** + * @param name + */ + public PollableChannelMonitor(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 long getReceiveCount() { + return receiveCount.get(); + } + + @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Errors") + public long getReceiveErrorCount() { + return receiveErrorCount.get(); + } + + @Override + public String toString() { + return String.format("MessageChannelMonitor: [name=%s, sends=%d, receives=%d]", getName(), getSendCount(), + receiveCount.get()); + } + +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/QueueChannelMonitor.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/QueueChannelMonitor.java new file mode 100644 index 0000000000..edf528b12a --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/QueueChannelMonitor.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 QueueChannelMonitor extends PollableChannelMonitor { + + private final QueueChannel channel; + + /** + * @param name + */ + public QueueChannelMonitor(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(); + } + +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageChannelMonitor.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageChannelMonitor.java new file mode 100644 index 0000000000..e17e830d44 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageChannelMonitor.java @@ -0,0 +1,182 @@ +/* + * 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 SimpleMessageChannelMonitor implements MethodInterceptor, MessageChannelMonitor { + + 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 ExponentialMovingAverageCumulativeHistory sendDuration = new ExponentialMovingAverageCumulativeHistory( + DEFAULT_MOVING_AVERAGE_WINDOW); + + private final ExponentialMovingAverageRateCumulativeHistory sendErrorRate = new ExponentialMovingAverageRateCumulativeHistory( + ONE_SECOND_SECONDS, ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW); + + private final ExponentialMovingAverageRatioCumulativeHistory sendSuccessRatio = new ExponentialMovingAverageRatioCumulativeHistory( + ONE_MINUTE_SECONDS, DEFAULT_MOVING_AVERAGE_WINDOW); + + private final ExponentialMovingAverageRateCumulativeHistory sendRate = new ExponentialMovingAverageRateCumulativeHistory( + 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 SimpleMessageChannelMonitor(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(); + sendSuccessRatio.success(); + sendDuration.append(timer.getTotalTimeSeconds()); + return result; + + } + catch (Throwable e) { + sendErrorCount.incrementAndGet(); + sendSuccessRatio.failure(); + 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 getSendRate() { + return sendRate.getMean(); + } + + @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Channel Error Rate per Second") + public double getErrorRate() { + return sendErrorRate.getMean(); + } + + @ManagedMetric(metricType = MetricType.GAUGE, displayName = "Mean Channel Error Ratio per Minute") + public double getErrorRatio() { + 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(); + } + + @Override + public String toString() { + return String.format("MessageChannelMonitor: [name=%s, sends=%d]", name, sendCount.get()); + } +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMonitor.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMonitor.java new file mode 100644 index 0000000000..793da869f3 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMonitor.java @@ -0,0 +1,137 @@ +/** + * + */ +package org.springframework.integration.monitor; + +import java.util.concurrent.atomic.AtomicInteger; + +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 SimpleMessageHandlerMonitor implements MessageHandler, MessageHandlerMonitor { + + private static final Log logger = LogFactory.getLog(SimpleMessageHandlerMonitor.class); + + private static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10; + + private final MessageHandler handler; + + private final AtomicInteger handleCount = new AtomicInteger(); + + private final AtomicInteger errorCount = new AtomicInteger(); + + private final ExponentialMovingAverageCumulativeHistory duration = new ExponentialMovingAverageCumulativeHistory( + DEFAULT_MOVING_AVERAGE_WINDOW); + + private String name; + + private String source; + + public SimpleMessageHandlerMonitor(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 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(); + + handler.handleMessage(message); + + timer.stop(); + duration.append(timer.getTotalTimeSeconds()); + } catch (RuntimeException e) { + errorCount.incrementAndGet(); + throw e; + } catch (Error e) { + errorCount.incrementAndGet(); + throw e; + } + } + + @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(); + } + + @Override + public String toString() { + 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/test/java/log4j.properties b/spring-integration-jmx/src/test/java/log4j.properties new file mode 100644 index 0000000000..364ed2ec18 --- /dev/null +++ b/spring-integration-jmx/src/test/java/log4j.properties @@ -0,0 +1,10 @@ +log4j.rootCategory=WARN, stdout + +log4j.appender.stdout=org.apache.log4j.ConsoleAppender +log4j.appender.stdout.layout=org.apache.log4j.PatternLayout +log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m%n + + +log4j.category.org.springframework=WARN +log4j.category.org.springframework.integration=DEBUG +log4j.category.org.springframework.integration.monitor=TRACE diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/IgnoredTestSuite.java b/spring-integration-jmx/src/test/java/org/springframework/integration/IgnoredTestSuite.java new file mode 100644 index 0000000000..19164f9608 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/IgnoredTestSuite.java @@ -0,0 +1,37 @@ +package org.springframework.integration; + +import org.junit.Ignore; +import org.junit.runner.RunWith; +import org.junit.runners.Suite; +import org.junit.runners.Suite.SuiteClasses; +import org.springframework.integration.control.ControlBusXmlTests; +import org.springframework.integration.monitor.MessageChannelsMonitorIntegrationTests; + +/* + * 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. + */ + +/** + * A test suite that is ignored, but can be resurrected to help debug ordering issues in tests. + * + * @author Dave Syer + * + */ +@RunWith(Suite.class) +@SuiteClasses(value = { ControlBusXmlTests.class, MessageChannelsMonitorIntegrationTests.class }) +@Ignore +public class IgnoredTestSuite { + +} diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusOperationChannelTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusOperationChannelTests.java index 657bc4d828..ca36142f16 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusOperationChannelTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusOperationChannelTests.java @@ -19,10 +19,9 @@ package org.springframework.integration.control; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; -import javax.management.MBeanServer; - import org.junit.Test; - +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.Message; @@ -30,8 +29,9 @@ import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.jmx.JmxHeaders; +import org.springframework.integration.monitor.IntegrationMBeanExporter; import org.springframework.integration.support.MessageBuilder; -import org.springframework.jmx.support.JmxUtils; +import org.springframework.jmx.support.MBeanServerFactoryBean; /** * @author Mark Fisher @@ -39,8 +39,6 @@ import org.springframework.jmx.support.JmxUtils; */ public class ControlBusOperationChannelTests { - private final MBeanServer server = JmxUtils.locateMBeanServer(); - private final String domain = "domain.test"; @@ -49,12 +47,9 @@ public class ControlBusOperationChannelTests { GenericApplicationContext context = new GenericApplicationContext(); RootBeanDefinition endpointDef = new RootBeanDefinition(EventDrivenConsumer.class); endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new DirectChannel()); - endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new TestHandler()); + endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new RootBeanDefinition(TestHandler.class)); context.registerBeanDefinition("testEndpoint", endpointDef); - RootBeanDefinition busDef = new RootBeanDefinition(ControlBus.class); - busDef.getConstructorArgumentValues().addGenericArgumentValue(server); - busDef.getConstructorArgumentValues().addGenericArgumentValue(domain); - context.registerBeanDefinition("controlBus", busDef); + registerControlBus(context, domain); context.refresh(); ControlBus controlBus = context.getBean("controlBus", ControlBus.class); EventDrivenConsumer endpoint = context.getBean("testEndpoint", EventDrivenConsumer.class); @@ -74,6 +69,20 @@ public class ControlBusOperationChannelTests { context.close(); } + private BeanDefinition registerControlBus(GenericApplicationContext context, String domain) { + RootBeanDefinition serverDef = new RootBeanDefinition(MBeanServerFactoryBean.class); + serverDef.getPropertyValues().add("locateExistingServerIfPossible", true); + context.registerBeanDefinition("mbeanServer", serverDef); + BeanDefinition exporterDef = new RootBeanDefinition(IntegrationMBeanExporter.class); + exporterDef.getPropertyValues().addPropertyValue("server", new RuntimeBeanReference("mbeanServer")); + exporterDef.getPropertyValues().addPropertyValue("domain", domain); + context.registerBeanDefinition("exporter", exporterDef); + BeanDefinition controlBusDef = new RootBeanDefinition(ControlBus.class); + controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("mbeanServer")); + controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("exporter")); + context.registerBeanDefinition("controlBus", controlBusDef); + return exporterDef; + } private static class TestHandler implements MessageHandler { public void handleMessage(Message message) { diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusTests.java index 07d4f29b22..190de35f7f 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusTests.java @@ -17,6 +17,7 @@ package org.springframework.integration.control; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import java.util.Collections; import java.util.concurrent.atomic.AtomicInteger; @@ -29,20 +30,21 @@ import javax.management.ObjectName; import org.junit.After; import org.junit.Before; import org.junit.Test; - import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.Message; -import org.springframework.integration.MessageChannel; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.handler.BridgeHandler; -import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.monitor.IntegrationMBeanExporter; +import org.springframework.integration.monitor.LifecycleMessageHandlerMonitor; +import org.springframework.integration.monitor.QueueChannelMonitor; +import org.springframework.integration.monitor.SimpleMessageChannelMonitor; import org.springframework.jmx.support.MBeanServerFactoryBean; import org.springframework.jmx.support.ObjectNameManager; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -76,62 +78,51 @@ public class ControlBusTests { this.context.close(); } - @Test public void directChannelRegistered() throws Exception { context.registerBeanDefinition("directChannel", new RootBeanDefinition(DirectChannel.class)); - BeanDefinition controlBusDef = new RootBeanDefinition(ControlBus.class); - controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("mbeanServer")); - controlBusDef.getConstructorArgumentValues().addGenericArgumentValue("domain.test1"); - context.registerBeanDefinition("controlBus", controlBusDef); + registerControlBus(context, "domain.test1"); context.refresh(); MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class); - ObjectInstance instance = mbeanServer.getObjectInstance( - ObjectNameManager.getInstance("domain.test1:type=channel,name=directChannel")); - assertEquals(DirectChannel.class.getName(), instance.getClassName()); + ObjectInstance instance = mbeanServer.getObjectInstance(ObjectNameManager + .getInstance("domain.test1:type=MessageChannel,name=directChannel")); + assertEquals(SimpleMessageChannelMonitor.class.getName(), instance.getClassName()); } @Test public void anonymousDirectChannelRegistered() throws Exception { - context.registerBeanDefinition("org.springframework.integration.generated#0", new RootBeanDefinition(DirectChannel.class)); - BeanDefinition controlBusDef = new RootBeanDefinition(ControlBus.class); - controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("mbeanServer")); - controlBusDef.getConstructorArgumentValues().addGenericArgumentValue("domain.test1b"); - context.registerBeanDefinition("controlBus", controlBusDef); + context.registerBeanDefinition("org.springframework.integration.generated#0", new RootBeanDefinition( + DirectChannel.class)); + registerControlBus(context, "domain.test1b"); context.refresh(); MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class); - ObjectInstance instance = mbeanServer.getObjectInstance( - ObjectNameManager.getInstance("domain.test1b:type=channel,name=anonymous,generated=org.springframework.integration.generated#0")); - assertEquals(DirectChannel.class.getName(), instance.getClassName()); + ObjectInstance instance = mbeanServer + .getObjectInstance(ObjectNameManager + .getInstance("domain.test1b:type=MessageChannel,name=org.springframework.integration.generated#0,source=anonymous")); + assertEquals(SimpleMessageChannelMonitor.class.getName(), instance.getClassName()); } @Test public void staticObjectNamePropertiesRegistered() throws Exception { context.registerBeanDefinition("directChannel", new RootBeanDefinition(DirectChannel.class)); - BeanDefinition controlBusDef = new RootBeanDefinition(ControlBus.class); - controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("mbeanServer")); - controlBusDef.getConstructorArgumentValues().addGenericArgumentValue("domain.test1a"); - controlBusDef.getPropertyValues().add("objectNameStaticProperties", Collections.singletonMap("foo","bar")); - context.registerBeanDefinition("controlBus", controlBusDef); + BeanDefinition exporterDef = registerControlBus(context, "domain.test1a"); + exporterDef.getPropertyValues().add("objectNameStaticProperties", Collections.singletonMap("foo", "bar")); context.refresh(); MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class); - ObjectInstance instance = mbeanServer.getObjectInstance( - ObjectNameManager.getInstance("domain.test1a:type=channel,foo=bar,name=directChannel")); - assertEquals(DirectChannel.class.getName(), instance.getClassName()); + ObjectInstance instance = mbeanServer.getObjectInstance(ObjectNameManager + .getInstance("domain.test1a:type=MessageChannel,name=directChannel,foo=bar")); + assertEquals(SimpleMessageChannelMonitor.class.getName(), instance.getClassName()); } @Test public void queueChannelRegistered() throws Exception { context.registerBeanDefinition("queueChannel", new RootBeanDefinition(QueueChannel.class)); - BeanDefinition controlBusDef = new RootBeanDefinition(ControlBus.class); - controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("mbeanServer")); - controlBusDef.getConstructorArgumentValues().addGenericArgumentValue("domain.test2"); - context.registerBeanDefinition("controlBus", controlBusDef); + registerControlBus(context, "domain.test2"); context.refresh(); MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class); - ObjectInstance instance = mbeanServer.getObjectInstance( - ObjectNameManager.getInstance("domain.test2:type=channel,name=queueChannel")); - assertEquals(QueueChannel.class.getName(), instance.getClassName()); + ObjectInstance instance = mbeanServer.getObjectInstance(ObjectNameManager + .getInstance("domain.test2:type=MessageChannel,name=queueChannel")); + assertEquals(QueueChannelMonitor.class.getName(), instance.getClassName()); } @Test @@ -139,17 +130,14 @@ public class ControlBusTests { context.registerBeanDefinition("testChannel", new RootBeanDefinition(DirectChannel.class)); RootBeanDefinition endpointDef = new RootBeanDefinition(EventDrivenConsumer.class); endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("testChannel")); - endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new BridgeHandler()); + endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new RootBeanDefinition(BridgeHandler.class)); context.registerBeanDefinition("eventDrivenConsumer", endpointDef); - BeanDefinition controlBusDef = new RootBeanDefinition(ControlBus.class); - controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("mbeanServer")); - controlBusDef.getConstructorArgumentValues().addGenericArgumentValue("domain.test3"); - context.registerBeanDefinition("controlBus", controlBusDef); + registerControlBus(context, "domain.test3"); context.refresh(); MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class); - ObjectInstance instance = mbeanServer.getObjectInstance( - ObjectNameManager.getInstance("domain.test3:type=endpoint,name=eventDrivenConsumer")); - assertEquals(EventDrivenConsumer.class.getName(), instance.getClassName()); + ObjectInstance instance = mbeanServer.getObjectInstance(ObjectNameManager + .getInstance("domain.test3:type=MessageHandler,name=eventDrivenConsumer,bean=endpoint")); + assertEquals(LifecycleMessageHandlerMonitor.class.getName(), instance.getClassName()); } @Test @@ -157,19 +145,16 @@ public class ControlBusTests { context.registerBeanDefinition("testChannel", new RootBeanDefinition(QueueChannel.class)); RootBeanDefinition endpointDef = new RootBeanDefinition(PollingConsumer.class); endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("testChannel")); - endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new BridgeHandler()); + endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new RootBeanDefinition(BridgeHandler.class)); endpointDef.getPropertyValues().add("trigger", new PeriodicTrigger(10000)); context.registerBeanDefinition("pollingConsumer", endpointDef); context.registerBeanDefinition("taskScheduler", new RootBeanDefinition(ThreadPoolTaskScheduler.class)); - BeanDefinition controlBusDef = new RootBeanDefinition(ControlBus.class); - controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("mbeanServer")); - controlBusDef.getConstructorArgumentValues().addGenericArgumentValue("domain.test4"); - context.registerBeanDefinition("controlBus", controlBusDef); + registerControlBus(context, "domain.test4"); context.refresh(); MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class); - ObjectInstance instance = mbeanServer.getObjectInstance( - ObjectNameManager.getInstance("domain.test4:type=endpoint,name=pollingConsumer")); - assertEquals(PollingConsumer.class.getName(), instance.getClassName()); + ObjectInstance instance = mbeanServer.getObjectInstance(ObjectNameManager + .getInstance("domain.test4:type=MessageHandler,name=pollingConsumer,bean=endpoint")); + assertEquals(LifecycleMessageHandlerMonitor.class.getName(), instance.getClassName()); } @Test @@ -180,6 +165,7 @@ public class ControlBusTests { endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("testChannel")); endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new MessageHandler() { private final AtomicInteger count = new AtomicInteger(); + public void handleMessage(Message message) { int current = count.incrementAndGet(); if (current % 10 == 0) { @@ -188,24 +174,23 @@ public class ControlBusTests { } }); context.registerBeanDefinition("testEndpoint", endpointDef); + registerControlBus(context, "domain.channel.monitor"); + context.refresh(); + MBeanServer server = context.getBean("mbeanServer", MBeanServer.class); + ObjectName objectName = ObjectNameManager.getInstance("domain.channel.monitor:type=MessageChannel,name=testChannel"); + assertNotNull(server.getObjectInstance(objectName)); + } + + private BeanDefinition registerControlBus(GenericApplicationContext context, String domain) { + BeanDefinition exporterDef = new RootBeanDefinition(IntegrationMBeanExporter.class); + exporterDef.getPropertyValues().addPropertyValue("server", new RuntimeBeanReference("mbeanServer")); + exporterDef.getPropertyValues().addPropertyValue("domain", domain); + context.registerBeanDefinition("exporter", exporterDef); BeanDefinition controlBusDef = new RootBeanDefinition(ControlBus.class); controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("mbeanServer")); - controlBusDef.getConstructorArgumentValues().addGenericArgumentValue("domain.channel.monitor"); + controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("exporter")); context.registerBeanDefinition("controlBus", controlBusDef); - context.refresh(); - MessageChannel channel = context.getBean("testChannel", MessageChannel.class); - for (int i = 0; i < 100; i++) { - try { - channel.send(new GenericMessage("foo")); - } - catch (Exception e) { - // ignore - } - } - MBeanServer server = context.getBean("mbeanServer", MBeanServer.class); - ObjectName objectName = ObjectNameManager.getInstance("domain.channel.monitor:type=channel,name=testChannel"); - assertEquals(90L, server.getAttribute(objectName, "SendSuccessCount")); - assertEquals(10L, server.getAttribute(objectName, "SendErrorCount")); + return exporterDef; } } diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests-context.xml index 180db93a6d..68177ab054 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests-context.xml +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests-context.xml @@ -22,9 +22,14 @@ + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests.java index 3b9a8801b3..2250bc8547 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests.java @@ -26,11 +26,11 @@ import javax.management.ObjectInstance; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.endpoint.EventDrivenConsumer; -import org.springframework.integration.endpoint.PollingConsumer; +import org.springframework.integration.monitor.LifecycleMessageHandlerMonitor; +import org.springframework.integration.monitor.QueueChannelMonitor; +import org.springframework.integration.monitor.SimpleMessageChannelMonitor; import org.springframework.jmx.support.ObjectNameManager; +import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -41,7 +41,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @RunWith(SpringJUnit4ClassRunner.class) public class ControlBusXmlTests { - private static final String DOMAIN = "org.springframework.integration"; + private static final String DOMAIN = "spring.application"; @Autowired @@ -51,37 +51,39 @@ public class ControlBusXmlTests { @Test public void directChannelRegistered() throws Exception { ObjectInstance instance = mbeanServer.getObjectInstance( - ObjectNameManager.getInstance(DOMAIN + ":type=channel,name=testDirectChannel")); - assertEquals(DirectChannel.class.getName(), instance.getClassName()); + ObjectNameManager.getInstance(DOMAIN + ":type=MessageChannel,name=testDirectChannel")); + assertEquals(SimpleMessageChannelMonitor.class.getName(), instance.getClassName()); } @Test public void queueChannelRegistered() throws Exception { ObjectInstance instance = mbeanServer.getObjectInstance( - ObjectNameManager.getInstance(DOMAIN + ":type=channel,name=testQueueChannel")); - assertEquals(QueueChannel.class.getName(), instance.getClassName()); + ObjectNameManager.getInstance(DOMAIN + ":type=MessageChannel,name=testQueueChannel")); + assertEquals(QueueChannelMonitor.class.getName(), instance.getClassName()); } @Test public void eventDrivenConsumerRegistered() throws Exception { ObjectInstance instance = mbeanServer.getObjectInstance( - ObjectNameManager.getInstance(DOMAIN + ":type=endpoint,name=testEventDrivenBridge")); - assertEquals(EventDrivenConsumer.class.getName(), instance.getClassName()); + ObjectNameManager.getInstance(DOMAIN + ":type=MessageHandler,name=testEventDrivenBridge,bean=endpoint")); + assertEquals(LifecycleMessageHandlerMonitor.class.getName(), instance.getClassName()); } @Test public void anonymousConsumerRegistered() throws Exception { Set instances = mbeanServer.queryMBeans( - ObjectNameManager.getInstance(DOMAIN + ":type=endpoint,name=anonymous,*"), null); + ObjectNameManager.getInstance(DOMAIN + ":type=MessageHandler,bean=anonymous,*"), null); assertEquals(1, instances.size()); - assertEquals(EventDrivenConsumer.class.getName(), instances.iterator().next().getClassName()); + assertEquals(LifecycleMessageHandlerMonitor.class.getName(), instances.iterator().next().getClassName()); } @Test + // Assume this one runs last and force MBeans to be unregistered to avoid clashes in later tests + @DirtiesContext public void pollingConsumerRegistered() throws Exception { ObjectInstance instance = mbeanServer.getObjectInstance( - ObjectNameManager.getInstance(DOMAIN + ":type=endpoint,name=testPollingBridge")); - assertEquals(PollingConsumer.class.getName(), instance.getClassName()); + ObjectNameManager.getInstance(DOMAIN + ":type=MessageHandler,name=testPollingBridge,bean=endpoint")); + assertEquals(LifecycleMessageHandlerMonitor.class.getName(), instance.getClassName()); } } diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests-context.xml new file mode 100644 index 0000000000..4554eeaea2 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests-context.xml @@ -0,0 +1,24 @@ + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests.java new file mode 100644 index 0000000000..61bb68a8da --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChannelIntegrationTests.java @@ -0,0 +1,59 @@ +/* + * 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 static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.message.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class ChannelIntegrationTests { + + @Autowired + private MessageChannel requests; + + @Autowired + private PollableChannel intermediate; + + @Autowired + private IntegrationMBeanExporter messageChannelsMonitor; + + @Test + @DirtiesContext + public void testMessageChannelStatistics() throws Exception { + + requests.send(new GenericMessage("foo")); + + double duration = messageChannelsMonitor.getChannelMeanSendDuration("" + requests); + assertTrue("No statistics for requests channel", duration >= 0); + + duration = messageChannelsMonitor.getChannelMeanSendDuration("" + intermediate); + assertTrue("No statistics for intermediate channel", duration >= 0); + + assertNotNull(intermediate.receive(100L)); + double count = messageChannelsMonitor.getChannelReceiveCount("" + intermediate); + assertTrue("No statistics for intermediate channel", count >= 0); + + } + +} diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageCumulativeHistoryTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageCumulativeHistoryTests.java new file mode 100644 index 0000000000..b53c2f47fd --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageCumulativeHistoryTests.java @@ -0,0 +1,53 @@ +/* + * 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 static org.junit.Assert.assertEquals; + +import org.junit.Test; + +/** + * @author Dave Syer + * + */ +public class ExponentialMovingAverageCumulativeHistoryTests { + + private ExponentialMovingAverageCumulativeHistory history = new ExponentialMovingAverageCumulativeHistory(10); + + @Test + public void testGetCount() { + assertEquals(0, history.getCount()); + history.append(1); + assertEquals(1, history.getCount()); + } + + @Test + public void testGetMean() throws Exception { + assertEquals(0, history.getMean(), 0.01); + history.append(1); + history.append(1); + assertEquals(1, history.getMean(), 0.01); + } + + @Test + public void testGetStandardDeviation() throws Exception { + assertEquals(0, history.getStandardDeviation(), 0.01); + history.append(1); + history.append(1); + assertEquals(0, history.getStandardDeviation(), 0.01); + } + +} diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageRateCumulativeHistoryTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageRateCumulativeHistoryTests.java new file mode 100644 index 0000000000..ec8f3ff328 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageRateCumulativeHistoryTests.java @@ -0,0 +1,77 @@ +/* + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; + +/** + * @author Dave Syer + * + */ +public class ExponentialMovingAverageRateCumulativeHistoryTests { + + private ExponentialMovingAverageRateCumulativeHistory history = new ExponentialMovingAverageRateCumulativeHistory( + 1., 10., 10); + + @Test + public void testGetCount() { + assertEquals(0, history.getCount()); + history.increment(); + assertEquals(1, history.getCount()); + } + + @Test + public void testGetTimeSinceLastMeasurement() throws Exception { + history.increment(); + Thread.sleep(20L); + assertTrue(history.getTimeSinceLastMeasurement() > 0); + } + + @Test + public void testGetEarlyMean() throws Exception { + assertEquals(0, history.getMean(), 0.01); + Thread.sleep(20L); + history.increment(); + assertEquals(50, history.getMean(), 10); + } + + @Test + public void testGetMean() throws Exception { + assertEquals(0, history.getMean(), 0.01); + Thread.sleep(20L); + history.increment(); + Thread.sleep(20L); + history.increment(); + assertEquals(50, history.getMean(), 10); + Thread.sleep(20L); + assertEquals(35, history.getMean(), 10); + } + + @Test + public void testGetStandardDeviation() throws Exception { + assertEquals(0, history.getStandardDeviation(), 0.01); + Thread.sleep(20L); + history.increment(); + Thread.sleep(22L); + history.increment(); + Thread.sleep(18L); + assertEquals(1.5, history.getStandardDeviation(), 1); + } + +} diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageRatioCumulativeHistoryTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageRatioCumulativeHistoryTests.java new file mode 100644 index 0000000000..f0602e4e0e --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageRatioCumulativeHistoryTests.java @@ -0,0 +1,117 @@ +/* + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import org.junit.Test; + +/** + * @author Dave Syer + * + */ +public class ExponentialMovingAverageRatioCumulativeHistoryTests { + + private ExponentialMovingAverageRatioCumulativeHistory history = new ExponentialMovingAverageRatioCumulativeHistory( + 0.5, 10); + + @Test + public void testGetCount() { + assertEquals(0, history.getCount()); + history.success(); + assertEquals(1, history.getCount()); + } + + @Test + public void testGetTimeSinceLastMeasurement() throws Exception { + history.success(); + Thread.sleep(20L); + assertTrue(history.getTimeSinceLastMeasurement() > 0); + } + + @Test + public void testGetEarlyMean() throws Exception { + assertEquals(0, history.getMean(), 0.01); + history.success(); + assertEquals(1, history.getMean(), 0.01); + } + + @Test + public void testGetEarlyFailure() throws Exception { + assertEquals(0, history.getMean(), 0.01); + history.failure(); + assertEquals(0, history.getMean(), 0.01); + } + + @Test + public void testDecayedMean() throws Exception { + history.failure(); + Thread.sleep(200L); + assertEquals(average(0, Math.exp(-0.4)), history.getMean(), 0.01); + } + + @Test + public void testGetMean() throws Exception { + assertEquals(0, history.getMean(), 0.01); + history.success(); + assertEquals(1, history.getMean(), 0.01); + history.success(); + assertEquals(1, history.getMean(), 0.01); + history.success(); + assertEquals(1, history.getMean(), 0.01); + } + + @Test + public void testGetMeanFailuresHighRate() throws Exception { + assertEquals(0, history.getMean(), 0.01); + history.success(); + assertEquals(average(1), history.getMean(), 0.01); + history.failure(); + assertEquals(average(1, 0.5), history.getMean(), 0.1); + history.success(); + assertEquals(average(1, 0.5, 0.67), history.getMean(), 0.1); + } + + @Test + public void testGetMeanFailuresLowRate() throws Exception { + assertEquals(0, history.getMean(), 0.01); + history.failure(); + assertEquals(average(0), history.getMean(), 0.01); + history.failure(); + assertEquals(average(0, 0), history.getMean(), 0.01); + history.success(); + assertEquals(average(0, 0, 0.33), history.getMean(), 0.1); + } + + @Test + public void testGetStandardDeviation() throws Exception { + assertEquals(0, history.getStandardDeviation(), 0.01); + history.success(); + assertEquals(0, history.getStandardDeviation(), 1); + } + + private double average(double... values) { + int count = 0; + double sum = 0; + for (double d : values) { + sum += d; + count++; + } + return sum / count; + } + +} diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/HandlerMonitoringIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/HandlerMonitoringIntegrationTests.java new file mode 100644 index 0000000000..82e2fe6f35 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/HandlerMonitoringIntegrationTests.java @@ -0,0 +1,127 @@ +/* + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.aspectj.lang.annotation.Aspect; +import org.aspectj.lang.annotation.Before; +import org.junit.Test; +import org.springframework.beans.factory.config.AutowireCapableBeanFactory; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.message.GenericMessage; + +public class HandlerMonitoringIntegrationTests { + + private static Log logger = LogFactory.getLog(HandlerMonitoringIntegrationTests.class); + + private MessageChannel channel; + + private Service service; + + private IntegrationMBeanExporter messageHandlersMonitor; + + public void setMessageHandlersMonitor(IntegrationMBeanExporter messageHandlersMonitor) { + this.messageHandlersMonitor = messageHandlersMonitor; + } + + public void setService(Service service) { + this.service = service; + } + + @Test + public void testSendAndHandleWithEndpointName() throws Exception { + // The handler monitor is registered under the endpoint id (since it is explicit) + doTest("explicit-handler.xml", "input", "explicit"); + } + + @Test + public void testSendAndHandleWithAnonymousHandler() throws Exception { + doTest("anonymous-handler.xml", "anonymous", "anonymous"); + } + + @Test + public void testSendAndHandleWithProxiedHandler() throws Exception { + doTest("proxy-handler.xml", "anonymous", "anonymous"); + } + + @Test + public void testErrorLogger() throws Exception { + + ClassPathXmlApplicationContext context = createContext("anonymous-handler.xml", "anonymous"); + try { + assertTrue(messageHandlersMonitor.getHandlerNames().contains("errorLogger")); + } + finally { + context.close(); + } + + } + + private void doTest(String config, String channelName, String monitor) throws Exception { + + ClassPathXmlApplicationContext context = createContext(config, channelName); + + try { + + int before = service.getCounter(); + channel.send(new GenericMessage("bar")); + assertEquals(before + 1, service.getCounter()); + + double duration = messageHandlersMonitor.getHandlerMeanDuration(monitor); + assertTrue("No statistics for input channel", duration > 0); + + } finally { + context.close(); + } + + } + + private ClassPathXmlApplicationContext createContext(String config, String channelName) { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(config, getClass()); + context.getAutowireCapableBeanFactory().autowireBeanProperties(this, AutowireCapableBeanFactory.AUTOWIRE_BY_TYPE, false); + channel = context.getBean(channelName, MessageChannel.class); + return context; + } + + public static interface Service { + void execute(String input) throws Exception; + int getCounter(); + } + + public static class SimpleService implements Service { + private int counter; + + public void execute(String input) throws Exception { + Thread.sleep(10L); // make the duration non-zero + counter++; + } + + public int getCounter() { + return counter; + } + } + + @Aspect + public static class HandlerInterceptor { + @Before("execution(* *..*Tests*(String)) && args(input)") + public void around(String input) { + logger.debug("Handling: "+input); + } + } + +} diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageChannelsMonitorIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageChannelsMonitorIntegrationTests.java new file mode 100644 index 0000000000..5e93db558c --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageChannelsMonitorIntegrationTests.java @@ -0,0 +1,211 @@ +/* + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.aspectj.lang.annotation.Aspect; +import org.aspectj.lang.annotation.Before; +import org.junit.Test; +import org.springframework.beans.factory.config.AutowireCapableBeanFactory; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.MessageHandlingException; +import org.springframework.integration.message.GenericMessage; + +public class MessageChannelsMonitorIntegrationTests { + + private static Log logger = LogFactory.getLog(MessageChannelsMonitorIntegrationTests.class); + + private MessageChannel channel; + + private Service service; + + private IntegrationMBeanExporter messageChannelsMonitor; + + public void setMessageHandlersMonitor(IntegrationMBeanExporter messageChannelsMonitor) { + this.messageChannelsMonitor = messageChannelsMonitor; + } + + public void setService(Service service) { + this.service = service; + } + + @Test + public void testSendWithAnonymousHandler() throws Exception { + doTest("anonymous-channel.xml", "anonymous"); + } + + @Test + public void testSendWithProxiedChannel() throws Exception { + doTest("proxy-channel.xml", "anonymous"); + } + + @Test + public void testRates() throws Exception { + + ClassPathXmlApplicationContext context = createContext("anonymous-channel.xml", "anonymous"); + + try { + + int before = service.getCounter(); + for (int i = 0; i < 50; i++) { + channel.send(new GenericMessage("bar")); + Thread.sleep(20L); + } + assertEquals(before + 50, service.getCounter()); + + // The handler monitor is registered under the endpoint id (since it is explicit) + double sends = messageChannelsMonitor.getChannelSendCount("" + channel); + assertEquals("No send statistics for input channel", 50, sends, 0.01); + double rate = messageChannelsMonitor.getChannelSendRate("" + channel); + assertTrue(String.format("Unexpected rate statistics for input channel %f %f", 60000. / 20L, rate), + 60000. / 20L > rate && rate > 0); + + } + finally { + context.close(); + } + + } + + @Test + public void testErrors() throws Exception { + + ClassPathXmlApplicationContext context = createContext("anonymous-channel.xml", "anonymous"); + try { + int before = service.getCounter(); + for (int i = 0; i < 5; i++) { + channel.send(new GenericMessage("bar")); + Thread.sleep(20L); + } + try { + channel.send(new GenericMessage("fail")); + } + catch (MessageHandlingException e) { + // ignore + } + for (int i = 0; i < 5; i++) { + channel.send(new GenericMessage("bar")); + Thread.sleep(20L); + } + assertEquals(before + 10, service.getCounter()); + + // The handler monitor is registered under the endpoint id (since it is explicit) + double sends = messageChannelsMonitor.getChannelSendCount("" + channel); + assertEquals("No send statistics for input channel", 11, sends, 0.01); + double errors = messageChannelsMonitor.getChannelSendErrorCount("" + channel); + assertEquals("No error statistics for input channel", 1, errors, 0.01); + double rate = messageChannelsMonitor.getChannelErrorRate("" + channel); + assertTrue(String.format("Unexpected error statistics for input channel %f %f", 60000. / 20L, rate), + 60000. / 20L > rate && rate > 0); + } + finally { + context.close(); + } + + } + + @Test + public void testQueues() throws Exception { + + ClassPathXmlApplicationContext context = createContext("queue-channel.xml", "queue"); + try { + int before = service.getCounter(); + for (int i = 0; i < 5; i++) { + channel.send(new GenericMessage("bar")); + Thread.sleep(20L); + } + try { + channel.send(new GenericMessage("fail")); + } + catch (MessageHandlingException e) { + // ignore + } + for (int i = 0; i < 5; i++) { + channel.send(new GenericMessage("bar")); + Thread.sleep(20L); + } + assertEquals(before + 10, service.getCounter()); + + // The handler monitor is registered under the endpoint id (since it is explicit) + long sends = messageChannelsMonitor.getChannelSendCount("" + channel); + assertEquals("No send statistics for input channel", 11, sends); + long receives = messageChannelsMonitor.getChannelReceiveCount("" + channel); + assertEquals("No send statistics for input channel", 11, receives); + long errors = messageChannelsMonitor.getChannelSendErrorCount("" + channel); + assertEquals("Expect no errors for input channel (handler fails)", 0, errors); + } + finally { + context.close(); + } + + } + + private void doTest(String config, String channelName) throws Exception { + + ClassPathXmlApplicationContext context = createContext(config, channelName); + + try { + + int before = service.getCounter(); + channel.send(new GenericMessage("bar")); + assertEquals(before + 1, service.getCounter()); + + // The handler monitor is registered under the endpoint id (since it is explicit) + double sends = messageChannelsMonitor.getChannelSendCount("" + channel); + assertEquals("No statistics for input channel", 1, sends, 0.01); + + } + finally { + context.close(); + } + + } + + private ClassPathXmlApplicationContext createContext(String config, String channelName) { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(config, getClass()); + context.getAutowireCapableBeanFactory().autowireBeanProperties(this, + AutowireCapableBeanFactory.AUTOWIRE_BY_TYPE, false); + channel = context.getBean(channelName, MessageChannel.class); + return context; + } + + public static class Service { + private int counter; + + public void execute(String input) { + if ("fail".equals(input)) { + throw new RuntimeException("Planned"); + } + counter++; + } + + public int getCounter() { + return counter; + } + } + + @Aspect + public static class ChannelInterceptor { + @Before("execution(* *..MessageChannel+.send(*)) && args(input)") + public void around(Message input) { + logger.debug("Handling: " + input); + } + } + +} diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/anonymous-channel.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/anonymous-channel.xml new file mode 100644 index 0000000000..faefa69571 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/anonymous-channel.xml @@ -0,0 +1,16 @@ + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/anonymous-handler.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/anonymous-handler.xml new file mode 100644 index 0000000000..734196369e --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/anonymous-handler.xml @@ -0,0 +1,16 @@ + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/common-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/common-context.xml new file mode 100644 index 0000000000..ae917d0fc0 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/common-context.xml @@ -0,0 +1,16 @@ + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/explicit-handler.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/explicit-handler.xml new file mode 100644 index 0000000000..a033a2a9cf --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/explicit-handler.xml @@ -0,0 +1,16 @@ + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/proxy-channel.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/proxy-channel.xml new file mode 100644 index 0000000000..6a5efa49fc --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/proxy-channel.xml @@ -0,0 +1,22 @@ + + + + + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/proxy-handler.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/proxy-handler.xml new file mode 100644 index 0000000000..8a69266d59 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/proxy-handler.xml @@ -0,0 +1,22 @@ + + + + + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/queue-channel.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/queue-channel.xml new file mode 100644 index 0000000000..78f03ab0e9 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/queue-channel.xml @@ -0,0 +1,20 @@ + + + + + + + + + + + + + + + +