diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java index 0a9197effc..0fa94560c7 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java @@ -30,7 +30,6 @@ import javax.management.modelmbean.ModelMBean; import org.aopalliance.aop.Advice; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.aop.Advisor; import org.springframework.aop.PointcutAdvisor; import org.springframework.aop.TargetSource; @@ -121,6 +120,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP private Map sourcesByName = new HashMap(); + private Map beansByEndpointName = new HashMap(); + private ClassLoader beanClassLoader; private volatile boolean autoStartup = true; @@ -374,6 +375,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP registerChannels(); registerHandlers(); registerSources(); + registerEndpoints(); } /** @@ -446,6 +448,14 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP return null; } + public int getSourceMessageCount(String name) { + if (sourcesByName.containsKey(name)) { + return sourcesByName.get(name).getMessageCount(); + } + logger.debug("No source found for (" + name + ")"); + return -1; + } + public int getChannelReceiveCount(String name) { if (channelsByName.containsKey(name)) { if (channelsByName.get(name) instanceof PollableChannelMetrics) { @@ -541,6 +551,25 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } } + private void registerEndpoints() { + String[] names = beanFactory.getBeanNamesForType(AbstractEndpoint.class); + for (String name : names) { + if (!beansByEndpointName.values().contains(name)) { + AbstractEndpoint endpoint = beanFactory.getBean(name, AbstractEndpoint.class); + String beanKey; + name = endpoint.getComponentName(); + if (name.startsWith("_org.springframework.integration")) { + beanKey = getEndpointBeanKey(endpoint, getInternalComponentName(name), "internal"); + } + else { + beanKey = getEndpointBeanKey(endpoint, endpoint.getComponentName(), "endpoint"); + } + ObjectName objectName = registerBeanInstance(new ManagedEndpoint(endpoint), beanKey); + logger.info("Registered endpoint without MessageSource: " + objectName); + } + } + } + private Object applyChannelInterceptor(Object bean, DirectChannelMetrics interceptor, ClassLoader beanClassLoader) { NameMatchMethodPointcutAdvisor channelsAdvice = new NameMatchMethodPointcutAdvisor(interceptor); channelsAdvice.addMethodName("send"); @@ -620,6 +649,11 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP handler.getSource()); } + private String getEndpointBeanKey(AbstractEndpoint endpoint, String name, String source) { + // This ordering of keys seems to work with default settings of JConsole + return String.format(domain + ":type=ManagedEndpoint,name=%s,bean=%s" + getStaticNames(), name, source); + } + private String getStaticNames() { if (objectNameStaticProperties.isEmpty()) { return ""; @@ -644,6 +678,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP String[] names = beanFactory.getBeanNamesForType(AbstractEndpoint.class); String name = null; + String endpointName = null; String source = "endpoint"; Object endpoint = null; @@ -658,11 +693,12 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } if (field == monitor.getMessageHandler()) { name = beanName; + endpointName = beanName; break; } } if (name != null && endpoint != null && name.startsWith("_org.springframework.integration")) { - name = name.substring("_org.springframework.integration".length() + 1); + name = getInternalComponentName(name); source = "internal"; } if (name != null && endpoint != null && name.startsWith("org.springframework.integration")) { @@ -707,6 +743,10 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP source = "handler"; } + if (endpointName != null) { + beansByEndpointName.put(name, endpointName); + } + monitor.setSource(source); monitor.setName(name); @@ -714,6 +754,10 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } + private String getInternalComponentName(String name) { + return name.substring("_org.springframework.integration".length() + 1); + } + private MessageSourceMetrics enhanceSourceMonitor(SimpleMessageSourceMetrics monitor) { MessageSourceMetrics result = monitor; @@ -726,6 +770,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP String[] names = beanFactory.getBeanNamesForType(AbstractEndpoint.class); String name = null; + String endpointName = null; String source = "endpoint"; Object endpoint = null; @@ -740,11 +785,12 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } if (field == monitor.getMessageSource()) { name = beanName; + endpointName = beanName; break; } } if (name != null && endpoint != null && name.startsWith("_org.springframework.integration")) { - name = name.substring("_org.springframework.integration".length() + 1); + name = getInternalComponentName(name); source = "internal"; } if (name != null && endpoint != null && name.startsWith("org.springframework.integration")) { @@ -789,6 +835,10 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP source = "handler"; } + if (endpointName != null) { + beansByEndpointName.put(name, endpointName); + } + monitor.setSource(source); monitor.setName(name); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ManagedEndpoint.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ManagedEndpoint.java new file mode 100644 index 0000000000..7b00459e8b --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ManagedEndpoint.java @@ -0,0 +1,54 @@ +/* + * 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.integration.endpoint.AbstractEndpoint; +import org.springframework.jmx.export.annotation.ManagedAttribute; +import org.springframework.jmx.export.annotation.ManagedOperation; +import org.springframework.jmx.export.annotation.ManagedResource; + +/** + * Wrapper for an {@link AbstractEndpoint} that exposes a management interface. + * + * @author Dave Syer + * + */ +@ManagedResource +public class ManagedEndpoint implements Lifecycle { + + private final AbstractEndpoint delegate; + + public ManagedEndpoint(AbstractEndpoint delegate) { + this.delegate = delegate; + } + + @ManagedAttribute + public final boolean isRunning() { + return delegate.isRunning(); + } + + @ManagedOperation + public final void start() { + delegate.start(); + } + + @ManagedOperation + public final void stop() { + delegate.stop(); + } + +} diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java index 92f6e81c79..c9f33caf22 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java @@ -18,7 +18,9 @@ import static org.junit.Assert.assertTrue; import java.util.Arrays; import java.util.Date; +import java.util.Set; +import javax.management.MBeanOperationInfo; import javax.management.MBeanServer; import javax.management.ObjectName; @@ -83,6 +85,50 @@ public class MBeanExporterIntegrationTests { assertNotNull(messageChannelsMonitor); } + @Test + public void testLifecycleInEndpointWithMessageSource() throws Exception { + context = new GenericXmlApplicationContext(getClass(), "lifecycle-source.xml"); + messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class); + assertNotNull(messageChannelsMonitor); + MBeanServer server = context.getBean(MBeanServer.class); + Set names = server.queryNames(ObjectName.getInstance("org.springframework.integration:type=ManagedEndpoint,*"), null); + assertEquals(0, names.size()); + names = server.queryNames(ObjectName.getInstance("org.springframework.integration:name=explicit,*"), null); + assertEquals(1, names.size()); + MBeanOperationInfo[] operations = server.getMBeanInfo(names.iterator().next()).getOperations(); + String startName = null; + for (MBeanOperationInfo info : operations) { + String name = info.getName(); + if (name.startsWith("start")) { + startName = name; + } + } + // Lifecycle method name + assertEquals("start", startName); + } + + @Test + public void testLifecycleInEndpointWithoutMessageSource() throws Exception { + context = new GenericXmlApplicationContext(getClass(), "lifecycle-no-source.xml"); + messageChannelsMonitor = context.getBean(IntegrationMBeanExporter.class); + assertNotNull(messageChannelsMonitor); + MBeanServer server = context.getBean(MBeanServer.class); + Set names = server.queryNames(ObjectName.getInstance("org.springframework.integration:type=ManagedEndpoint,*"), null); + assertEquals(1, names.size()); + names = server.queryNames(ObjectName.getInstance("org.springframework.integration:name=gateway,*"), null); + assertEquals(1, names.size()); + MBeanOperationInfo[] operations = server.getMBeanInfo(names.iterator().next()).getOperations(); + String startName = null; + for (MBeanOperationInfo info : operations) { + String name = info.getName(); + if (name.startsWith("start")) { + startName = name; + } + } + // Lifecycle method name + assertEquals("start", startName); + } + public static class DateFactoryBean implements FactoryBean { private Date date; @@ -166,4 +212,23 @@ public class MBeanExporterIntegrationTests { } + public static interface Service { + String execute() throws Exception; + int getCounter(); + } + + public static class SimpleService implements Service { + private int counter; + + public String execute() throws Exception { + Thread.sleep(10L); // make the duration non-zero + counter++; + return "count="+counter; + } + + public int getCounter() { + return counter; + } + } + } diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageSourceMonitoringIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageSourceMonitoringIntegrationTests.java new file mode 100644 index 0000000000..446caf2e85 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageSourceMonitoringIntegrationTests.java @@ -0,0 +1,95 @@ +/* + * 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.assertTrue; + +import org.junit.Test; +import org.springframework.beans.factory.config.AutowireCapableBeanFactory; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.core.PollableChannel; + +public class MessageSourceMonitoringIntegrationTests { + + private PollableChannel channel; + + private Service service; + + private IntegrationMBeanExporter exporter; + + public void setMessageHandlersMonitor(IntegrationMBeanExporter exporter) { + this.exporter = exporter; + } + + public void setService(Service service) { + this.service = service; + } + + @Test + public void testSendAndHandleWithEndpointName() throws Exception { + // The message source monitor is registered under the endpoint id (since it is explicit) + doTest("explicit-source.xml", "input", "explicit"); + } + + @Test + public void testSendAndHandleWithAnonymous() throws Exception { + // The message source monitor is registered under the channel name + doTest("anonymous-source.xml", "anonymous", "anonymous"); + } + + private void doTest(String config, String channelName, String monitor) throws Exception { + + ClassPathXmlApplicationContext context = createContext(config, channelName); + + try { + + int before = service.getCounter(); + channel.receive(1000L); + assertTrue(before < service.getCounter()); + + int count = exporter.getSourceMessageCount(monitor); + assertTrue("No statistics for input channel", count > 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, PollableChannel.class); + return context; + } + + public static interface Service { + String execute() throws Exception; + int getCounter(); + } + + public static class SimpleService implements Service { + private int counter; + + public String execute() throws Exception { + Thread.sleep(10L); // make the duration non-zero + counter++; + return "count="+counter; + } + + public int getCounter() { + return counter; + } + } + +} diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/anonymous-source.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/anonymous-source.xml new file mode 100644 index 0000000000..eb8cd16c64 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/anonymous-source.xml @@ -0,0 +1,20 @@ + + + + + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/explicit-source.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/explicit-source.xml new file mode 100644 index 0000000000..f93fc252fb --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/explicit-source.xml @@ -0,0 +1,20 @@ + + + + + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-no-source.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-no-source.xml new file mode 100644 index 0000000000..33020fd5d2 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-no-source.xml @@ -0,0 +1,19 @@ + + + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-source.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-source.xml new file mode 100644 index 0000000000..e8e690b072 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-source.xml @@ -0,0 +1,20 @@ + + + + + + + + + + + + + + + +