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

INT-3448: Add @Async test for @Gateway

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

INT-3448: Addressing PR comments

Late Resolution of @Publisher.defaultChannel

Early resolution of the publisher default channel caused
`@ Configuration` factory beans to be instantiated before
Spring Integration bean post processors had been registered.

Use the channel name instead of channel reference and resolve
it when processing the first publish.

Merge branch 'INT-3448' of https://github.com/garyrussell/spring-integration into INT-3448

Conflicts:
	spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java

INT-3448: Polishing
This commit is contained in:
Artem Bilan
2014-07-03 23:21:17 +03:00
committed by Gary Russell
parent 23c80906d8
commit 5eb68866d9
7 changed files with 142 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -115,7 +115,7 @@ subprojects { subproject ->
springSecurityVersion = '3.2.4.RELEASE'
springSocialTwitterVersion = '1.1.0.RELEASE'
springRetryVersion = '1.1.0.RELEASE'
springVersion = project.hasProperty('springVersion') ? project.springVersion : '4.0.5.RELEASE'
springVersion = project.hasProperty('springVersion') ? project.springVersion : '4.0.6.RELEASE'
springWsVersion = '2.2.0.RELEASE'
xmlUnitVersion = '1.5'
xstreamVersion = '1.4.7'

29
spring-integration-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java
View File

@@ -56,6 +56,7 @@ import org.springframework.util.StringUtils;
*
* @author Mark Fisher
* @author Artem Bilan
* @author Gary Russell
* @since 2.0
*/
public class MessagePublishingInterceptor implements MethodInterceptor, BeanFactoryAware {
@@ -74,6 +75,8 @@ public class MessagePublishingInterceptor implements MethodInterceptor, BeanFact
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile String defaultChannelName;
public MessagePublishingInterceptor(PublisherMetadataSource metadataSource) {
Assert.notNull(metadataSource, "metadataSource must not be null");
this.metadataSource = metadataSource;
@@ -85,8 +88,22 @@ public class MessagePublishingInterceptor implements MethodInterceptor, BeanFact
this.metadataSource = metadataSource;
}
/**
* @deprecated Use {@link #setDefaultChannelName(String)}.
* @param defaultChannel the default channel.
*/
@Deprecated
public void setDefaultChannel(MessageChannel defaultChannel) {
this.messagingTemplate.setDefaultDestination(defaultChannel);
this.defaultChannelName = null;
}
/**
* @param defaultChannelName the default channel name.
* @since 4.0.3
*/
public void setDefaultChannelName(String defaultChannelName) {
this.defaultChannelName = defaultChannelName;
}
public void setChannelResolver(DestinationResolver<MessageChannel> channelResolver) {
@@ -100,6 +117,7 @@ public class MessagePublishingInterceptor implements MethodInterceptor, BeanFact
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);
}
@Override
public final Object invoke(final MethodInvocation invocation) throws Throwable {
Assert.notNull(this.metadataSource, "PublisherMetadataSource is required.");
final StandardEvaluationContext context = ExpressionUtils.createStandardEvaluationContext(this.beanFactory);
@@ -163,6 +181,17 @@ public class MessagePublishingInterceptor implements MethodInterceptor, BeanFact
this.messagingTemplate.send(channel, message);
}
else {
if (this.defaultChannelName != null) {
synchronized(this) {
if (this.defaultChannelName != null && this.messagingTemplate.getDefaultDestination() == null) {
Assert.state(this.channelResolver != null,
"ChannelResolver is required to resolve channel names.");
this.messagingTemplate.setDefaultChannel(
this.channelResolver.resolveDestination(this.defaultChannelName));
}
this.defaultChannelName = null;
}
}
this.messagingTemplate.send(message);
}
}

21
spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -46,6 +46,7 @@ import org.springframework.util.Assert;
* the default will be {@link Publisher @Publisher}.
*
* @author Mark Fisher
* @author Gary Russell
* @since 2.0
*/
@SuppressWarnings("serial")
@@ -68,19 +69,35 @@ public class PublisherAnnotationAdvisor extends AbstractPointcutAdvisor implemen
}
/**
* @deprecated Use {@link #setDefaultChannelName(String)}.
* @param defaultChannel the default channel.
*/
@Deprecated
public void setDefaultChannel(MessageChannel defaultChannel) {
this.interceptor.setDefaultChannel(defaultChannel);
}
/**
* @param defaultChannelName the default channel name.
* @since 4.0.3
*/
public void setDefaultChannelName(String defaultChannelName) {
this.interceptor.setDefaultChannelName(defaultChannelName);
}
@Override
public void setBeanFactory(BeanFactory beanFactory) {
this.interceptor.setChannelResolver(new BeanFactoryChannelResolver(beanFactory));
this.interceptor.setBeanFactory(beanFactory);
}
@Override
public Advice getAdvice() {
return this.interceptor;
}
@Override
public Pointcut getPointcut() {
return this.buildPointcut();
}
@@ -149,10 +166,12 @@ public class PublisherAnnotationAdvisor extends AbstractPointcutAdvisor implemen
}
@Override
public ClassFilter getClassFilter() {
return this.classFilter;
}
@Override
public MethodMatcher getMethodMatcher() {
return this.methodMatcher;
}

