diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/RouterFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/RouterFactoryBean.java index 4abfdb3346..ddc902ac7b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/RouterFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/RouterFactoryBean.java @@ -1,21 +1,20 @@ /* * 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. + * + * 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.config; +import org.springframework.aop.TargetSource; +import org.springframework.aop.framework.Advised; import org.springframework.integration.MessageChannel; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.router.AbstractChannelNameResolvingMessageRouter; @@ -48,7 +47,6 @@ public class RouterFactoryBean extends AbstractMessageHandlerFactoryBean { private volatile Boolean ignoreSendFailures; - public void setChannelResolver(ChannelResolver channelResolver) { this.channelResolver = channelResolver; } @@ -79,9 +77,49 @@ public class RouterFactoryBean extends AbstractMessageHandlerFactoryBean { @Override MessageHandler createMethodInvokingHandler(Object targetObject, String targetMethodName) { + Assert.notNull(targetObject, "target object must not be null"); - AbstractMessageRouter router = this.createRouter(targetObject, targetMethodName); - return this.configureRouter(router); + AbstractMessageRouter router = extractRouter(targetObject); + + if (router == null) { + router = this.createRouter(targetObject, targetMethodName); + this.configureRouter(router); + return router; + } + + Assert.isTrue(!StringUtils.hasText(targetMethodName), "target method should not be provided when the target " + + "object is an implementation of AbstractMessageRouter"); + this.configureRouter(router); + + if (targetObject instanceof MessageHandler) { + return (MessageHandler) targetObject; + } + return router; + + } + + private AbstractMessageRouter extractRouter(Object targetObject) { + if (targetObject instanceof AbstractMessageRouter) { + return (AbstractMessageRouter) targetObject; + } + if (targetObject instanceof Advised) { + return extractAopTarget((Advised) targetObject); + } + return null; + } + + private AbstractMessageRouter extractAopTarget(Advised advised) { + TargetSource targetSource = advised.getTargetSource(); + if (targetSource == null) { + return null; + } + Object target; + try { + target = targetSource.getTarget(); + } catch (Exception e) { + throw new IllegalStateException(e); + } + return extractRouter(target); } @Override @@ -90,21 +128,13 @@ public class RouterFactoryBean extends AbstractMessageHandlerFactoryBean { } private AbstractMessageRouter createRouter(Object targetObject, String targetMethodName) { - if (targetObject instanceof AbstractMessageRouter) { - Assert.isTrue(!StringUtils.hasText(targetMethodName), - "target method should not be provided when the target " + - "object is an implementation of AbstractMessageRouter"); - return (AbstractMessageRouter) targetObject; - } - MethodInvokingRouter router = (StringUtils.hasText(targetMethodName)) - ? new MethodInvokingRouter(targetObject, targetMethodName) - : new MethodInvokingRouter(targetObject); + MethodInvokingRouter router = (StringUtils.hasText(targetMethodName)) ? new MethodInvokingRouter(targetObject, + targetMethodName) : new MethodInvokingRouter(targetObject); return router; } private AbstractMessageRouter configureRouter(AbstractMessageRouter router) { - if (this.channelResolver != null && - router instanceof AbstractChannelNameResolvingMessageRouter) { + if (this.channelResolver != null && router instanceof AbstractChannelNameResolvingMessageRouter) { ((AbstractChannelNameResolvingMessageRouter) router).setChannelResolver(this.channelResolver); } if (this.defaultOutputChannel != null) { @@ -114,10 +144,11 @@ public class RouterFactoryBean extends AbstractMessageHandlerFactoryBean { router.setTimeout(timeout.longValue()); } if (this.ignoreChannelNameResolutionFailures != null) { - Assert.isTrue(router instanceof AbstractChannelNameResolvingMessageRouter, + Assert.isTrue(router instanceof AbstractChannelNameResolvingMessageRouter, "The 'ignoreChannelNameResolutionFailures' property can only be set on routers that extend " - + AbstractChannelNameResolvingMessageRouter.class.getName()); - ((AbstractChannelNameResolvingMessageRouter) router).setIgnoreChannelNameResolutionFailures(ignoreChannelNameResolutionFailures); + + AbstractChannelNameResolvingMessageRouter.class.getName()); + ((AbstractChannelNameResolvingMessageRouter) router) + .setIgnoreChannelNameResolutionFailures(ignoreChannelNameResolutionFailures); } if (this.applySequence != null) { router.setApplySequence(this.applySequence); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PayloadTypeRouterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PayloadTypeRouterParser.java index 8bf6023e29..5a87301842 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PayloadTypeRouterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/PayloadTypeRouterParser.java @@ -20,6 +20,7 @@ import java.util.List; import org.w3c.dom.Element; +import org.springframework.beans.BeanMetadataElement; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.BeanDefinitionBuilder; @@ -39,14 +40,13 @@ import org.springframework.util.xml.DomUtils; public class PayloadTypeRouterParser extends AbstractRouterParser { @Override - @SuppressWarnings("unchecked") protected BeanDefinition parseRouter(Element element, ParserContext parserContext) { BeanDefinitionBuilder payloadTypeRouterBuilder = BeanDefinitionBuilder.genericBeanDefinition( IntegrationNamespaceUtils.BASE_PACKAGE + ".router.PayloadTypeRouter"); List childElements = DomUtils.getChildElementsByTagName(element, "mapping"); Assert.notEmpty(childElements, "Type mapping must be provided (e.g., )"); - ManagedMap channelMap = new ManagedMap(); + ManagedMap channelMap = new ManagedMap(); for (Element childElement : childElements) { String typeName = childElement.getAttribute("type"); ClassLoader classLoader = parserContext.getReaderContext().getBeanClassLoader(); 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 d64d367fc6..d704b7a52c 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 @@ -167,8 +167,9 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP else { monitor = new SimpleMessageHandlerMonitor((MessageHandler) bean); } + Object advised = applyHandlerInterceptor(bean, monitor, beanClassLoader); handlers.add(monitor); - return monitor; + return advised; } if (bean instanceof MessageChannel) { DirectChannelMonitor monitor; @@ -404,6 +405,12 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP return applyAdvice(bean, channelsAdvice, beanClassLoader); } + private Object applyHandlerInterceptor(Object bean, SimpleMessageHandlerMonitor interceptor, ClassLoader beanClassLoader) { + NameMatchMethodPointcutAdvisor handlerAdvice = new NameMatchMethodPointcutAdvisor(interceptor); + handlerAdvice.addMethodName("handleMessage"); + return applyAdvice(bean, handlerAdvice, beanClassLoader); + } + private Object extractTarget(Object bean) { if (!(bean instanceof Advised)) { return bean; 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 index 67d99257df..0247a47289 100644 --- 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 @@ -17,6 +17,8 @@ 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; @@ -36,7 +38,7 @@ import org.springframework.util.StopWatch; * */ @ManagedResource -public class SimpleMessageHandlerMonitor implements MessageHandler, MessageHandlerMonitor { +public class SimpleMessageHandlerMonitor implements MethodInterceptor, MessageHandlerMonitor { private static final Log logger = LogFactory.getLog(SimpleMessageHandlerMonitor.class); @@ -81,7 +83,17 @@ public class SimpleMessageHandlerMonitor implements MessageHandler, MessageHandl return handler; } - public void handleMessage(Message message) throws MessageRejectedException, MessageHandlingException, + public Object invoke(MethodInvocation invocation) throws Throwable { + String method = invocation.getMethod().getName(); + if ("handleMessage".equals(method)) { + Message message = (Message) invocation.getArguments()[0]; + handleMessage(message); + return null; + } + return invocation.proceed(); + } + + private void handleMessage(Message message) throws MessageRejectedException, MessageHandlingException, MessageDeliveryException { if (logger.isTraceEnabled()) { logger.trace("messageHandler(" + handler + ") message(" + message + ") :"); diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests.java index 6d76d6a468..01cd5bceb9 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests.java @@ -40,7 +40,7 @@ public class MBeanExporterParserTests { private ApplicationContext context; @Test - public void test() throws InterruptedException { + public void testMBeanExporterExists() throws InterruptedException { IntegrationMBeanExporter exporter = this.context.getBean(IntegrationMBeanExporter.class); MBeanServer server = this.context.getBean("mbs", MBeanServer.class); assertEquals(server, exporter.getServer()); diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/RouterMBeanTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/RouterMBeanTests-context.xml new file mode 100644 index 0000000000..31190c0e19 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/RouterMBeanTests-context.xml @@ -0,0 +1,30 @@ + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/RouterMBeanTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/RouterMBeanTests.java new file mode 100644 index 0000000000..6f79b88245 --- /dev/null +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/RouterMBeanTests.java @@ -0,0 +1,47 @@ +/* + * 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 RouterMBeanTests { + + @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(); + } + +}