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

INT-4064: Simplify IdempotentReInt Java Config

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4064

We configure `IdempotentReceiver` via `<int:idempotent-receiver>` component or `@IdempotentReceiver` annotation.
In case of regular Java config, e.g. direct `ConsumerEndpointFactoryBean` usage or Java DSL,
it isn't possible to configure `idempotentReceiverInterceptor` enough easy

* Introduce `if...else` logic into `ConsumerEndpointFactoryBean` to proxy `MessageHandler`,
if `adviceChain` contains an newly-introduced `HandleMessageAdvice`.
And do that independently if `MessageHandler` is `AbstractReplyProducingMessageHandler`
* Make `idempotentReceiverInterceptor extends HandleMessageAdvice`
* Skip `HandleMessageAdvice` in the `AbstractReplyProducingMessageHandler`
* Add advice applying logic into the `AbstractMethodAnnotationPostProcessor` as well

Introduce `HandleMessageAdvice` marker interceptor to cover the case when an `Advice` can be advices as well.
Remove unused `setAdviceChainIfPresent()` method in the `AbstractMethodAnnotationPostProcessor`

Document `HandleMessageAdvice`

Increase wait latch timeouts in the `LockRegistryLeaderInitiatorTests`

Doc Polishing
This commit is contained in:
Artem Bilan
2016-07-01 17:29:30 -04:00
committed by Gary Russell
parent f74ddc2aa6
commit d9f9f9e3ca
9 changed files with 216 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
View File

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

41
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
View File

@@ -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<T extends Annotation
return null;
}
}
List<Advice> 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<T extends Annotation
}
}
if (!CollectionUtils.isEmpty(adviceChain)) {
for (Advice advice : adviceChain) {
if (advice instanceof HandleMessageAdvice) {
NameMatchMethodPointcutAdvisor handlerAdvice = new NameMatchMethodPointcutAdvisor(advice);
handlerAdvice.addMethodName("handleMessage");
if (handler instanceof Advised) {
((Advised) handler).addAdvisor(handlerAdvice);
}
else {
ProxyFactory proxyFactory = new ProxyFactory(handler);
proxyFactory.addAdvisor(handlerAdvice);
handler = (MessageHandler) proxyFactory.getProxy(this.beanFactory.getBeanClassLoader());
}
}
}
}
AbstractEndpoint endpoint = createEndpoint(handler, method, annotations);
if (endpoint != null) {
return endpoint;
@@ -217,7 +244,8 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
return true;
}
protected final void setAdviceChainIfPresent(String beanName, List<Annotation> annotations, MessageHandler handler) {
private List<Advice> extractAdviceChain(String beanName, List<Annotation> annotations) {
List<Advice> adviceChain = null;
String[] adviceChainNames = MessagingAnnotationUtils.resolveAttribute(annotations, ADVICE_CHAIN_ATTRIBUTE,
String[].class);
/*
@@ -226,10 +254,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
* by setting an empty array on the custom annotation.
*/
if (adviceChainNames != null && adviceChainNames.length > 0) {
if (!(handler instanceof AbstractReplyProducingMessageHandler)) {
throw new IllegalArgumentException("Cannot apply advice chain to " + handler.getClass().getName());
}
List<Advice> adviceChain = new ArrayList<Advice>();
adviceChain = new ArrayList<Advice>();
for (String adviceChainName : adviceChainNames) {
Object adviceChainBean = this.beanFactory.getBean(adviceChainName);
if (adviceChainBean instanceof Advice) {
@@ -247,11 +272,11 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
}
else {
throw new IllegalArgumentException("Invalid advice chain type:" +
adviceChainName.getClass().getName() + " for bean '" + beanName + "'");
adviceChainName.getClass().getName() + " for bean '" + beanName + "'");
}
}
((AbstractReplyProducingMessageHandler) handler).setAdviceChain(adviceChain);
}
return adviceChain;
}
protected AbstractEndpoint createEndpoint(MessageHandler handler, Method method, List<Annotation> annotations) {

11
spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java
View File

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

65
spring-integration-core/src/main/java/org/springframework/integration/handler/advice/AbstractHandleMessageAdvice.java Normal file
View File

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

30
spring-integration-core/src/main/java/org/springframework/integration/handler/advice/HandleMessageAdvice.java Normal file
View File

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

31
spring-integration-core/src/main/java/org/springframework/integration/handler/advice/IdempotentReceiverInterceptor.java
View File

@@ -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 {

35
spring-integration-core/src/test/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiatorTests.java
View File

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

3
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/IdempotentReceiverIntegrationTests.java
View File

@@ -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() {

38
src/reference/asciidoc/handler-advice.adoc
View File

@@ -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 `<advice-chain/>` 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 <<mhac-intro, the introduction to this section>>, 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 <<idempotent-receiver, Idempotent Receiver>> 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 <<idempotent-receiver, Idempotent Receiver>>), 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]
----
<some-reply-producing-endpoint ... >
<int:request-handler-advice-chain>
<tx:advice ... />
<bean ref="myHandleMessageAdvice" />
</int:request-handler-advice-chain>
</some-reply-producing-endpoint>
----
The `<tx:advice>` is applied to the `AbstractReplyProducingMessageHandler.handleRequestMessage()`, but `myHandleMessageAdvice` is applied for to `MessageHandler.handleMessage()` and, therefore, invoked **before** the `<tx:advice>`.
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 <<handle-message-advice>> for more information.