27
spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationBeanPostProcessor.java
View File

@@ -36,6 +36,7 @@ import org.springframework.util.ClassUtils;
*
* @author Oleg Zhurakousky
* @author Mark Fisher
* @author Gary Russell
* @since 2.0
*/
@SuppressWarnings("serial")
@@ -44,6 +45,8 @@ public class PublisherAnnotationBeanPostProcessor extends ProxyConfig
private volatile MessageChannel defaultChannel;
private volatile String defaultChannelName;
private volatile PublisherAnnotationAdvisor advisor;
private volatile int order = Ordered.LOWEST_PRECEDENCE;
@@ -56,13 +59,23 @@ public class PublisherAnnotationBeanPostProcessor extends ProxyConfig
/**
* Set the default channel where Messages should be sent if the annotation
* itself does not provide a channel.
*
* @param defaultChannel The default channel.
* @deprecated Use {@link #setDefaultChannelName(String)}
*/
@Deprecated
public void setDefaultChannel(MessageChannel defaultChannel){
this.defaultChannel = defaultChannel;
}
/**
* Set the default channel where Messages should be sent if the annotation
* itself does not provide a channel.
* @since 4.0.3
*/
public void setDefaultChannelName(String defaultChannelName) {
this.defaultChannelName = defaultChannelName;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
@@ -82,11 +95,17 @@ public class PublisherAnnotationBeanPostProcessor extends ProxyConfig
return this.order;
}
@SuppressWarnings("deprecation")
@Override
public void afterPropertiesSet(){
advisor = new PublisherAnnotationAdvisor();
advisor.setBeanFactory(beanFactory);
advisor.setDefaultChannel(defaultChannel);
this.advisor = new PublisherAnnotationAdvisor();
this.advisor.setBeanFactory(this.beanFactory);
if (this.defaultChannel != null) {
this.advisor.setDefaultChannel(this.defaultChannel);
}
else {
this.advisor.setDefaultChannelName(this.defaultChannelName);
}
}
@Override

24
spring-integration-core/src/main/java/org/springframework/integration/config/PublisherRegistrar.java
View File

