diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java index 2b627f1ade..96bd24ca1a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java @@ -40,6 +40,7 @@ import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.handler.advice.HandleMessageAdvice; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -185,30 +186,30 @@ public class ConsumerEndpointFactoryBean } } } + if (!CollectionUtils.isEmpty(this.adviceChain)) { /* * ARPMHs advise the handleRequestMessage method internally and already have the advice chain injected. - * So we only advise handlers that are not reply-producing. If the handler is already advised, + * So we only advise handlers that are not reply-producing. + * Or if one (or more) of advices is IdempotentReceiverInterceptor. + * If the handler is already advised, * add the configured advices to its chain, otherwise create a proxy. */ Class targetClass = AopUtils.getTargetClass(this.handler); + boolean replyMessageHandler = AbstractReplyProducingMessageHandler.class.isAssignableFrom(targetClass); - if (!(AbstractReplyProducingMessageHandler.class.isAssignableFrom(targetClass))) { - if (AopUtils.isAopProxy(this.handler)) { - for (Advice advice : this.adviceChain) { - NameMatchMethodPointcutAdvisor handlerAdvice = new NameMatchMethodPointcutAdvisor(advice); - handlerAdvice.addMethodName("handleMessage"); - if (AopUtils.canApply(handlerAdvice.getPointcut(), targetClass)) { - ((Advised) this.handler).addAdvice(advice); - } + for (Advice advice : this.adviceChain) { + if (!replyMessageHandler || advice instanceof HandleMessageAdvice) { + NameMatchMethodPointcutAdvisor handlerAdvice = new NameMatchMethodPointcutAdvisor(advice); + handlerAdvice.addMethodName("handleMessage"); + if (this.handler instanceof Advised) { + ((Advised) this.handler).addAdvisor(handlerAdvice); } - } - else { - ProxyFactory proxyFactory = new ProxyFactory(this.handler); - for (Advice advice : this.adviceChain) { - proxyFactory.addAdvice(advice); + else { + ProxyFactory proxyFactory = new ProxyFactory(this.handler); + proxyFactory.addAdvisor(handlerAdvice); + this.handler = (MessageHandler) proxyFactory.getProxy(this.beanClassLoader); } - this.handler = (MessageHandler) proxyFactory.getProxy(this.beanClassLoader); } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java index ce16f8df88..dfc7496cae 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java @@ -32,6 +32,7 @@ import org.springframework.aop.framework.Advised; import org.springframework.aop.framework.ProxyFactory; import org.springframework.aop.support.DefaultBeanFactoryPointcutAdvisor; import org.springframework.aop.support.NameMatchMethodPointcut; +import org.springframework.aop.support.NameMatchMethodPointcutAdvisor; import org.springframework.beans.factory.NoSuchBeanDefinitionException; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.support.BeanDefinitionValidationException; @@ -55,6 +56,7 @@ import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.integration.handler.AbstractMessageProducingHandler; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.handler.advice.HandleMessageAdvice; import org.springframework.integration.router.AbstractMessageRouter; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.support.channel.BeanFactoryChannelResolver; @@ -70,6 +72,7 @@ import org.springframework.scheduling.support.CronTrigger; import org.springframework.scheduling.support.PeriodicTrigger; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; +import org.springframework.util.CollectionUtils; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -134,8 +137,15 @@ public abstract class AbstractMethodAnnotationPostProcessor adviceChain = extractAdviceChain(beanName, annotations); + MessageHandler handler = createHandler(bean, method, annotations); - setAdviceChainIfPresent(beanName, annotations, handler); + + if (!CollectionUtils.isEmpty(adviceChain) && handler instanceof AbstractReplyProducingMessageHandler) { + ((AbstractReplyProducingMessageHandler) handler).setAdviceChain(adviceChain); + } + if (handler instanceof Orderable) { Order orderAnnotation = AnnotationUtils.findAnnotation(method, Order.class); if (orderAnnotation != null) { @@ -189,6 +199,23 @@ public abstract class AbstractMethodAnnotationPostProcessor annotations, MessageHandler handler) { + private List extractAdviceChain(String beanName, List annotations) { + List adviceChain = null; String[] adviceChainNames = MessagingAnnotationUtils.resolveAttribute(annotations, ADVICE_CHAIN_ATTRIBUTE, String[].class); /* @@ -226,10 +254,7 @@ public abstract class AbstractMethodAnnotationPostProcessor 0) { - if (!(handler instanceof AbstractReplyProducingMessageHandler)) { - throw new IllegalArgumentException("Cannot apply advice chain to " + handler.getClass().getName()); - } - List adviceChain = new ArrayList(); + adviceChain = new ArrayList(); for (String adviceChainName : adviceChainNames) { Object adviceChainBean = this.beanFactory.getBean(adviceChainName); if (adviceChainBean instanceof Advice) { @@ -247,11 +272,11 @@ public abstract class AbstractMethodAnnotationPostProcessor annotations) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java index 5dbcba0edf..ca6c2f9dab 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java @@ -22,6 +22,7 @@ import org.aopalliance.aop.Advice; import org.springframework.aop.framework.ProxyFactory; import org.springframework.beans.factory.BeanClassLoaderAware; +import org.springframework.integration.handler.advice.HandleMessageAdvice; import org.springframework.messaging.Message; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -81,10 +82,16 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa super.onInit(); if (!CollectionUtils.isEmpty(this.adviceChain)) { ProxyFactory proxyFactory = new ProxyFactory(new AdvisedRequestHandler()); + boolean advised = false; for (Advice advice : this.adviceChain) { - proxyFactory.addAdvice(advice); + if (!(advice instanceof HandleMessageAdvice)) { + proxyFactory.addAdvice(advice); + advised = true; + } + } + if (advised) { + this.advisedRequestHandler = (RequestHandler) proxyFactory.getProxy(this.beanClassLoader); } - this.advisedRequestHandler = (RequestHandler) proxyFactory.getProxy(this.beanClassLoader); } doInit(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/AbstractHandleMessageAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/AbstractHandleMessageAdvice.java new file mode 100644 index 0000000000..bbd6aa8739 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/AbstractHandleMessageAdvice.java @@ -0,0 +1,65 @@ +/* + * Copyright 2016 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.handler.advice; + +import java.lang.reflect.Method; + +import org.aopalliance.intercept.MethodInvocation; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHandler; + +/** + * The base {@link HandleMessageAdvice} for advices which can be applied only + * for the {@link MessageHandler#handleMessage(Message)}. + * + * @author Artem Bilan + * @since 4.3.1 + */ +public abstract class AbstractHandleMessageAdvice implements HandleMessageAdvice { + + protected final Log logger = LogFactory.getLog(this.getClass()); + + @Override + public final Object invoke(MethodInvocation invocation) throws Throwable { + Method method = invocation.getMethod(); + Object invocationThis = invocation.getThis(); + Object[] arguments = invocation.getArguments(); + boolean isMessageHandler = invocationThis != null && invocationThis instanceof MessageHandler; + boolean isMessageMethod = method.getName().equals("handleMessage") + && (arguments.length == 1 && arguments[0] instanceof Message); + if (!isMessageHandler || !isMessageMethod) { + if (this.logger.isWarnEnabled()) { + String clazzName = invocationThis == null + ? method.getDeclaringClass().getName() + : invocationThis.getClass().getName(); + this.logger.warn("This advice " + getClass().getName() + + " can only be used for MessageHandlers; an attempt to advise method '" + + method.getName() + "' in '" + clazzName + "' is ignored."); + } + return invocation.proceed(); + } + + Message message = (Message) arguments[0]; + return doInvoke(invocation, message); + } + + protected abstract Object doInvoke(MethodInvocation invocation, Message message) throws Throwable; + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/HandleMessageAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/HandleMessageAdvice.java new file mode 100644 index 0000000000..d098112212 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/HandleMessageAdvice.java @@ -0,0 +1,30 @@ +/* + * Copyright 2016 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.handler.advice; + +import org.aopalliance.intercept.MethodInterceptor; + +/** + * The marker {@link MethodInterceptor} interface extension + * to distinguish advices for some reason. + * + * @author Artem Bilan + * @since 4.3.1 + */ +public interface HandleMessageAdvice extends MethodInterceptor { + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/IdempotentReceiverInterceptor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/IdempotentReceiverInterceptor.java index 3d6416cdc7..0a9009098b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/IdempotentReceiverInterceptor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/IdempotentReceiverInterceptor.java @@ -16,12 +16,8 @@ package org.springframework.integration.handler.advice; -import java.lang.reflect.Method; - 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.beans.BeansException; import org.springframework.beans.factory.BeanFactory; @@ -59,9 +55,7 @@ import org.springframework.util.Assert; * @see org.springframework.integration.selector.MetadataStoreSelector * @see org.springframework.integration.config.IdempotentReceiverAutoProxyCreatorInitializer */ -public class IdempotentReceiverInterceptor implements MethodInterceptor, BeanFactoryAware { - - protected final Log logger = LogFactory.getLog(this.getClass()); +public class IdempotentReceiverInterceptor extends AbstractHandleMessageAdvice implements BeanFactoryAware { private final MessagingTemplate messagingTemplate = new MessagingTemplate(); @@ -142,26 +136,7 @@ public class IdempotentReceiverInterceptor implements MethodInterceptor, BeanFac } @Override - public Object invoke(MethodInvocation invocation) throws Throwable { - Method method = invocation.getMethod(); - Object invocationThis = invocation.getThis(); - Object[] arguments = invocation.getArguments(); - boolean isMessageHandler = invocationThis != null && invocationThis instanceof MessageHandler; - boolean isMessageMethod = method.getName().equals("handleMessage") - && (arguments.length == 1 && arguments[0] instanceof Message); - if (!isMessageHandler || !isMessageMethod) { - if (this.logger.isWarnEnabled()) { - String clazzName = invocationThis == null - ? method.getDeclaringClass().getName() - : invocationThis.getClass().getName(); - this.logger.warn("This advice " + this.getClass().getName() + - " can only be used for MessageHandlers; an attempt to advise method '" - + method.getName() + "' in '" + clazzName + "' is ignored"); - } - return invocation.proceed(); - } - - Message message = (Message) arguments[0]; + protected Object doInvoke(MethodInvocation invocation, Message message) throws Throwable { boolean accept = this.messageSelector.accept(message); if (!accept) { boolean discarded = false; @@ -175,7 +150,7 @@ public class IdempotentReceiverInterceptor implements MethodInterceptor, BeanFac } if (!discarded) { - arguments[0] = getMessageBuilderFactory().fromMessage(message) + invocation.getArguments()[0] = getMessageBuilderFactory().fromMessage(message) .setHeader(IntegrationMessageHeaderAccessor.DUPLICATE_MESSAGE, true).build(); } else { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiatorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiatorTests.java index 01724d276c..455e96528d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiatorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiatorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2015 the original author or authors. + * Copyright 2012-2016 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. @@ -24,6 +24,7 @@ import java.util.concurrent.TimeUnit; import org.junit.Before; import org.junit.Test; + import org.springframework.integration.leader.Context; import org.springframework.integration.leader.DefaultCandidate; import org.springframework.integration.leader.event.LeaderEventPublisher; @@ -32,17 +33,20 @@ import org.springframework.integration.support.locks.LockRegistry; /** * @author Dave Syer - * + * @since 4.3.1 */ public class LockRegistryLeaderInitiatorTests { private CountDownLatch granted = new CountDownLatch(1); + private CountDownLatch revoked = new CountDownLatch(1); + private LockRegistry registry = new DefaultLockRegistry(); + private CountingPublisher publisher = new CountingPublisher(this.granted, this.revoked); - private LockRegistryLeaderInitiator initiator = new LockRegistryLeaderInitiator( - this.registry, new DefaultCandidate()); + private LockRegistryLeaderInitiator initiator = + new LockRegistryLeaderInitiator(this.registry, new DefaultCandidate()); @Before public void init() { @@ -54,12 +58,12 @@ public class LockRegistryLeaderInitiatorTests { assertThat(this.initiator.getContext().isLeader(), is(false)); this.initiator.start(); assertThat(this.initiator.isRunning(), is(true)); - this.granted.await(2, TimeUnit.SECONDS); + this.granted.await(10, TimeUnit.SECONDS); assertThat(this.initiator.getContext().isLeader(), is(true)); Thread.sleep(200L); assertThat(this.initiator.getContext().isLeader(), is(true)); this.initiator.stop(); - this.revoked.await(2, TimeUnit.SECONDS); + this.revoked.await(10, TimeUnit.SECONDS); assertThat(this.initiator.getContext().isLeader(), is(false)); } @@ -68,29 +72,31 @@ public class LockRegistryLeaderInitiatorTests { assertThat(this.initiator.getContext().isLeader(), is(false)); this.initiator.start(); assertThat(this.initiator.isRunning(), is(true)); - this.granted.await(2, TimeUnit.SECONDS); + this.granted.await(10, TimeUnit.SECONDS); assertThat(this.initiator.getContext().isLeader(), is(true)); this.initiator.getContext().yield(); - assertThat(this.revoked.await(2, TimeUnit.SECONDS), is(true)); + assertThat(this.revoked.await(10, TimeUnit.SECONDS), is(true)); } @Test public void competing() throws Exception { - LockRegistryLeaderInitiator another = new LockRegistryLeaderInitiator(this.registry, - new DefaultCandidate()); + LockRegistryLeaderInitiator another = + new LockRegistryLeaderInitiator(this.registry, new DefaultCandidate()); CountDownLatch other = new CountDownLatch(1); another.setLeaderEventPublisher(new CountingPublisher(other)); this.initiator.start(); - assertThat(this.granted.await(2, TimeUnit.SECONDS), is(true)); + assertThat(this.granted.await(10, TimeUnit.SECONDS), is(true)); another.start(); this.initiator.stop(); - assertThat(other.await(2, TimeUnit.SECONDS), is(true)); + assertThat(other.await(10, TimeUnit.SECONDS), is(true)); assertThat(another.getContext().isLeader(), is(true)); } private static class CountingPublisher implements LeaderEventPublisher { - private CountDownLatch granted; - private CountDownLatch revoked; + + private final CountDownLatch granted; + + private final CountDownLatch revoked; CountingPublisher(CountDownLatch granted, CountDownLatch revoked) { this.granted = granted; @@ -110,6 +116,7 @@ public class LockRegistryLeaderInitiatorTests { public void publishOnGranted(Object source, Context context, String role) { this.granted.countDown(); } + } } diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/IdempotentReceiverIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/IdempotentReceiverIntegrationTests.java index 896c790205..d032b54b7a 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/IdempotentReceiverIntegrationTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/IdempotentReceiverIntegrationTests.java @@ -232,8 +232,7 @@ public class IdempotentReceiverIntegrationTests { @Bean @org.springframework.integration.annotation.Transformer(inputChannel = "input", - outputChannel = "output", adviceChain = "fooAdvice") - @IdempotentReceiver("idempotentReceiverInterceptor") + outputChannel = "output", adviceChain = {"fooAdvice", "idempotentReceiverInterceptor"}) public Transformer transformer() { return new Transformer() { diff --git a/src/reference/asciidoc/handler-advice.adoc b/src/reference/asciidoc/handler-advice.adoc index e4cddef408..aafa627446 100644 --- a/src/reference/asciidoc/handler-advice.adoc +++ b/src/reference/asciidoc/handler-advice.adoc @@ -1,6 +1,9 @@ [[message-handler-advice-chain]] === Adding Behavior to Endpoints +[[mhac-intro]] +==== Introduction + Prior to Spring Integration _2.2_, you could add behavior to an entire Integration flow by adding an AOP Advice to a poller's `` element. However, let's say you want to retry, say, just a REST Web Service call, and not any downstream endpoints. @@ -452,6 +455,38 @@ For more information, see the http://docs.spring.io/spring-framework/docs/curren While the abstract class mentioned above is provided as a convenience, you can add any `Advice` to the chain, including a transaction advice. +[[handle-message-advice]] +==== Handle Message Advice + +As discussed in <>, advice objects in a request handler advice chain are applied to just the current endpoint, not the downstream flow (if any). +For `MessageHandler` s that produce a reply (`AbstractReplyProducingMessageHandler`), the advice is applied to an internal method +`handleRequestMessage()` (called from `MessageHandler.handleMessage()`). +For other message handlers, the advice is applied to `MessageHandler.handleMessage()`. + +There are some circumstances where, even if a message handler is an `AbstractReplyProducingMessageHandler`, the advice must be applied to the `handleMessage` method - for example, the <> might return `null` and this would cause an exception if the handler's `replyRequired` property is true. + +Starting with _version 4.3.1_, a new `HandleMessageAdvice` and the `AbstractHandleMessageAdvice` base implementation have been introduced. +`Advice` s that implement `HandleMessageAdvice` will always be applied to the `handleMessage()` method, regardless of the handler type. + +It is important to understand that `HandleMessageAdvice` implementations (such as <>), when applied to a handler that returns a response, are dissociated from the `adviceChain` and properly applied to the `MessageHandler.handleMessage()` method. +Bear in mind, however, that this means the advice chain order is not complied with; and, with configuration such as: + +[source,xml] +---- + + + + + + +---- + +The `` is applied to the `AbstractReplyProducingMessageHandler.handleRequestMessage()`, but `myHandleMessageAdvice` is applied for to `MessageHandler.handleMessage()` and, therefore, invoked **before** the ``. +To retain the order, you should follow with standard http://docs.spring.io/spring/docs/current/spring-framework-reference/html/aop-api.html[Spring AOP] configuration approach and use endpoint `id` together with the `.handler` suffix to obtain the target `MessageHandler` bean. +Note, however, that in that case, the entire downstream flow would be within the transaction scope. + +In the case of a `MessageHandler` that does **not** return a response, the advice chain order is retained. + [[advising-filters]] ==== Advising Filters @@ -620,3 +655,6 @@ public MessageHandler myService() { .... } ---- + +NOTE: The `IdempotentReceiverInterceptor` is designed only for the `MessageHandler.handleMessage(Message)` method and starting with _version 4.3.1_ it implements `HandleMessageAdvice`, with the `AbstractHandleMessageAdvice` as a base class, for better dissociation. +See <> for more information.