Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Expose MessageSources and their polling endpoints to JMX

Browse Source
This commit is contained in:
Dave Syer
2010-10-05 15:47:18 -07:00
parent 07043a8875
commit 76d17bd187
6 changed files with 401 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

139
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java
View File

@@ -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<Object, AtomicLong> anonymousCounters = new HashMap<Object, AtomicLong>();
private Map<Object, AtomicLong> anonymousHandlerCounters = new HashMap<Object, AtomicLong>();
private Map<Object, AtomicLong> anonymousSourceCounters = new HashMap<Object, AtomicLong>();
private Set<SimpleMessageHandlerMonitor> handlers = new HashSet<SimpleMessageHandlerMonitor>();
private Set<SimpleMessageSourceMonitor> sources = new HashSet<SimpleMessageSourceMonitor>();
private Set<DirectChannelMonitor> channels = new HashSet<DirectChannelMonitor>();
private Map<String, DirectChannelMonitor> channelsByName = new HashMap<String, DirectChannelMonitor>();
private Map<String, MessageHandlerMonitor> handlersByName = new HashMap<String, MessageHandlerMonitor>();
private Map<String, MessageSourceMonitor> sourcesByName = new HashMap<String, MessageSourceMonitor>();
private Map<String, String> objectNamesByName = new HashMap<String, String>();
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);

75
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageSourceMonitor.java Normal file
View File

@@ -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();
}
}

36
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageSourceMonitor.java Normal file
View File

@@ -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();
}

79
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageSourceMonitor.java Normal file
View File

@@ -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());
}
}

26
spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests-context.xml Normal file
View File

@@ -0,0 +1,26 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:jmx="http://www.springframework.org/schema/integration/jmx"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jmx
http://www.springframework.org/schema/integration/jmx/spring-integration-jmx.xsd">
<context:mbean-server id="mbs" />
<int:channel id="testChannel" />
<int:inbound-channel-adapter channel="testChannel" method="get">
<bean class="org.springframework.integration.jmx.config.PollingAdapterMBeanTests$Source"/>
<int:poller fixed-rate="5000"/>
</int:inbound-channel-adapter>
<int:logging-channel-adapter channel="testChannel"/>
</beans>

54
spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PollingAdapterMBeanTests.java Normal file
View File

@@ -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";
}
}
}