GH-2695: Add proxy option to @EnablePublisher (#2701)
* GH-2695: Add proxy option to @EnablePublisher Fixes spring-projects/spring-integration#2695 To configure a `proxyTargetClass=true` we need declare a `PublisherAnnotationBeanPostProcessor` bean manually, but that may cause a confuse when `@EnablePublisher` is still present. So, target service is proxied twice * Expose `proxyTargetClass` and `order` into the `@EnablePublisher` and `<enable-publisher>` * Refactor `PublisherAnnotationBeanPostProcessor` to extend an `AbstractBeanFactoryAwareAdvisingPostProcessor` to avoid AOP boilerplate code altogether * Add assertion into the `PublisherAnnotationBeanPostProcessor` to be sure that only one of its instance is present in the application context * * Polishing error message and Docs
This commit is contained in:
Artem Bilan
committed by
Gary Russell
Gary Russell
parent
cd8cbaa99c
commit
462dc98803
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2018 the original author or authors.
|
||||
* Copyright 2002-2019 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.
|
||||
@@ -16,26 +16,19 @@
|
||||
|
||||
package org.springframework.integration.aop;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
|
||||
import org.springframework.aop.framework.Advised;
|
||||
import org.springframework.aop.framework.ProxyConfig;
|
||||
import org.springframework.aop.framework.ProxyFactory;
|
||||
import org.springframework.aop.support.AopUtils;
|
||||
import org.springframework.beans.BeansException;
|
||||
import org.springframework.beans.factory.BeanClassLoaderAware;
|
||||
import org.springframework.aop.framework.autoproxy.AbstractBeanFactoryAwareAdvisingPostProcessor;
|
||||
import org.springframework.beans.factory.BeanCreationNotAllowedException;
|
||||
import org.springframework.beans.factory.BeanFactory;
|
||||
import org.springframework.beans.factory.BeanFactoryAware;
|
||||
import org.springframework.beans.factory.BeanNameAware;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.beans.factory.config.BeanPostProcessor;
|
||||
import org.springframework.core.Ordered;
|
||||
import org.springframework.util.ClassUtils;
|
||||
import org.springframework.beans.factory.NoUniqueBeanDefinitionException;
|
||||
|
||||
/**
|
||||
* Post-processes beans that contain the
|
||||
* method-level @{@link org.springframework.integration.annotation.Publisher} annotation.
|
||||
* <p>
|
||||
* Only one bean instance of this processor can be declared in the application context, manual
|
||||
* or automatic by thr framework via annotation or XML processing.
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
* @author Mark Fisher
|
||||
@@ -46,21 +39,14 @@ import org.springframework.util.ClassUtils;
|
||||
* @since 2.0
|
||||
*/
|
||||
@SuppressWarnings("serial")
|
||||
public class PublisherAnnotationBeanPostProcessor extends ProxyConfig
|
||||
implements BeanPostProcessor, BeanClassLoaderAware, BeanFactoryAware, InitializingBean, Ordered {
|
||||
public class PublisherAnnotationBeanPostProcessor extends AbstractBeanFactoryAwareAdvisingPostProcessor
|
||||
implements BeanNameAware, InitializingBean {
|
||||
|
||||
private volatile String defaultChannelName;
|
||||
private String defaultChannelName;
|
||||
|
||||
private volatile PublisherAnnotationAdvisor advisor;
|
||||
private String name;
|
||||
|
||||
private volatile int order = Ordered.LOWEST_PRECEDENCE;
|
||||
|
||||
private volatile BeanFactory beanFactory;
|
||||
|
||||
private volatile ClassLoader beanClassLoader = ClassUtils.getDefaultClassLoader();
|
||||
|
||||
private final Set<Class<?>> nonApplicableCache =
|
||||
Collections.newSetFromMap(new ConcurrentHashMap<Class<?>, Boolean>(256));
|
||||
private BeanFactory beanFactory;
|
||||
|
||||
/**
|
||||
* Set the default channel where Messages should be sent if the annotation
|
||||
@@ -72,64 +58,32 @@ public class PublisherAnnotationBeanPostProcessor extends ProxyConfig
|
||||
this.defaultChannelName = defaultChannelName;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setBeanName(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setBeanFactory(BeanFactory beanFactory) {
|
||||
this.beanFactory = beanFactory;
|
||||
super.setBeanFactory(beanFactory);
|
||||
PublisherAnnotationAdvisor publisherAnnotationAdvisor = new PublisherAnnotationAdvisor();
|
||||
publisherAnnotationAdvisor.setBeanFactory(beanFactory);
|
||||
publisherAnnotationAdvisor.setDefaultChannelName(this.defaultChannelName);
|
||||
this.advisor = publisherAnnotationAdvisor;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setBeanClassLoader(ClassLoader classLoader) {
|
||||
this.beanClassLoader = classLoader;
|
||||
}
|
||||
|
||||
public void setOrder(int order) {
|
||||
this.order = order;
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getOrder() {
|
||||
return this.order;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterPropertiesSet() {
|
||||
this.advisor = new PublisherAnnotationAdvisor();
|
||||
this.advisor.setBeanFactory(this.beanFactory);
|
||||
this.advisor.setDefaultChannelName(this.defaultChannelName);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
|
||||
return bean;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
|
||||
Class<?> targetClass = AopUtils.getTargetClass(bean);
|
||||
|
||||
// the set will hold records of prior class scans and will contain the bean classes that can not
|
||||
// be assigned to the Advisor interface and therefore can be short circuited
|
||||
if (this.nonApplicableCache.contains(targetClass)) {
|
||||
return bean;
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
try {
|
||||
this.beanFactory.getBean(PublisherAnnotationBeanPostProcessor.class);
|
||||
}
|
||||
|
||||
if (AopUtils.canApply(this.advisor, targetClass)) {
|
||||
if (bean instanceof Advised) {
|
||||
((Advised) bean).addAdvisor(this.advisor);
|
||||
return bean;
|
||||
}
|
||||
else {
|
||||
ProxyFactory proxyFactory = new ProxyFactory(bean);
|
||||
// Copy our properties (proxyTargetClass etc) inherited from ProxyConfig.
|
||||
proxyFactory.copyFrom(this);
|
||||
proxyFactory.addAdvisor(this.advisor);
|
||||
return proxyFactory.getProxy(this.beanClassLoader);
|
||||
}
|
||||
}
|
||||
else {
|
||||
// cannot apply advisor
|
||||
this.nonApplicableCache.add(targetClass);
|
||||
return bean;
|
||||
catch (NoUniqueBeanDefinitionException ex) {
|
||||
throw new BeanCreationNotAllowedException(this.name,
|
||||
"Only one 'PublisherAnnotationBeanPostProcessor' bean can be defined in the application context." +
|
||||
" Do not use '@EnablePublisher' (or '<int:enable-publisher>') if you declare a" +
|
||||
" 'PublisherAnnotationBeanPostProcessor' bean definition manually. " +
|
||||
"Bean names found: " + ex.getBeanNamesFound());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2014-2018 the original author or authors.
|
||||
* Copyright 2014-2019 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.
|
||||
@@ -23,11 +23,15 @@ import java.lang.annotation.RetentionPolicy;
|
||||
import java.lang.annotation.Target;
|
||||
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.core.Ordered;
|
||||
import org.springframework.core.annotation.AliasFor;
|
||||
|
||||
/**
|
||||
* Provides the registration for the {@link org.springframework.integration.aop.PublisherAnnotationBeanPostProcessor}
|
||||
* Provides the registration for the
|
||||
* {@link org.springframework.integration.aop.PublisherAnnotationBeanPostProcessor}
|
||||
* to allow the use of the {@link org.springframework.integration.annotation.Publisher} annotation.
|
||||
* In addition the {@code default-publisher-channel} name can be configured as the {@code value} of this annotation.
|
||||
* In addition the {@code default-publisher-channel} name can be configured as
|
||||
* the {@link #defaultChannel()} of this annotation.
|
||||
*
|
||||
* @author Artem Bilan
|
||||
*
|
||||
@@ -40,7 +44,39 @@ import org.springframework.context.annotation.Import;
|
||||
public @interface EnablePublisher {
|
||||
|
||||
/**
|
||||
* @return the {@code default-publisher-channel} name.
|
||||
* Alias for the {@link #defaultChannel()} attribute.
|
||||
* The {@code default-publisher-channel} name.
|
||||
* @return the channel bean name.
|
||||
*/
|
||||
@AliasFor("defaultChannel")
|
||||
String value() default "";
|
||||
|
||||
/**
|
||||
* The {@code default-publisher-channel} name.
|
||||
* @return the channel bean name.
|
||||
* @since 5.1.3
|
||||
*/
|
||||
@AliasFor("value")
|
||||
String defaultChannel() default "";
|
||||
|
||||
/**
|
||||
* Indicate whether subclass-based (CGLIB) proxies are to be created as opposed
|
||||
* to standard Java interface-based proxies.
|
||||
* @return whether proxy target class or not.
|
||||
* @since 5.1.3
|
||||
*/
|
||||
boolean proxyTargetClass() default false;
|
||||
|
||||
/**
|
||||
* Indicate the order in which the
|
||||
* {@link org.springframework.integration.aop.PublisherAnnotationBeanPostProcessor}
|
||||
* should be applied.
|
||||
* <p>The default is {@link Ordered#LOWEST_PRECEDENCE} in order to run
|
||||
* after all other post-processors, so that it can add an advisor to
|
||||
* existing proxies rather than double-proxy.
|
||||
* @return the order for the bean post-processor.
|
||||
* @since 5.1.3
|
||||
*/
|
||||
int order() default Ordered.LOWEST_PRECEDENCE;
|
||||
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2014 the original author or authors.
|
||||
* Copyright 2014-2019 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.
|
||||
@@ -37,6 +37,7 @@ import org.springframework.util.StringUtils;
|
||||
/**
|
||||
* @author Artem Bilan
|
||||
* @author Gary Russell
|
||||
*
|
||||
* @since 4.0
|
||||
*/
|
||||
public class PublisherRegistrar implements ImportBeanDefinitionRegistrar {
|
||||
@@ -48,21 +49,28 @@ public class PublisherRegistrar implements ImportBeanDefinitionRegistrar {
|
||||
Map<String, Object> annotationAttributes =
|
||||
importingClassMetadata.getAnnotationAttributes(EnablePublisher.class.getName());
|
||||
|
||||
String value = (annotationAttributes == null
|
||||
? (String) AnnotationUtils.getDefaultValue(EnablePublisher.class)
|
||||
: (String) annotationAttributes.get("value"));
|
||||
String defaultChannel =
|
||||
annotationAttributes == null
|
||||
? (String) AnnotationUtils.getDefaultValue(EnablePublisher.class)
|
||||
: (String) annotationAttributes.get("defaultChannel");
|
||||
if (!registry.containsBeanDefinition(IntegrationContextUtils.PUBLISHER_ANNOTATION_POSTPROCESSOR_NAME)) {
|
||||
BeanDefinitionBuilder builder =
|
||||
BeanDefinitionBuilder.genericBeanDefinition(PublisherAnnotationBeanPostProcessor.class)
|
||||
.setRole(BeanDefinition.ROLE_INFRASTRUCTURE);
|
||||
|
||||
if (StringUtils.hasText(value)) {
|
||||
builder.addPropertyValue("defaultChannelName", value);
|
||||
if (StringUtils.hasText(defaultChannel)) {
|
||||
builder.addPropertyValue("defaultChannelName", defaultChannel);
|
||||
if (logger.isInfoEnabled()) {
|
||||
logger.info("Setting '@Publisher' default-output-channel to '" + value + "'.");
|
||||
logger.info("Setting '@Publisher' default-output-channel to '" + defaultChannel + "'.");
|
||||
}
|
||||
}
|
||||
|
||||
if (annotationAttributes != null) {
|
||||
Object proxyTargetClass = annotationAttributes.get("proxyTargetClass");
|
||||
builder.addPropertyValue("proxyTargetClass", proxyTargetClass);
|
||||
Object order = annotationAttributes.get("order");
|
||||
builder.addPropertyValue("order", order);
|
||||
}
|
||||
registry.registerBeanDefinition(IntegrationContextUtils.PUBLISHER_ANNOTATION_POSTPROCESSOR_NAME,
|
||||
builder.getBeanDefinition());
|
||||
}
|
||||
@@ -71,17 +79,17 @@ public class PublisherRegistrar implements ImportBeanDefinitionRegistrar {
|
||||
registry.getBeanDefinition(IntegrationContextUtils.PUBLISHER_ANNOTATION_POSTPROCESSOR_NAME);
|
||||
MutablePropertyValues propertyValues = beanDefinition.getPropertyValues();
|
||||
PropertyValue defaultChannelPropertyValue = propertyValues.getPropertyValue("defaultChannelName");
|
||||
if (StringUtils.hasText(value)) {
|
||||
if (StringUtils.hasText(defaultChannel)) {
|
||||
if (defaultChannelPropertyValue == null) {
|
||||
propertyValues.addPropertyValue("defaultChannelName", value);
|
||||
propertyValues.addPropertyValue("defaultChannelName", defaultChannel);
|
||||
if (logger.isInfoEnabled()) {
|
||||
logger.info("Setting '@Publisher' default-output-channel to '" + value + "'.");
|
||||
logger.info("Setting '@Publisher' default-output-channel to '" + defaultChannel + "'.");
|
||||
}
|
||||
}
|
||||
else if (!value.equals(defaultChannelPropertyValue.getValue())) {
|
||||
else if (!defaultChannel.equals(defaultChannelPropertyValue.getValue())) {
|
||||
throw new BeanDefinitionStoreException("When more than one enable publisher definition " +
|
||||
"(@EnablePublisher or <annotation-config>)" +
|
||||
" is found in the context, they all must have the same 'default-publisher-channel' value.");
|
||||
"(@EnablePublisher or <annotation-config>) is found in the context, " +
|
||||
"they all must have the same 'default-publisher-channel' attribute value.");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2002-2018 the original author or authors.
|
||||
* Copyright 2002-2019 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.
|
||||
@@ -16,7 +16,7 @@
|
||||
|
||||
package org.springframework.integration.config.xml;
|
||||
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
import org.w3c.dom.Element;
|
||||
@@ -40,8 +40,6 @@ public class AnnotationConfigParser implements BeanDefinitionParser {
|
||||
|
||||
@Override
|
||||
public BeanDefinition parse(final Element element, ParserContext parserContext) {
|
||||
IntegrationRegistrar integrationRegistrar = new IntegrationRegistrar();
|
||||
|
||||
StandardAnnotationMetadata importingClassMetadata =
|
||||
new StandardAnnotationMetadata(Object.class) {
|
||||
|
||||
@@ -51,8 +49,13 @@ public class AnnotationConfigParser implements BeanDefinitionParser {
|
||||
Element enablePublisherElement =
|
||||
DomUtils.getChildElementByTagName(element, "enable-publisher");
|
||||
if (enablePublisherElement != null) {
|
||||
return Collections.singletonMap("value",
|
||||
Map<String, Object> attributes = new HashMap<>();
|
||||
attributes.put("defaultChannel",
|
||||
enablePublisherElement.getAttribute("default-publisher-channel"));
|
||||
attributes.put("proxyTargetClass",
|
||||
enablePublisherElement.getAttribute("proxy-target-class"));
|
||||
attributes.put("order", enablePublisherElement.getAttribute("order"));
|
||||
return attributes;
|
||||
}
|
||||
else {
|
||||
return null;
|
||||
@@ -65,7 +68,7 @@ public class AnnotationConfigParser implements BeanDefinitionParser {
|
||||
|
||||
};
|
||||
|
||||
integrationRegistrar.registerBeanDefinitions(importingClassMetadata, parserContext.getRegistry());
|
||||
new IntegrationRegistrar().registerBeanDefinitions(importingClassMetadata, parserContext.getRegistry());
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -35,6 +35,30 @@
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="proxy-target-class" default="false">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Indicate whether subclass-based (CGLIB) proxies are to be created as opposed
|
||||
to standard Java interface-based proxies.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:simpleType>
|
||||
<xsd:union memberTypes="xsd:boolean xsd:string" />
|
||||
</xsd:simpleType>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="order" default="2147483647">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Indicate the order in which the
|
||||
'org.springframework.integration.aop.PublisherAnnotationBeanPostProcessor'
|
||||
should be applied.
|
||||
Defaults to 'org.springframework.core.Ordered.LOWEST_PRECEDENCE'.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:simpleType>
|
||||
<xsd:union memberTypes="xsd:integer xsd:string" />
|
||||
</xsd:simpleType>
|
||||
</xsd:attribute>
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
</xsd:sequence>
|
||||
|
||||
@@ -15,7 +15,7 @@
|
||||
<message-history tracked-components="publishedChannel,input,annotationTestService*"/>
|
||||
|
||||
<annotation-config>
|
||||
<enable-publisher default-publisher-channel="publishedChannel"/>
|
||||
<enable-publisher default-publisher-channel="publishedChannel" proxy-target-class="true" order="2147483646"/>
|
||||
</annotation-config>
|
||||
|
||||
<channel-interceptor pattern="none">
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2014-2018 the original author or authors.
|
||||
* Copyright 2014-2019 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.
|
||||
@@ -19,6 +19,7 @@ package org.springframework.integration.configuration;
|
||||
import static org.hamcrest.CoreMatchers.equalTo;
|
||||
import static org.hamcrest.CoreMatchers.instanceOf;
|
||||
import static org.hamcrest.CoreMatchers.not;
|
||||
import static org.hamcrest.Matchers.contains;
|
||||
import static org.hamcrest.Matchers.containsInAnyOrder;
|
||||
import static org.hamcrest.Matchers.containsString;
|
||||
import static org.hamcrest.Matchers.sameInstance;
|
||||
@@ -51,8 +52,6 @@ import java.util.concurrent.atomic.AtomicReference;
|
||||
import org.aopalliance.aop.Advice;
|
||||
import org.aopalliance.intercept.MethodInterceptor;
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.hamcrest.BaseMatcher;
|
||||
import org.hamcrest.Description;
|
||||
import org.hamcrest.Matchers;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
@@ -92,6 +91,7 @@ import org.springframework.integration.annotation.Role;
|
||||
import org.springframework.integration.annotation.ServiceActivator;
|
||||
import org.springframework.integration.annotation.Transformer;
|
||||
import org.springframework.integration.annotation.UseSpelInvoker;
|
||||
import org.springframework.integration.aop.PublisherAnnotationBeanPostProcessor;
|
||||
import org.springframework.integration.channel.AbstractMessageChannel;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.channel.NullChannel;
|
||||
@@ -127,7 +127,6 @@ import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageDeliveryException;
|
||||
import org.springframework.messaging.MessageHandler;
|
||||
import org.springframework.messaging.MessagingException;
|
||||
import org.springframework.messaging.PollableChannel;
|
||||
import org.springframework.messaging.handler.annotation.Payload;
|
||||
import org.springframework.messaging.support.ChannelInterceptor;
|
||||
@@ -295,6 +294,9 @@ public class EnableIntegrationTests {
|
||||
@Qualifier("enableIntegrationTests.ChildConfiguration.autoCreatedChannelMessageSource.inboundChannelAdapter")
|
||||
private Lifecycle autoCreatedChannelMessageSourceAdapter;
|
||||
|
||||
@Autowired
|
||||
private PublisherAnnotationBeanPostProcessor publisherAnnotationBeanPostProcessor;
|
||||
|
||||
@Test
|
||||
public void testAnnotatedServiceActivator() throws Exception {
|
||||
this.serviceActivatorEndpoint.start();
|
||||
@@ -666,7 +668,7 @@ public class EnableIntegrationTests {
|
||||
List<Integer> integers = ref.get();
|
||||
assertEquals(5, integers.size());
|
||||
|
||||
assertThat(integers, Matchers.<Integer>contains(2, 4, 6, 8, 10));
|
||||
assertThat(integers, contains(2, 4, 6, 8, 10));
|
||||
}
|
||||
|
||||
@Test
|
||||
@@ -744,13 +746,17 @@ public class EnableIntegrationTests {
|
||||
|
||||
assertNotNull(receive);
|
||||
assertEquals(testDate, receive.getPayload());
|
||||
|
||||
assertTrue(this.publisherAnnotationBeanPostProcessor.isProxyTargetClass());
|
||||
assertEquals(Integer.MAX_VALUE - 1, this.publisherAnnotationBeanPostProcessor.getOrder());
|
||||
}
|
||||
|
||||
@Configuration
|
||||
@ComponentScan
|
||||
@IntegrationComponentScan
|
||||
@EnableIntegration
|
||||
// INT-3853 @PropertySource("classpath:org/springframework/integration/configuration/EnableIntegrationTests.properties")
|
||||
// INT-3853
|
||||
// @PropertySource("classpath:org/springframework/integration/configuration/EnableIntegrationTests.properties")
|
||||
@EnableMessageHistory({ "input", "publishedChannel", "annotationTestService*" })
|
||||
public static class ContextConfiguration {
|
||||
|
||||
@@ -996,7 +1002,7 @@ public class EnableIntegrationTests {
|
||||
@EnableIntegration
|
||||
@ImportResource("classpath:org/springframework/integration/configuration/EnableIntegrationTests-context.xml")
|
||||
@EnableMessageHistory("${message.history.tracked.components}")
|
||||
@EnablePublisher("publishedChannel")
|
||||
@EnablePublisher(defaultChannel = "publishedChannel")
|
||||
@EnableAsync
|
||||
public static class ContextConfiguration2 {
|
||||
|
||||
@@ -1026,14 +1032,9 @@ public class EnableIntegrationTests {
|
||||
@ServiceActivator(inputChannel = "sendAsyncChannel")
|
||||
@Role("foo")
|
||||
public MessageHandler sendAsyncHandler() {
|
||||
return new MessageHandler() {
|
||||
|
||||
@Override
|
||||
public void handleMessage(Message<?> message) throws MessagingException {
|
||||
asyncAnnotationProcessThread().set(Thread.currentThread());
|
||||
asyncAnnotationProcessLatch().countDown();
|
||||
}
|
||||
|
||||
return message -> {
|
||||
asyncAnnotationProcessThread().set(Thread.currentThread());
|
||||
asyncAnnotationProcessLatch().countDown();
|
||||
};
|
||||
}
|
||||
|
||||
@@ -1139,7 +1140,7 @@ public class EnableIntegrationTests {
|
||||
|
||||
@Bean
|
||||
public AnnotationTestService annotationTestService() {
|
||||
return new AnnotationTestServiceImpl();
|
||||
return new AnnotationTestService();
|
||||
}
|
||||
|
||||
@Bean
|
||||
@@ -1155,7 +1156,8 @@ public class EnableIntegrationTests {
|
||||
@Bean
|
||||
@ServiceActivator(inputChannel = "myHandlerChannel", adviceChain = "myHandlerAdvice")
|
||||
public MessageHandler myHandler() {
|
||||
return message -> { };
|
||||
return message -> {
|
||||
};
|
||||
}
|
||||
|
||||
@Bean
|
||||
@@ -1198,53 +1200,12 @@ public class EnableIntegrationTests {
|
||||
|
||||
}
|
||||
|
||||
public interface AnnotationTestService {
|
||||
|
||||
String handle(String payload);
|
||||
|
||||
String handle1(String payload);
|
||||
|
||||
String handle2(String payload);
|
||||
|
||||
String handle3(String payload);
|
||||
|
||||
String handle4(String payload);
|
||||
|
||||
String transform(Message<String> message);
|
||||
|
||||
String transform2(Message<String> message);
|
||||
|
||||
Integer count();
|
||||
|
||||
String foo();
|
||||
|
||||
Message<?> message();
|
||||
|
||||
Integer annCount();
|
||||
|
||||
Integer annCount1();
|
||||
|
||||
Integer annCount2();
|
||||
|
||||
Integer annCount5();
|
||||
|
||||
Integer annCount8();
|
||||
|
||||
Integer annAgg1(List<?> messages);
|
||||
|
||||
Integer annAgg2(List<?> messages);
|
||||
|
||||
Integer multiply(Integer value);
|
||||
|
||||
}
|
||||
|
||||
public static class AnnotationTestServiceImpl implements Lifecycle, AnnotationTestService {
|
||||
public static class AnnotationTestService implements Lifecycle {
|
||||
|
||||
private final AtomicInteger counter = new AtomicInteger();
|
||||
|
||||
private boolean running;
|
||||
|
||||
@Override
|
||||
@ServiceActivator(inputChannel = "input", outputChannel = "output", autoStartup = "false",
|
||||
poller = @Poller(maxMessagesPerPoll = "${poller.maxMessagesPerPoll}",
|
||||
fixedDelay = "${poller.interval}",
|
||||
@@ -1256,7 +1217,6 @@ public class EnableIntegrationTests {
|
||||
return payload.toUpperCase();
|
||||
}
|
||||
|
||||
@Override
|
||||
@ServiceActivator(inputChannel = "input1", outputChannel = "output",
|
||||
poller = @Poller(maxMessagesPerPoll = "${poller.maxMessagesPerPoll}",
|
||||
fixedRate = "${poller.interval}"))
|
||||
@@ -1266,7 +1226,6 @@ public class EnableIntegrationTests {
|
||||
return payload.toUpperCase();
|
||||
}
|
||||
|
||||
@Override
|
||||
@ServiceActivator(inputChannel = "input2", outputChannel = "output",
|
||||
poller = @Poller(maxMessagesPerPoll = "${poller.maxMessagesPerPoll}", cron = "0 5 7 * * *"))
|
||||
@Publisher
|
||||
@@ -1275,7 +1234,6 @@ public class EnableIntegrationTests {
|
||||
return payload.toUpperCase();
|
||||
}
|
||||
|
||||
@Override
|
||||
@ServiceActivator(inputChannel = "input3", outputChannel = "output", poller = @Poller("myPoller"))
|
||||
@Publisher
|
||||
@Payload("#args[0].toLowerCase()")
|
||||
@@ -1283,7 +1241,6 @@ public class EnableIntegrationTests {
|
||||
return payload.toUpperCase();
|
||||
}
|
||||
|
||||
@Override
|
||||
@ServiceActivator(inputChannel = "input4", outputChannel = "output",
|
||||
poller = @Poller(trigger = "myTrigger"))
|
||||
@Publisher
|
||||
@@ -1303,7 +1260,6 @@ public class EnableIntegrationTests {
|
||||
return payload.toUpperCase();
|
||||
}*/
|
||||
|
||||
@Override
|
||||
@Transformer(inputChannel = "gatewayChannel")
|
||||
public String transform(Message<String> message) {
|
||||
assertTrue(message.getHeaders().containsKey("foo"));
|
||||
@@ -1313,7 +1269,6 @@ public class EnableIntegrationTests {
|
||||
return this.handle(message.getPayload()) + Arrays.asList(new Throwable().getStackTrace()).toString();
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transformer(inputChannel = "gatewayChannel2")
|
||||
@UseSpelInvoker(compilerMode = "${xxxxxxxx:IMMEDIATE}")
|
||||
public String transform2(Message<String> message) {
|
||||
@@ -1324,20 +1279,17 @@ public class EnableIntegrationTests {
|
||||
return this.handle(message.getPayload()) + "2" + Arrays.asList(new Throwable().getStackTrace()).toString();
|
||||
}
|
||||
|
||||
@Override
|
||||
@MyInboundChannelAdapter1
|
||||
public Integer count() {
|
||||
return this.counter.incrementAndGet();
|
||||
}
|
||||
|
||||
@Override
|
||||
@InboundChannelAdapter(value = "fooChannel",
|
||||
poller = @Poller(trigger = "onlyOnceTrigger", maxMessagesPerPoll = "2"))
|
||||
public String foo() {
|
||||
return "foo";
|
||||
}
|
||||
|
||||
@Override
|
||||
@InboundChannelAdapter(value = "messageChannel", poller = @Poller(fixedDelay = "${poller.interval}",
|
||||
maxMessagesPerPoll = "1"))
|
||||
public Message<?> message() {
|
||||
@@ -1361,44 +1313,37 @@ public class EnableIntegrationTests {
|
||||
|
||||
// metaAnnotation tests
|
||||
|
||||
@Override
|
||||
@MyServiceActivator
|
||||
public Integer annCount() {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
@MyServiceActivator1(inputChannel = "annInput1", autoStartup = "true",
|
||||
adviceChain = { "annAdvice1" }, poller = @Poller(fixedRate = "2000"))
|
||||
public Integer annCount1() {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
@MyServiceActivatorNoLocalAtts
|
||||
public Integer annCount2() {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
@MyServiceActivator5
|
||||
public Integer annCount5() {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
@MyServiceActivator8
|
||||
public Integer annCount8() {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
@MyAggregator
|
||||
public Integer annAgg1(List<?> messages) {
|
||||
return 42;
|
||||
}
|
||||
|
||||
@Override
|
||||
@MyAggregatorDefaultOverrideDefaults
|
||||
public Integer annAgg2(List<?> messages) {
|
||||
return 42;
|
||||
@@ -1408,7 +1353,6 @@ public class EnableIntegrationTests {
|
||||
/*@BridgeFrom("")
|
||||
public void invalidBridgeAnnotationMethod(Object payload) {}*/
|
||||
|
||||
@Override
|
||||
@ServiceActivator(inputChannel = "monoChannel")
|
||||
public Integer multiply(Integer value) {
|
||||
return value * 2;
|
||||
@@ -1693,25 +1637,4 @@ public class EnableIntegrationTests {
|
||||
|
||||
}
|
||||
|
||||
public class RegexMatcher<T> extends BaseMatcher<T> {
|
||||
|
||||
private final String regex;
|
||||
|
||||
public RegexMatcher(String regex) {
|
||||
this.regex = regex;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean matches(Object o) {
|
||||
return ((String) o).matches(regex);
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void describeTo(Description description) {
|
||||
description.appendText("matches regex=");
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -142,6 +142,8 @@ public class IntegrationConfiguration {
|
||||
----
|
||||
====
|
||||
|
||||
Starting with version 5.1.3, the `<int:enable-publisher>` component, as well as the `@EnablePublisher` annotation have the `proxy-target-class` and `order` attributes for tuning the `ProxyFactory` configuration.
|
||||
|
||||
Similar to other Spring annotations (`@Component`, `@Scheduled`, and so on), you can also use `@Publisher` as a meta-annotation.
|
||||
This means that you can define your own annotations that are treated in the same way as the `@Publisher` itself.
|
||||
The following example shows how to do so:
|
||||
@@ -321,7 +323,7 @@ Another way of handling this type of scenario is with a wire-tap. See <<channel-
|
||||
In the preceding sections, we looked at the message-publishing feature, which constructs and publishes messages as by-products of method invocations.
|
||||
However, in those cases, you are still responsible for invoking the method.
|
||||
Spring Integration 2.0 added support for scheduled message producers and publishers with the new `expression` attribute on the 'inbound-channel-adapter' element.
|
||||
You can schedul based on several triggers, any one of which can be configured on the 'poller' element.
|
||||
You can schedule based on several triggers, any one of which can be configured on the 'poller' element.
|
||||
Currently, we support `cron`, `fixed-rate`, `fixed-delay` and any custom trigger implemented by you and referenced by the 'trigger' attribute value.
|
||||
|
||||
As mentioned earlier, support for scheduled producers and publishers is provided via the `<inbound-channel-adapter>` XML element.
|
||||
|
||||
@@ -119,6 +119,7 @@ See <<aggregator>> for more information.
|
||||
==== @Publisher annotation changes
|
||||
|
||||
Starting with version 5.1, you must explicitly turn on the `@Publisher` AOP functionality by using `@EnablePublisher` or by using the `<int:enable-publisher>` child element on `<int:annotation-config>`.
|
||||
Also the `proxy-target-class` and `order` attributes have been added for tuning the `ProxyFactory` configuration.
|
||||
|
||||
See <<publisher-annotation>> for more information.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user