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 d704b7a52c..015da2e62c 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 @@ -41,6 +41,7 @@ import org.springframework.integration.MessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.MessageProducer; +import org.springframework.integration.core.MessageSource; import org.springframework.integration.core.PollableChannel; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.jmx.export.MBeanExporter; @@ -92,16 +93,22 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP private ListableBeanFactory beanFactory; - private Map anonymousCounters = new HashMap(); + private Map anonymousHandlerCounters = new HashMap(); + + private Map anonymousSourceCounters = new HashMap(); private Set handlers = new HashSet(); + private Set sources = new HashSet(); + private Set channels = new HashSet(); private Map channelsByName = new HashMap(); private Map handlersByName = new HashMap(); + private Map sourcesByName = new HashMap(); + private Map objectNamesByName = new HashMap(); private ClassLoader beanClassLoader; @@ -170,6 +177,11 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP Object advised = applyHandlerInterceptor(bean, monitor, beanClassLoader); handlers.add(monitor); return advised; + } else if (bean instanceof MessageSource) { + SimpleMessageSourceMonitor monitor = new SimpleMessageSourceMonitor((MessageSource) bean); + Object advised = applySourceInterceptor(bean, monitor, beanClassLoader); + sources.add(monitor); + return advised; } if (bean instanceof MessageChannel) { DirectChannelMonitor monitor; @@ -268,6 +280,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP protected void doStart() { registerChannels(); registerHandlers(); + registerSources(); logger.info("Summary on start: " + objectNamesByName); } @@ -384,7 +397,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP private void registerHandlers() { for (SimpleMessageHandlerMonitor source : handlers) { - MessageHandlerMonitor monitor = enhanceMonitor(source); + MessageHandlerMonitor monitor = enhanceHandlerMonitor(source); String name = monitor.getName(); // Only register once... if (!handlersByName.containsKey(name)) { @@ -398,6 +411,22 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } } + private void registerSources() { + for (SimpleMessageSourceMonitor source : sources) { + MessageSourceMonitor monitor = enhanceSourceMonitor(source); + String name = monitor.getName(); + // Only register once... + if (!sourcesByName.containsKey(name)) { + String beanKey = getSourceBeanKey(monitor); + if (name != null) { + sourcesByName.put(name, monitor); + objectNamesByName.put(name, beanKey); + } + registerBeanNameOrInstance(monitor, beanKey); + } + } + } + private Object applyChannelInterceptor(Object bean, DirectChannelMonitor interceptor, ClassLoader beanClassLoader) { NameMatchMethodPointcutAdvisor channelsAdvice = new NameMatchMethodPointcutAdvisor(interceptor); channelsAdvice.addMethodName("send"); @@ -411,6 +440,12 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP return applyAdvice(bean, handlerAdvice, beanClassLoader); } + private Object applySourceInterceptor(Object bean, SimpleMessageSourceMonitor interceptor, ClassLoader beanClassLoader) { + NameMatchMethodPointcutAdvisor sourceAdvice = new NameMatchMethodPointcutAdvisor(interceptor); + sourceAdvice.addMethodName("receive"); + return applyAdvice(bean, sourceAdvice, beanClassLoader); + } + private Object extractTarget(Object bean) { if (!(bean instanceof Advised)) { return bean; @@ -458,6 +493,12 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP handler.getSource()); } + private String getSourceBeanKey(MessageSourceMonitor handler) { + // This ordering of keys seems to work with default settings of JConsole + return String.format(domain + ":type=MessageSource,name=%s,bean=%s" + getStaticNames(), handler.getName(), + handler.getSource()); + } + private String getStaticNames() { if (objectNameStaticProperties.isEmpty()) { return ""; @@ -469,7 +510,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP return builder.toString(); } - private MessageHandlerMonitor enhanceMonitor(SimpleMessageHandlerMonitor monitor) { + private MessageHandlerMonitor enhanceHandlerMonitor(SimpleMessageHandlerMonitor monitor) { MessageHandlerMonitor result = monitor; @@ -488,12 +529,12 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP endpoint = beanFactory.getBean(beanName); Object field = null; try { - field = getField(endpoint, "handler"); + field = extractTarget(getField(endpoint, "handler")); } catch (Exception e) { logger.debug("Could not get handler from bean = " + beanName); } - if (field == monitor) { + if (field == monitor.getMessageHandler()) { name = beanName; break; } @@ -517,10 +558,10 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } Object field = getField(target, "inputChannel"); if (field != null) { - if (!anonymousCounters.containsKey(field)) { - anonymousCounters.put(field, new AtomicLong()); + if (!anonymousHandlerCounters.containsKey(field)) { + anonymousHandlerCounters.put(field, new AtomicLong()); } - AtomicLong count = anonymousCounters.get(field); + AtomicLong count = anonymousHandlerCounters.get(field); long total = count.incrementAndGet(); String suffix = ""; /* @@ -551,6 +592,88 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } + private MessageSourceMonitor enhanceSourceMonitor(SimpleMessageSourceMonitor monitor) { + + MessageSourceMonitor 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 = extractTarget(getField(endpoint, "source")); + } + catch (Exception e) { + logger.debug("Could not get source from bean = " + beanName); + } + if (field == monitor.getMessageSource()) { + 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, "outputChannel"); + if (field != null) { + if (!anonymousSourceCounters.containsKey(field)) { + anonymousSourceCounters.put(field, new AtomicLong()); + } + AtomicLong count = anonymousSourceCounters.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 LifecycleMessageSourceMonitor((Lifecycle) endpoint, monitor); + } + + if (name == null) { + name = monitor.getMessageSource().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); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageSourceMonitor.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageSourceMonitor.java new file mode 100644 index 0000000000..fc69442bb1 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageSourceMonitor.java @@ -0,0 +1,75 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.monitor; + +import org.springframework.context.Lifecycle; +import org.springframework.jmx.export.annotation.ManagedAttribute; +import org.springframework.jmx.export.annotation.ManagedOperation; +import org.springframework.jmx.export.annotation.ManagedResource; + +/** + * A {@link MessageSourceMonitor} that exposes in addition the {@link Lifecycle} interface. The lifecycle methods can + * be used to stop and start polling endpoints, for instance, in a live system. + * + * @author Dave Syer + * + * @since 2.0 + * + */ +@ManagedResource +public class LifecycleMessageSourceMonitor implements MessageSourceMonitor, Lifecycle { + + private final Lifecycle lifecycle; + + private final MessageSourceMonitor delegate; + + public LifecycleMessageSourceMonitor(Lifecycle lifecycle, MessageSourceMonitor delegate) { + this.lifecycle = lifecycle; + this.delegate = delegate; + } + + @ManagedAttribute + public boolean isRunning() { + return lifecycle.isRunning(); + } + + @ManagedOperation + public void start() { + lifecycle.start(); + } + + @ManagedOperation + public void stop() { + lifecycle.stop(); + } + + public String getName() { + return delegate.getName(); + } + + public String getSource() { + return delegate.getSource(); + } + + /** + * @return + * @see org.springframework.integration.monitor.MessageSourceMonitor#getMessageCount() + */ + public int getMessageCount() { + return delegate.getMessageCount(); + } + +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageSourceMonitor.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageSourceMonitor.java new file mode 100644 index 0000000000..0b988a9968 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageSourceMonitor.java @@ -0,0 +1,36 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + +package org.springframework.integration.monitor; + +import org.springframework.jmx.export.annotation.ManagedMetric; +import org.springframework.jmx.support.MetricType; + +/** + * @author Dave Syer + * + * @since 2.0 + */ +public interface MessageSourceMonitor { + + /** + * @return the number of successful handler calls + */ + @ManagedMetric(metricType = MetricType.COUNTER, displayName = "Message Source Message Count", description = "rate=1h") + int getMessageCount(); + + String getName(); + + String getSource(); + +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageSourceMonitor.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageSourceMonitor.java new file mode 100644 index 0000000000..c54508eda1 --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageSourceMonitor.java @@ -0,0 +1,79 @@ +/* + * Copyright 2002-2010 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. + */ + +package org.springframework.integration.monitor; + +import java.util.concurrent.atomic.AtomicInteger; + +import org.aopalliance.intercept.MethodInterceptor; +import org.aopalliance.intercept.MethodInvocation; +import org.springframework.integration.core.MessageSource; + +/** + * @author Dave Syer + * + * @since 2.0 + */ +public class SimpleMessageSourceMonitor implements MethodInterceptor, MessageSourceMonitor { + + private final AtomicInteger messageCount = new AtomicInteger(); + + private final MessageSource messageSource; + + private String source; + + private String name; + + public SimpleMessageSourceMonitor(MessageSource messageSource) { + this.messageSource = messageSource; + } + + + public void setName(String name) { + this.name = name; + } + + public String getName() { + return name; + } + + public void setSource(String source) { + this.source = source; + } + + public String getSource() { + return this.source; + } + + public MessageSource getMessageSource() { + return messageSource; + } + + public int getMessageCount() { + return messageCount.get(); + } + + public Object invoke(MethodInvocation invocation) throws Throwable { + String method = invocation.getMethod().getName(); + if ("receive".equals(method)) { + messageCount.incrementAndGet(); + } + return invocation.proceed(); + } + + @Override + public String toString() { + return String.format("MessageSourceMonitor: [name=%s, source=%s, count=%d]", name, source, messageCount.get()); + } + +} diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests-context.xml new file mode 100644 index 0000000000..14b1f093d0 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests-context.xml @@ -0,0 +1,26 @@ + + + + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests.java new file mode 100644 index 0000000000..1a08698dce --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests.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.jmx.config; + +import static org.junit.Assert.assertEquals; + +import javax.management.MBeanServer; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.integration.monitor.IntegrationMBeanExporter; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Dave Syer + * @since 2.0 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class PollingAdapterMBeanTests { + + @Autowired + private ApplicationContext context; + + @Test + public void testMBeanExporterExists() throws InterruptedException { + IntegrationMBeanExporter exporter = this.context.getBean(IntegrationMBeanExporter.class); + MBeanServer server = this.context.getBean("mbs", MBeanServer.class); + assertEquals(server, exporter.getServer()); + exporter.destroy(); + } + + public static class Source { + public String get() { + System.err.println("*** " + System.currentTimeMillis()); + return "foo"; + } + } + +}