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

INT-1806: added ManagedEndpoint for endpoints that are not message sources or message handlers

Browse Source
This commit is contained in:
Dave Syer
2011-05-03 10:04:15 +01:00
parent 4009ca25b3
commit 3441971540
8 changed files with 346 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<String, MessageSourceMetrics> sourcesByName = new HashMap<String, MessageSourceMetrics>();
private Map<String, String> beansByEndpointName = new HashMap<String, String>();
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);

54
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ManagedEndpoint.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.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();
}
}

65
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java
View File

@@ -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<ObjectName> 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<ObjectName> 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<Date> {
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;
}
}
}

95
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageSourceMonitoringIntegrationTests.java Normal file
View File

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

20
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/anonymous-source.xml Normal file
View File

@@ -0,0 +1,20 @@
<?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"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd">
<import resource="common-context.xml" />
<int:channel id="anonymous">
<int:queue />
</int:channel>
<int:inbound-channel-adapter ref="service" method="execute" channel="anonymous">
<int:poller fixed-rate="200" />
</int:inbound-channel-adapter>
<bean id="service" class="org.springframework.integration.monitor.MessageSourceMonitoringIntegrationTests$SimpleService" />
</beans>

20
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/explicit-source.xml Normal file
View File

@@ -0,0 +1,20 @@
<?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"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd">
<import resource="common-context.xml" />
<int:channel id="input">
<int:queue />
</int:channel>
<int:inbound-channel-adapter id="explicit" ref="service" method="execute" channel="input">
<int:poller fixed-rate="200" />
</int:inbound-channel-adapter>
<bean id="service" class="org.springframework.integration.monitor.MessageSourceMonitoringIntegrationTests$SimpleService" />
</beans>

19
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-no-source.xml Normal file
View File

@@ -0,0 +1,19 @@
<?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"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd">
<import resource="common-context.xml" />
<int:channel id="input"/>
<int:gateway id="gateway" service-interface="org.springframework.integration.monitor.MBeanExporterIntegrationTests$Service"
default-request-channel="input" />
<int:service-activator id="explicit" input-channel="input" ref="service" method="execute"/>
<bean id="service" class="org.springframework.integration.monitor.MBeanExporterIntegrationTests$SimpleService" />
</beans>

20
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/lifecycle-source.xml Normal file
View File

@@ -0,0 +1,20 @@
<?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"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd">
<import resource="common-context.xml" />
<int:channel id="input">
<int:queue />
</int:channel>
<int:inbound-channel-adapter id="explicit" ref="service" method="execute" channel="input">
<int:poller fixed-rate="200" />
</int:inbound-channel-adapter>
<bean id="service" class="org.springframework.integration.monitor.MBeanExporterIntegrationTests$SimpleService" />
</beans>