@@ -25,7 +25,6 @@ import org.springframework.beans.MutablePropertyValues;
import org.springframework.beans.PropertyValue;
import org.springframework.beans.factory.BeanDefinitionStoreException;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.context.annotation.ImportBeanDefinitionRegistrar;
@@ -36,6 +35,7 @@ import org.springframework.util.StringUtils;
/**
* @author Artem Bilan
* @author Gary Russell
* @since 4.0
*/
public class PublisherRegistrar implements ImportBeanDefinitionRegistrar {
@@ -44,36 +44,40 @@ public class PublisherRegistrar implements ImportBeanDefinitionRegistrar {
@Override
public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {
Map<String, Object> annotationAttributes = importingClassMetadata.getAnnotationAttributes(EnablePublisher.class.getName());
Map<String, Object> annotationAttributes =
importingClassMetadata.getAnnotationAttributes(EnablePublisher.class.getName());
if (annotationAttributes == null) {
return;
}
String value = (String) annotationAttributes.get("value");
if (!registry.containsBeanDefinition(IntegrationContextUtils.PUBLISHER_ANNOTATION_POSTPROCESSOR_NAME)) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(PublisherAnnotationBeanPostProcessor.class)
.setRole(BeanDefinition.ROLE_INFRASTRUCTURE);
BeanDefinitionBuilder builder =
BeanDefinitionBuilder.genericBeanDefinition(PublisherAnnotationBeanPostProcessor.class)
.setRole(BeanDefinition.ROLE_INFRASTRUCTURE);
if (StringUtils.hasText(value)) {
builder.addPropertyReference("defaultChannel", value);
builder.addPropertyValue("defaultChannelName", value);
if (logger.isInfoEnabled()) {
logger.info("Setting '@Publisher' default-output-channel to '" + value + "'.");
}
}
registry.registerBeanDefinition(IntegrationContextUtils.PUBLISHER_ANNOTATION_POSTPROCESSOR_NAME, builder.getBeanDefinition());
registry.registerBeanDefinition(IntegrationContextUtils.PUBLISHER_ANNOTATION_POSTPROCESSOR_NAME,
builder.getBeanDefinition());
}
else {
BeanDefinition beanDefinition = registry.getBeanDefinition(IntegrationContextUtils.PUBLISHER_ANNOTATION_POSTPROCESSOR_NAME);
BeanDefinition beanDefinition =
registry.getBeanDefinition(IntegrationContextUtils.PUBLISHER_ANNOTATION_POSTPROCESSOR_NAME);
MutablePropertyValues propertyValues = beanDefinition.getPropertyValues();
PropertyValue defaultChannelPropertyValue = propertyValues.getPropertyValue("defaultChannel");
PropertyValue defaultChannelPropertyValue = propertyValues.getPropertyValue("defaultChannelName");
if (StringUtils.hasText(value)) {
if (defaultChannelPropertyValue == null) {
propertyValues.addPropertyValue("defaultChannel", new RuntimeBeanReference(value));
propertyValues.addPropertyValue("defaultChannelName", value);
if (logger.isInfoEnabled()) {
logger.info("Setting '@Publisher' default-output-channel to '" + value + "'.");
}
}
else if (!value.equals(((RuntimeBeanReference) defaultChannelPropertyValue.getValue()).getBeanName())) {
else if (!value.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.");

5
spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherExpressionTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -60,7 +60,7 @@ public class PublisherExpressionTests {
PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor();
advisor.setBeanFactory(context);
QueueChannel testChannel = context.getBean("testChannel", QueueChannel.class);
advisor.setDefaultChannel(testChannel);
advisor.setDefaultChannelName("testChannel");
ProxyFactory pf = new ProxyFactory(new TestBeanImpl());
pf.addAdvisor(advisor);
TestBean proxy = (TestBean) pf.getProxy();
@@ -79,6 +79,7 @@ public class PublisherExpressionTests {
static class TestBeanImpl implements TestBean {
@Override
@Publisher
@Payload("#return + @foo")
public String test(@Header("foo") String foo) {

62
spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java
View File

@@ -17,14 +17,7 @@
package org.springframework.integration.configuration;
import static org.hamcrest.Matchers.containsString;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.junit.Assert.*;
import static org.mockito.Mockito.mock;
import java.lang.annotation.ElementType;
@@ -33,8 +26,11 @@ import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.aopalliance.intercept.MethodInterceptor;
import org.hamcrest.Matchers;
@@ -94,12 +90,16 @@ import org.springframework.integration.test.util.TestUtils;
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.support.ChannelInterceptor;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.TriggerContext;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.stereotype.Component;
@@ -112,7 +112,8 @@ import org.springframework.test.context.support.AnnotationConfigContextLoader;
* @author Artem Bilan
* @since 4.0
*/
@ContextConfiguration(loader = AnnotationConfigContextLoader.class, classes = {EnableIntegrationTests.ContextConfiguration.class, EnableIntegrationTests.ContextConfiguration2.class})
@ContextConfiguration(loader = AnnotationConfigContextLoader.class,
classes = {EnableIntegrationTests.ContextConfiguration.class, EnableIntegrationTests.ContextConfiguration2.class})
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class EnableIntegrationTests {
@@ -165,6 +166,12 @@ public class EnableIntegrationTests {
@Autowired
private TestGateway testGateway;
@Autowired
private CountDownLatch asyncAnnotationProcessLatch;
@Autowired
private AtomicReference<Thread> asyncAnnotationProcessThread;
@Autowired
private TestGateway2 testGateway2;
@@ -338,10 +345,13 @@ public class EnableIntegrationTests {
}
@Test
public void testMessagingGateway() {
public void testMessagingGateway() throws InterruptedException {
String payload = "bar";
assertEquals(payload.toUpperCase(), this.testGateway.echo(payload));
assertEquals(payload.toUpperCase() + "2", this.testGateway2.echo2(payload));
this.testGateway.sendAsync("foo");
assertTrue(this.asyncAnnotationProcessLatch.await(1, TimeUnit.SECONDS));
assertNotSame(Thread.currentThread(), this.asyncAnnotationProcessThread.get());
}
@Test
@@ -756,6 +766,7 @@ public class EnableIntegrationTests {
@ImportResource("classpath:org/springframework/integration/configuration/EnableIntegrationTests-context.xml")
@EnableMessageHistory("${message.history.tracked.components}")
@EnablePublisher("publishedChannel")
@EnableAsync
public static class ContextConfiguration2 {
@Bean
@@ -763,6 +774,33 @@ public class EnableIntegrationTests {
return new PropertySourcesPlaceholderConfigurer();
}
@Bean
public MessageChannel sendAsyncChannel() {
return new DirectChannel();
}
@Bean
public CountDownLatch asyncAnnotationProcessLatch() {
return new CountDownLatch(1);
}
@Bean
public AtomicReference<Thread> asyncAnnotationProcessThread() {
return new AtomicReference<Thread>();
}
@Bean
@ServiceActivator(inputChannel = "sendAsyncChannel")
public MessageHandler sendAsyncHandler() {
return new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
asyncAnnotationProcessLatch().countDown();
asyncAnnotationProcessThread().set(Thread.currentThread());
}
};
}
@Bean
public PollableChannel publishedChannel() {
return new QueueChannel();
@@ -1000,6 +1038,10 @@ public class EnableIntegrationTests {
@Gateway(headers = @GatewayHeader(name = "calledMethod", expression = "#gatewayMethod.name"))
String echo(String payload);
@Gateway(requestChannel = "sendAsyncChannel")
@Async
void sendAsync(String payload);
}
@TestMessagingGateway2