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