From 5eb68866d9732ac32282b970aae379f0b1126ed1 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 3 Jul 2014 23:21:17 +0300 Subject: [PATCH] INT-3448: Add `@Async` test for `@Gateway` 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 --- build.gradle | 2 +- .../aop/MessagePublishingInterceptor.java | 29 +++++++++ .../aop/PublisherAnnotationAdvisor.java | 21 ++++++- .../PublisherAnnotationBeanPostProcessor.java | 27 ++++++-- .../config/PublisherRegistrar.java | 24 ++++--- .../aop/PublisherExpressionTests.java | 5 +- .../configuration/EnableIntegrationTests.java | 62 ++++++++++++++++--- 7 files changed, 142 insertions(+), 28 deletions(-) diff --git a/build.gradle b/build.gradle index f228ecf3ef..ec05aef609 100644 --- a/build.gradle +++ b/build.gradle @@ -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' diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java index c0bf07576a..daebe27e71 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java @@ -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 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); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java index 0256bce85d..212670c46c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java @@ -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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationBeanPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationBeanPostProcessor.java index 262222bd9d..7399cdcc65 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationBeanPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationBeanPostProcessor.java @@ -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 diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/PublisherRegistrar.java b/spring-integration-core/src/main/java/org/springframework/integration/config/PublisherRegistrar.java index 53ff997d0c..0c9b2ea5d5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/PublisherRegistrar.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/PublisherRegistrar.java @@ -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 annotationAttributes = importingClassMetadata.getAnnotationAttributes(EnablePublisher.class.getName()); + Map 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 )" + " is found in the context, they all must have the same 'default-publisher-channel' value."); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherExpressionTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherExpressionTests.java index 7398c517c3..e1212e8df5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherExpressionTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherExpressionTests.java @@ -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) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java index 20bbe80280..c522e65c05 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java @@ -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 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 asyncAnnotationProcessThread() { + return new AtomicReference(); + } + + @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