diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierConsumer.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierConsumer.java index ccf7324dcf..e851db75e5 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierConsumer.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierConsumer.java @@ -29,7 +29,6 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.InitializingBean; -import org.springframework.integration.channel.BlockingChannel; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.endpoint.AbstractReplyProducingMessageConsumer; import org.springframework.integration.message.Message; @@ -67,8 +66,6 @@ import org.springframework.util.ObjectUtils; public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProducingMessageConsumer implements TaskSchedulerAware, InitializingBean { - public final static long DEFAULT_SEND_TIMEOUT = 1000; - public final static long DEFAULT_TIMEOUT = 60000; public final static long DEFAULT_REAPER_INTERVAL = 1000; @@ -79,8 +76,6 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc private volatile MessageChannel discardChannel; - protected volatile long sendTimeout = DEFAULT_SEND_TIMEOUT; - protected final ConcurrentMap barriers = new ConcurrentHashMap(); @@ -109,13 +104,6 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc this.discardChannel = discardChannel; } - /** - * Set the timeout for sending aggregation results and discarded Messages. - */ - public void setSendTimeout(long sendTimeout) { - this.sendTimeout = sendTimeout; - } - /** * Specify whether to aggregate and send the resulting Message when the * timeout elapses prior to the CompletionStrategy returning true. @@ -225,12 +213,7 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc } } if (replyChannel != null) { - if (replyChannel instanceof BlockingChannel && this.sendTimeout >= 0) { - ((BlockingChannel) replyChannel).send(result, this.sendTimeout); - } - else { - replyChannel.send(result); - } + this.sendReplyMessage(result, replyChannel); } else if (logger.isWarnEnabled()) { logger.warn("unable to determine reply target for aggregation result: " + result); @@ -240,9 +223,7 @@ public abstract class AbstractMessageBarrierConsumer extends AbstractReplyProduc private void sendToDiscardChannelIfAvailable(Message message) { if (this.discardChannel != null) { - boolean sent = (this.discardChannel instanceof BlockingChannel && this.sendTimeout >= 0) - ? ((BlockingChannel) this.discardChannel).send(message, this.sendTimeout) - : this.discardChannel.send(message); + boolean sent = this.sendReplyMessage(message, this.discardChannel); if (!sent) { if (logger.isWarnEnabled()) { logger.warn("unable to send to 'discardChannel', message: " + message); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java index e91259b143..8e5d698836 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java @@ -18,16 +18,18 @@ package org.springframework.integration.config.annotation; import java.lang.annotation.Annotation; import java.lang.reflect.Method; +import java.util.ArrayList; +import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.annotation.Poller; -import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.SubscribableChannel; +import org.springframework.integration.config.MessageBusParser; import org.springframework.integration.endpoint.AbstractMessageConsumer; import org.springframework.integration.endpoint.AbstractReplyProducingMessageConsumer; import org.springframework.integration.endpoint.MessageEndpoint; @@ -35,6 +37,12 @@ import org.springframework.integration.endpoint.PollingConsumerEndpoint; import org.springframework.integration.endpoint.SubscribingConsumerEndpoint; import org.springframework.integration.message.MessageConsumer; import org.springframework.integration.scheduling.IntervalTrigger; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.interceptor.NoRollbackRuleAttribute; +import org.springframework.transaction.interceptor.RollbackRuleAttribute; +import org.springframework.transaction.interceptor.RuleBasedTransactionAttribute; +import org.springframework.transaction.interceptor.TransactionAttribute; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -50,23 +58,23 @@ public abstract class AbstractMethodAnnotationPostProcessor rollBackRules = new ArrayList(); + Class[] rbf = annotation.rollbackFor(); + for (int i = 0; i < rbf.length; ++i) { + RollbackRuleAttribute rule = new RollbackRuleAttribute(rbf[i]); + rollBackRules.add(rule); + } + String[] rbfc = annotation.rollbackForClassName(); + for (int i = 0; i < rbfc.length; ++i) { + RollbackRuleAttribute rule = new RollbackRuleAttribute(rbfc[i]); + rollBackRules.add(rule); + } + Class[] nrbf = annotation.noRollbackFor(); + for (int i = 0; i < nrbf.length; ++i) { + NoRollbackRuleAttribute rule = new NoRollbackRuleAttribute(nrbf[i]); + rollBackRules.add(rule); + } + String[] nrbfc = annotation.noRollbackForClassName(); + for (int i = 0; i < nrbfc.length; ++i) { + NoRollbackRuleAttribute rule = new NoRollbackRuleAttribute(nrbfc[i]); + rollBackRules.add(rule); + } + rbta.getRollbackRules().addAll(rollBackRules); + return rbta; + } + + /** + * Subclasses must implement this method to create the MessageConsumer. + */ protected abstract MessageConsumer createConsumer(Object bean, Method method, T annotation); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java index 36140b3b73..ce0d0447d3 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java @@ -19,13 +19,13 @@ package org.springframework.integration.config.annotation; import java.lang.annotation.Annotation; import java.lang.reflect.Method; +import org.springframework.beans.factory.BeanFactory; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.aggregator.AbstractMessageAggregator; import org.springframework.integration.aggregator.CompletionStrategyAdapter; import org.springframework.integration.aggregator.MethodInvokingAggregator; import org.springframework.integration.annotation.Aggregator; import org.springframework.integration.annotation.CompletionStrategy; -import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.MessageConsumer; import org.springframework.util.Assert; @@ -39,8 +39,8 @@ import org.springframework.util.StringUtils; */ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { - public AggregatorAnnotationPostProcessor(MessageBus messageBus) { - super(messageBus); + public AggregatorAnnotationPostProcessor(BeanFactory beanFactory) { + super(beanFactory); } @@ -50,7 +50,7 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP this.configureCompletionStrategy(bean, aggregator); String discardChannelName = annotation.discardChannel(); if (StringUtils.hasText(discardChannelName)) { - MessageChannel discardChannel = this.getChannelRegistry().lookupChannel(discardChannelName); + MessageChannel discardChannel = this.channelRegistry.lookupChannel(discardChannelName); Assert.notNull(discardChannel, "unable to resolve discardChannel '" + discardChannelName + "'"); aggregator.setDiscardChannel(discardChannel); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationDrivenParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationDrivenParser.java index 40748e038f..d38d2a2559 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationDrivenParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationDrivenParser.java @@ -58,8 +58,6 @@ public class AnnotationDrivenParser implements BeanDefinitionParser { private void registerMessagingAnnotationPostProcessor(ParserContext parserContext) { BeanDefinition bd = new RootBeanDefinition(MessagingAnnotationPostProcessor.class); - bd.getConstructorArgumentValues().addGenericArgumentValue( - new RuntimeBeanReference(MessageBusParser.MESSAGE_BUS_BEAN_NAME)); BeanComponentDefinition bcd = new BeanComponentDefinition( bd, MESSAGING_ANNOTATION_POST_PROCESSOR_BEAN_NAME); parserContext.registerBeanComponent(bcd); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java index 53193689c6..370c06ba9f 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java @@ -22,11 +22,12 @@ import org.springframework.beans.factory.config.ConfigurableBeanFactory; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.annotation.ChannelAdapter; import org.springframework.integration.annotation.Poller; -import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.channel.SubscribableChannel; +import org.springframework.integration.config.MessageBusParser; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.endpoint.PollingConsumerEndpoint; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; @@ -45,16 +46,16 @@ import org.springframework.util.ClassUtils; */ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPostProcessor { - private final MessageBus messageBus; - private final ConfigurableBeanFactory beanFactory; + private final ChannelRegistry channelRegistry; - public ChannelAdapterAnnotationPostProcessor(MessageBus messageBus, ConfigurableBeanFactory beanFactory) { - Assert.notNull(messageBus, "MessageBus must not be null"); + + public ChannelAdapterAnnotationPostProcessor(ConfigurableBeanFactory beanFactory) { Assert.notNull(beanFactory, "BeanFactory must not be null"); - this.messageBus = messageBus; this.beanFactory = beanFactory; + this.channelRegistry = (ChannelRegistry) + this.beanFactory.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); } @@ -62,7 +63,7 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo Assert.notNull(this.beanFactory, "BeanFactory must not be null"); MessageEndpoint endpoint = null; String channelName = annotation.value(); - MessageChannel channel = this.messageBus.lookupChannel(channelName); + MessageChannel channel = this.channelRegistry.lookupChannel(channelName); if (channel == null) { DirectChannel directChannel = new DirectChannel(); directChannel.setBeanName(channelName); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java index 96059e07b8..bf4aa49ab5 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java @@ -41,6 +41,7 @@ import org.springframework.integration.annotation.Splitter; import org.springframework.integration.annotation.Transformer; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.ChannelRegistryAware; +import org.springframework.integration.config.MessageBusParser; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.stereotype.Component; import org.springframework.util.Assert; @@ -57,7 +58,7 @@ import org.springframework.util.StringUtils; */ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, BeanFactoryAware, InitializingBean { - private final MessageBus messageBus; + private volatile MessageBus messageBus; private volatile ConfigurableBeanFactory beanFactory; @@ -66,12 +67,6 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean new HashMap, MethodAnnotationPostProcessor>(); - public MessagingAnnotationPostProcessor(MessageBus messageBus) { - Assert.notNull(messageBus, "MessageBus must not be null."); - this.messageBus = messageBus; - } - - public void setBeanFactory(BeanFactory beanFactory) { Assert.isAssignable(ConfigurableBeanFactory.class, beanFactory.getClass(), "a ConfigurableBeanFactory is required"); @@ -80,12 +75,13 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean public void afterPropertiesSet() { Assert.notNull(this.beanFactory, "BeanFactory must not be null"); - postProcessors.put(Aggregator.class, new AggregatorAnnotationPostProcessor(this.messageBus)); - postProcessors.put(ChannelAdapter.class, new ChannelAdapterAnnotationPostProcessor(this.messageBus, this.beanFactory)); - postProcessors.put(Router.class, new RouterAnnotationPostProcessor(this.messageBus)); - postProcessors.put(ServiceActivator.class, new ServiceActivatorAnnotationPostProcessor(this.messageBus)); - postProcessors.put(Splitter.class, new SplitterAnnotationPostProcessor(this.messageBus)); - postProcessors.put(Transformer.class, new TransformerAnnotationPostProcessor(this.messageBus)); + this.messageBus = (MessageBus) this.beanFactory.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); + postProcessors.put(Aggregator.class, new AggregatorAnnotationPostProcessor(this.beanFactory)); + postProcessors.put(ChannelAdapter.class, new ChannelAdapterAnnotationPostProcessor(this.beanFactory)); + postProcessors.put(Router.class, new RouterAnnotationPostProcessor(this.beanFactory)); + postProcessors.put(ServiceActivator.class, new ServiceActivatorAnnotationPostProcessor(this.beanFactory)); + postProcessors.put(Splitter.class, new SplitterAnnotationPostProcessor(this.beanFactory)); + postProcessors.put(Transformer.class, new TransformerAnnotationPostProcessor(this.beanFactory)); } public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java index 7737e86126..dcb9b17bb0 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java @@ -18,8 +18,8 @@ package org.springframework.integration.config.annotation; import java.lang.reflect.Method; +import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.annotation.Router; -import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.MessageConsumer; import org.springframework.integration.router.MethodInvokingChannelResolver; @@ -34,8 +34,8 @@ import org.springframework.util.StringUtils; */ public class RouterAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { - public RouterAnnotationPostProcessor(MessageBus messageBus) { - super(messageBus); + public RouterAnnotationPostProcessor(BeanFactory beanFactory) { + super(beanFactory); } @@ -45,7 +45,7 @@ public class RouterAnnotationPostProcessor extends AbstractMethodAnnotationPostP RouterEndpoint router = new RouterEndpoint(resolver); String defaultOutputChannelName = annotation.defaultOutputChannel(); if (StringUtils.hasText(defaultOutputChannelName)) { - MessageChannel defaultOutputChannel = this.getChannelRegistry().lookupChannel(defaultOutputChannelName); + MessageChannel defaultOutputChannel = this.channelRegistry.lookupChannel(defaultOutputChannelName); Assert.notNull(defaultOutputChannel, "unable to resolve defaultOutputChannel '" + defaultOutputChannelName + "'"); router.setDefaultOutputChannel(defaultOutputChannel); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java index 796ac18af5..261fbbd1d2 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java @@ -18,8 +18,8 @@ package org.springframework.integration.config.annotation; import java.lang.reflect.Method; +import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.integration.bus.MessageBus; import org.springframework.integration.endpoint.ServiceActivatorEndpoint; import org.springframework.integration.message.MessageConsumer; import org.springframework.integration.message.MessageMappingMethodInvoker; @@ -31,8 +31,8 @@ import org.springframework.integration.message.MessageMappingMethodInvoker; */ public class ServiceActivatorAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { - public ServiceActivatorAnnotationPostProcessor(MessageBus messageBus) { - super(messageBus); + public ServiceActivatorAnnotationPostProcessor(BeanFactory beanFactory) { + super(beanFactory); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java index 886e7960c6..f52b3cf539 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java @@ -18,8 +18,8 @@ package org.springframework.integration.config.annotation; import java.lang.reflect.Method; +import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.annotation.Splitter; -import org.springframework.integration.bus.MessageBus; import org.springframework.integration.message.MessageConsumer; import org.springframework.integration.splitter.MethodInvokingSplitter; @@ -30,8 +30,8 @@ import org.springframework.integration.splitter.MethodInvokingSplitter; */ public class SplitterAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { - public SplitterAnnotationPostProcessor(MessageBus messageBus) { - super(messageBus); + public SplitterAnnotationPostProcessor(BeanFactory beanFactory) { + super(beanFactory); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java index 12c16d9286..e8132425c8 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java @@ -18,8 +18,8 @@ package org.springframework.integration.config.annotation; import java.lang.reflect.Method; +import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.annotation.Transformer; -import org.springframework.integration.bus.MessageBus; import org.springframework.integration.message.MessageConsumer; import org.springframework.integration.transformer.MethodInvokingTransformer; import org.springframework.integration.transformer.TransformerEndpoint; @@ -31,8 +31,8 @@ import org.springframework.integration.transformer.TransformerEndpoint; */ public class TransformerAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { - public TransformerAnnotationPostProcessor(MessageBus messageBus) { - super(messageBus); + public TransformerAnnotationPostProcessor(BeanFactory beanFactory) { + super(beanFactory); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractReplyProducingMessageConsumer.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractReplyProducingMessageConsumer.java index 11170c338b..e3a5236882 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractReplyProducingMessageConsumer.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractReplyProducingMessageConsumer.java @@ -40,6 +40,9 @@ import org.springframework.util.Assert; */ public abstract class AbstractReplyProducingMessageConsumer extends AbstractMessageConsumer implements ChannelRegistryAware { + public static final long DEFAULT_SEND_TIMEOUT = 1000; + + private MessageChannel outputChannel; private volatile ChannelRegistry channelRegistry; @@ -48,7 +51,13 @@ public abstract class AbstractReplyProducingMessageConsumer extends AbstractMess private volatile boolean requiresReply = false; - private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); + private final MessageChannelTemplate channelTemplate; + + + public AbstractReplyProducingMessageConsumer() { + this.channelTemplate = new MessageChannelTemplate(); + this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT); + } public void setOutputChannel(MessageChannel outputChannel) { @@ -59,6 +68,13 @@ public abstract class AbstractReplyProducingMessageConsumer extends AbstractMess return this.outputChannel; } + /** + * Set the timeout for sending reply Messages. + */ + public void setSendTimeout(long sendTimeout) { + this.channelTemplate.setSendTimeout(sendTimeout); + } + public void setChannelRegistry(ChannelRegistry channelRegistry) { this.channelRegistry = channelRegistry; } @@ -122,7 +138,7 @@ public abstract class AbstractReplyProducingMessageConsumer extends AbstractMess return false; } - private boolean sendReplyMessage(Message replyMessage, MessageChannel replyChannel) { + protected boolean sendReplyMessage(Message replyMessage, MessageChannel replyChannel) { return this.channelTemplate.send(replyMessage, replyChannel); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java index b436671091..b7b08800a6 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java @@ -28,6 +28,7 @@ import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.channel.ThreadLocalChannel; +import org.springframework.integration.config.MessageBusParser; import org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor; import org.springframework.integration.endpoint.AbstractReplyProducingMessageConsumer; import org.springframework.integration.endpoint.ServiceActivatorEndpoint; @@ -76,7 +77,8 @@ public class DirectChannelSubscriptionTests { public void testSendAndReceiveForAnnotatedEndpoint() { GenericApplicationContext context = new GenericApplicationContext(); bus.setApplicationContext(context); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(bus); + context.getBeanFactory().registerSingleton(MessageBusParser.MESSAGE_BUS_BEAN_NAME, bus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); TestEndpoint endpoint = new TestEndpoint(); @@ -109,10 +111,11 @@ public class DirectChannelSubscriptionTests { public void testExceptionThrownFromAnnotatedEndpoint() { GenericApplicationContext context = new GenericApplicationContext(); bus.setApplicationContext(context); + context.getBeanFactory().registerSingleton(MessageBusParser.MESSAGE_BUS_BEAN_NAME, bus); QueueChannel errorChannel = new QueueChannel(); errorChannel.setBeanName(ChannelRegistry.ERROR_CHANNEL_NAME); context.getBeanFactory().registerSingleton(ChannelRegistry.ERROR_CHANNEL_NAME, errorChannel); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(bus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); FailingTestEndpoint endpoint = new FailingTestEndpoint(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java index 5f8d34d37f..8d52894b2a 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java @@ -37,6 +37,7 @@ import org.springframework.integration.endpoint.SubscribingConsumerEndpoint; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.util.MethodInvoker; +import org.springframework.integration.util.TestUtils; /** * @author Marius Bogoevici @@ -91,7 +92,7 @@ public class AggregatorParserTests { Assert.assertEquals("The AggregatorEndpoint is not injected with the appropriate discard channel", discardChannel, accessor.getPropertyValue("discardChannel")); Assert.assertEquals("The AggregatorEndpoint is not set with the appropriate timeout value", - 86420000l, accessor.getPropertyValue("sendTimeout")); + 86420000l, TestUtils.getPropertyValue(consumer, "channelTemplate.sendTimeout")); Assert.assertEquals( "The AggregatorEndpoint is not configured with the appropriate 'send partial results on timeout' flag", true, accessor.getPropertyValue("sendPartialResultOnTimeout")); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java index 6e1fb26847..e94dd80fde 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java @@ -16,16 +16,21 @@ package org.springframework.integration.config; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.springframework.integration.util.TestUtils.getPropertyValue; + import java.util.ArrayList; import java.util.List; -import org.junit.Assert; import org.junit.Before; import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.aggregator.Resequencer; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.PollableChannel; import org.springframework.integration.endpoint.SubscribingConsumerEndpoint; @@ -60,34 +65,32 @@ public class ResequencerParserTests { Message message1 = outputChannel.receive(500); Message message2 = outputChannel.receive(500); Message message3 = outputChannel.receive(500); - Assert.assertNotNull(message1); - Assert.assertEquals(new Integer(1), message1.getHeaders().getSequenceNumber()); - Assert.assertNotNull(message2); - Assert.assertEquals(new Integer(2), message2.getHeaders().getSequenceNumber()); - Assert.assertNotNull(message3); - Assert.assertEquals(new Integer(3), message3.getHeaders().getSequenceNumber()); + assertNotNull(message1); + assertEquals(new Integer(1), message1.getHeaders().getSequenceNumber()); + assertNotNull(message2); + assertEquals(new Integer(2), message2.getHeaders().getSequenceNumber()); + assertNotNull(message3); + assertEquals(new Integer(3), message3.getHeaders().getSequenceNumber()); } @Test public void testDefaultResequencerProperties() { SubscribingConsumerEndpoint endpoint = (SubscribingConsumerEndpoint) context.getBean("defaultResequencer"); - DirectFieldAccessor accessor = new DirectFieldAccessor(new DirectFieldAccessor(endpoint).getPropertyValue("consumer")); - Assert.assertNull(accessor.getPropertyValue("outputChannel")); - Assert.assertNull(accessor.getPropertyValue("discardChannel")); - Assert.assertEquals("The ResequencerEndpoint is not set with the appropriate timeout value", - 1000l, accessor.getPropertyValue("sendTimeout")); - Assert.assertEquals( - "The ResequencerEndpoint is not configured with the appropriate 'send partial results on timeout' flag", - false, accessor.getPropertyValue("sendPartialResultOnTimeout")); - Assert.assertEquals("The ResequencerEndpoint is not configured with the appropriate reaper interval", - 1000l, accessor.getPropertyValue("reaperInterval")); - Assert.assertEquals( - "The ResequencerEndpoint is not configured with the appropriate tracked correlationId capacity", - 1000, accessor.getPropertyValue("trackedCorrelationIdCapacity")); - Assert.assertEquals("The ResequencerEndpoint is not configured with the appropriate timeout", - 60000l, accessor.getPropertyValue("timeout")); - Assert.assertEquals("The ResequencerEndpoint is not configured with the appropriate 'release partial sequences' flag", - true, accessor.getPropertyValue("releasePartialSequences")); + Resequencer resequencer = (Resequencer) new DirectFieldAccessor(endpoint).getPropertyValue("consumer"); + assertNull(getPropertyValue(resequencer, "outputChannel")); + assertNull(getPropertyValue(resequencer, "discardChannel")); + assertEquals("The ResequencerEndpoint is not set with the appropriate timeout value", + 1000l, getPropertyValue(resequencer, "channelTemplate.sendTimeout")); + assertEquals("The ResequencerEndpoint is not configured with the appropriate 'send partial results on timeout' flag", + false, getPropertyValue(resequencer, "sendPartialResultOnTimeout")); + assertEquals("The ResequencerEndpoint is not configured with the appropriate reaper interval", + 1000l, getPropertyValue(resequencer, "reaperInterval")); + assertEquals("The ResequencerEndpoint is not configured with the appropriate tracked correlationId capacity", + 1000, getPropertyValue(resequencer, "trackedCorrelationIdCapacity")); + assertEquals("The ResequencerEndpoint is not configured with the appropriate timeout", + 60000l, getPropertyValue(resequencer, "timeout")); + assertEquals("The ResequencerEndpoint is not configured with the appropriate 'release partial sequences' flag", + true, getPropertyValue(resequencer, "releasePartialSequences")); } @Test @@ -95,27 +98,26 @@ public class ResequencerParserTests { SubscribingConsumerEndpoint endpoint = (SubscribingConsumerEndpoint) context.getBean("completelyDefinedResequencer"); MessageChannel outputChannel = (MessageChannel) context.getBean("outputChannel"); MessageChannel discardChannel = (MessageChannel) context.getBean("discardChannel"); - DirectFieldAccessor accessor = new DirectFieldAccessor(new DirectFieldAccessor(endpoint).getPropertyValue("consumer")); - Assert.assertEquals("The ResequencerEndpoint is not injected with the appropriate output channel", - outputChannel, accessor.getPropertyValue("outputChannel")); - Assert.assertEquals("The ResequencerEndpoint is not injected with the appropriate discard channel", - discardChannel, accessor.getPropertyValue("discardChannel")); - Assert.assertEquals("The ResequencerEndpoint is not set with the appropriate timeout value", - 86420000l, accessor.getPropertyValue("sendTimeout")); - Assert.assertEquals( - "The ResequencerEndpoint is not configured with the appropriate 'send partial results on timeout' flag", - true, accessor.getPropertyValue("sendPartialResultOnTimeout")); - Assert.assertEquals("The ResequencerEndpoint is not configured with the appropriate reaper interval", - 135l, accessor.getPropertyValue("reaperInterval")); - Assert.assertEquals( - "The ResequencerEndpoint is not configured with the appropriate tracked correlationId capacity", - 99, accessor.getPropertyValue("trackedCorrelationIdCapacity")); - Assert.assertEquals("The ResequencerEndpoint is not configured with the appropriate timeout", - 42l, accessor.getPropertyValue("timeout")); - Assert.assertEquals("The ResequencerEndpoint is not configured with the appropriate 'release partial sequences' flag", - false, accessor.getPropertyValue("releasePartialSequences")); + Resequencer resequencer = (Resequencer) new DirectFieldAccessor(endpoint).getPropertyValue("consumer"); + assertEquals("The ResequencerEndpoint is not injected with the appropriate output channel", + outputChannel, getPropertyValue(resequencer, "outputChannel")); + assertEquals("The ResequencerEndpoint is not injected with the appropriate discard channel", + discardChannel, getPropertyValue(resequencer, "discardChannel")); + assertEquals("The ResequencerEndpoint is not set with the appropriate timeout value", + 86420000l, getPropertyValue(resequencer, "channelTemplate.sendTimeout")); + assertEquals("The ResequencerEndpoint is not configured with the appropriate 'send partial results on timeout' flag", + true, getPropertyValue(resequencer, "sendPartialResultOnTimeout")); + assertEquals("The ResequencerEndpoint is not configured with the appropriate reaper interval", + 135l, getPropertyValue(resequencer, "reaperInterval")); + assertEquals("The ResequencerEndpoint is not configured with the appropriate tracked correlationId capacity", + 99, getPropertyValue(resequencer, "trackedCorrelationIdCapacity")); + assertEquals("The ResequencerEndpoint is not configured with the appropriate timeout", + 42l, getPropertyValue(resequencer, "timeout")); + assertEquals("The ResequencerEndpoint is not configured with the appropriate 'release partial sequences' flag", + false, getPropertyValue(resequencer, "releasePartialSequences")); } + private static Message createMessage(T payload, Object correlationId, int sequenceSize, int sequenceNumber, MessageChannel outputChannel) { return MessageBuilder.withPayload(payload) diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ServiceActivatorAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/ServiceActivatorAnnotationPostProcessorTests.java index 8504a9e603..7abdebb44f 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/ServiceActivatorAnnotationPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ServiceActivatorAnnotationPostProcessorTests.java @@ -24,7 +24,6 @@ import java.util.concurrent.TimeUnit; import org.junit.Test; -import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.annotation.MessageEndpoint; @@ -51,7 +50,6 @@ public class ServiceActivatorAnnotationPostProcessorTests { String busBeanName = MessageBusParser.MESSAGE_BUS_BEAN_NAME; context.registerBeanDefinition(busBeanName, new RootBeanDefinition(DefaultMessageBus.class)); RootBeanDefinition postProcessorDef = new RootBeanDefinition(MessagingAnnotationPostProcessor.class); - postProcessorDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(busBeanName)); context.registerBeanDefinition("postProcessor", postProcessorDef); context.refresh(); context.start(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java index b1dea7b150..5315d4e248 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java @@ -16,6 +16,13 @@ package org.springframework.integration.config.annotation; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; + +import static org.springframework.integration.util.TestUtils.getPropertyValue; + import java.lang.reflect.Method; import org.junit.Assert; @@ -42,17 +49,18 @@ public class AggregatorAnnotationTests { ApplicationContext context = new ClassPathXmlApplicationContext( new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); final String endpointName = "endpointWithDefaultAnnotation"; - DirectFieldAccessor accessor = getDirectFieldAccessorForAggregatingHandler(context, - endpointName); - Assert.assertTrue(accessor.getPropertyValue("completionStrategy") instanceof SequenceSizeCompletionStrategy); - Assert.assertNull(accessor.getPropertyValue("outputChannel")); - Assert.assertNull(accessor.getPropertyValue("discardChannel")); - Assert.assertEquals(AbstractMessageAggregator.DEFAULT_SEND_TIMEOUT, accessor.getPropertyValue("sendTimeout")); - Assert.assertEquals(AbstractMessageAggregator.DEFAULT_TIMEOUT, accessor.getPropertyValue("timeout")); - Assert.assertEquals(false, accessor.getPropertyValue("sendPartialResultOnTimeout")); - Assert.assertEquals(AbstractMessageAggregator.DEFAULT_REAPER_INTERVAL, accessor.getPropertyValue("reaperInterval")); - Assert.assertEquals(AbstractMessageAggregator.DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY, - accessor.getPropertyValue("trackedCorrelationIdCapacity")); + AbstractMessageAggregator aggregator = this.getAggregator(context, endpointName); + assertTrue(getPropertyValue(aggregator, "completionStrategy") instanceof SequenceSizeCompletionStrategy); + assertNull(getPropertyValue(aggregator, "outputChannel")); + assertNull(getPropertyValue(aggregator, "discardChannel")); + assertEquals(AbstractMessageAggregator.DEFAULT_SEND_TIMEOUT, + getPropertyValue(aggregator, "channelTemplate.sendTimeout")); + assertEquals(AbstractMessageAggregator.DEFAULT_TIMEOUT, getPropertyValue(aggregator, "timeout")); + assertEquals(false, getPropertyValue(aggregator, "sendPartialResultOnTimeout")); + assertEquals(AbstractMessageAggregator.DEFAULT_REAPER_INTERVAL, + getPropertyValue(aggregator, "reaperInterval")); + assertEquals(AbstractMessageAggregator.DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY, + getPropertyValue(aggregator, "trackedCorrelationIdCapacity")); } @Test @@ -60,15 +68,18 @@ public class AggregatorAnnotationTests { ApplicationContext context = new ClassPathXmlApplicationContext( new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); final String endpointName = "endpointWithCustomizedAnnotation"; - DirectFieldAccessor accessor = getDirectFieldAccessorForAggregatingHandler(context, endpointName); - Assert.assertTrue(accessor.getPropertyValue("completionStrategy") instanceof SequenceSizeCompletionStrategy); - Assert.assertEquals(getMessageBus(context).lookupChannel("outputChannel"), accessor.getPropertyValue("outputChannel")); - Assert.assertEquals(getMessageBus(context).lookupChannel("discardChannel"), accessor.getPropertyValue("discardChannel")); - Assert.assertEquals(98765432l, accessor.getPropertyValue("sendTimeout")); - Assert.assertEquals(4567890l, accessor.getPropertyValue("timeout")); - Assert.assertEquals(true, accessor.getPropertyValue("sendPartialResultOnTimeout")); - Assert.assertEquals(1234l, accessor.getPropertyValue("reaperInterval")); - Assert.assertEquals(42, accessor.getPropertyValue("trackedCorrelationIdCapacity")); + AbstractMessageAggregator aggregator = this.getAggregator(context, endpointName); + assertTrue(getPropertyValue(aggregator, "completionStrategy") + instanceof SequenceSizeCompletionStrategy); + assertEquals(getMessageBus(context).lookupChannel("outputChannel"), + getPropertyValue(aggregator, "outputChannel")); + assertEquals(getMessageBus(context).lookupChannel("discardChannel"), + getPropertyValue(aggregator, "discardChannel")); + assertEquals(98765432l, getPropertyValue(aggregator, "channelTemplate.sendTimeout")); + assertEquals(4567890l, getPropertyValue(aggregator, "timeout")); + assertEquals(true, getPropertyValue(aggregator, "sendPartialResultOnTimeout")); + assertEquals(1234l, getPropertyValue(aggregator, "reaperInterval")); + assertEquals(42, getPropertyValue(aggregator, "trackedCorrelationIdCapacity")); } @Test @@ -76,24 +87,24 @@ public class AggregatorAnnotationTests { ApplicationContext context = new ClassPathXmlApplicationContext( new String[] { "classpath:/org/springframework/integration/config/annotation/testAnnotatedAggregator.xml" }); final String endpointName = "endpointWithDefaultAnnotationAndCustomCompletionStrategy"; - DirectFieldAccessor aggregatingMessageHandlerAccessor = getDirectFieldAccessorForAggregatingHandler(context, endpointName); - Object completionStrategy = aggregatingMessageHandlerAccessor.getPropertyValue("completionStrategy"); + AbstractMessageAggregator aggregator = this.getAggregator(context, endpointName); + Object completionStrategy = getPropertyValue(aggregator, "completionStrategy"); Assert.assertTrue(completionStrategy instanceof CompletionStrategyAdapter); CompletionStrategyAdapter completionStrategyAdapter = (CompletionStrategyAdapter) completionStrategy; DirectFieldAccessor invokerAccessor = new DirectFieldAccessor( new DirectFieldAccessor(completionStrategyAdapter).getPropertyValue("invoker")); Object targetObject = invokerAccessor.getPropertyValue("object"); - Assert.assertSame(context.getBean(endpointName), targetObject); + assertSame(context.getBean(endpointName), targetObject); Method completionCheckerMethod = (Method) invokerAccessor.getPropertyValue("method"); - Assert.assertEquals("completionChecker", completionCheckerMethod.getName()); + assertEquals("completionChecker", completionCheckerMethod.getName()); } @SuppressWarnings("unchecked") - private DirectFieldAccessor getDirectFieldAccessorForAggregatingHandler(ApplicationContext context, final String endpointName) { + private AbstractMessageAggregator getAggregator(ApplicationContext context, final String endpointName) { SubscribingConsumerEndpoint endpoint = (SubscribingConsumerEndpoint) context.getBean( endpointName + ".aggregatingMethod.aggregator"); - return new DirectFieldAccessor(new DirectFieldAccessor(endpoint).getPropertyValue("consumer")); + return (AbstractMessageAggregator) new DirectFieldAccessor(endpoint).getPropertyValue("consumer"); } private MessageBus getMessageBus(ApplicationContext context) { diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java index b1f25d5b1f..a71735e8f6 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java @@ -29,6 +29,7 @@ import org.junit.Test; import org.springframework.aop.framework.ProxyFactory; import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.NoSuchBeanDefinitionException; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.context.support.GenericApplicationContext; @@ -61,11 +62,13 @@ public class MessagingAnnotationPostProcessorTests { public void testServiceActivatorAnnotation() { GenericApplicationContext context = new GenericApplicationContext(); DefaultMessageBus messageBus = new DefaultMessageBus(); + context.getBeanFactory().registerSingleton( + MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); messageBus.setApplicationContext(context); QueueChannel inputChannel = new QueueChannel(); inputChannel.setBeanName("inputChannel"); messageBus.registerChannel(inputChannel); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); ServiceActivatorAnnotatedBean bean = new ServiceActivatorAnnotatedBean(); @@ -139,8 +142,10 @@ public class MessagingAnnotationPostProcessorTests { public void testTargetAnnotation() throws InterruptedException { GenericApplicationContext context = new GenericApplicationContext(); DefaultMessageBus messageBus = new DefaultMessageBus(); + context.getBeanFactory().registerSingleton( + MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); messageBus.setApplicationContext(context); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); CountDownLatch latch = new CountDownLatch(1); @@ -155,20 +160,33 @@ public class MessagingAnnotationPostProcessorTests { messageBus.stop(); } - @Test(expected=IllegalArgumentException.class) - public void testPostProcessorWithNullMessageBus() { - new MessagingAnnotationPostProcessor(null); + @Test(expected = IllegalArgumentException.class) + public void testPostProcessorWithoutBeanFactory() { + MessagingAnnotationPostProcessor postProcessor = + new MessagingAnnotationPostProcessor(); + postProcessor.afterPropertiesSet(); + } + + @Test(expected = NoSuchBeanDefinitionException.class) + public void testPostProcessorWithoutMessageBus() { + GenericApplicationContext context = new GenericApplicationContext(); + MessagingAnnotationPostProcessor postProcessor = + new MessagingAnnotationPostProcessor(); + postProcessor.setBeanFactory(context.getBeanFactory()); + postProcessor.afterPropertiesSet(); } @Test public void testChannelRegistryAwareBean() { GenericApplicationContext context = new GenericApplicationContext(); DefaultMessageBus messageBus = new DefaultMessageBus(); + context.getBeanFactory().registerSingleton( + MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); messageBus.setApplicationContext(context); QueueChannel inputChannel = new QueueChannel(); inputChannel.setBeanName("inputChannel"); messageBus.registerChannel(inputChannel); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); ChannelRegistryAwareTestBean testBean = new ChannelRegistryAwareTestBean(); @@ -182,15 +200,17 @@ public class MessagingAnnotationPostProcessorTests { @Test public void testProxiedMessageEndpointAnnotation() { GenericApplicationContext context = new GenericApplicationContext(); + DefaultMessageBus messageBus = new DefaultMessageBus(); + context.getBeanFactory().registerSingleton( + MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); QueueChannel inputChannel = new QueueChannel(); QueueChannel outputChannel = new QueueChannel(); inputChannel.setBeanName("inputChannel"); outputChannel.setBeanName("outputChannel"); context.getBeanFactory().registerSingleton("inputChannel", inputChannel); context.getBeanFactory().registerSingleton("outputChannel", outputChannel); - DefaultMessageBus messageBus = new DefaultMessageBus(); messageBus.setApplicationContext(context); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); ProxyFactory proxyFactory = new ProxyFactory(new SimpleAnnotatedEndpoint()); @@ -206,15 +226,17 @@ public class MessagingAnnotationPostProcessorTests { @Test public void testMessageEndpointAnnotationInherited() { GenericApplicationContext context = new GenericApplicationContext(); + DefaultMessageBus messageBus = new DefaultMessageBus(); + context.getBeanFactory().registerSingleton( + MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); QueueChannel inputChannel = new QueueChannel(); QueueChannel outputChannel = new QueueChannel(); inputChannel.setBeanName("inputChannel"); outputChannel.setBeanName("outputChannel"); context.getBeanFactory().registerSingleton("inputChannel", inputChannel); context.getBeanFactory().registerSingleton("outputChannel", outputChannel); - DefaultMessageBus messageBus = new DefaultMessageBus(); messageBus.setApplicationContext(context); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); postProcessor.postProcessAfterInitialization(new SimpleAnnotatedEndpointSubclass(), "subclass"); @@ -228,15 +250,17 @@ public class MessagingAnnotationPostProcessorTests { @Test public void testMessageEndpointAnnotationInheritedWithProxy() { GenericApplicationContext context = new GenericApplicationContext(); + DefaultMessageBus messageBus = new DefaultMessageBus(); + context.getBeanFactory().registerSingleton( + MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); QueueChannel inputChannel = new QueueChannel(); QueueChannel outputChannel = new QueueChannel(); inputChannel.setBeanName("inputChannel"); outputChannel.setBeanName("outputChannel"); context.getBeanFactory().registerSingleton("inputChannel", inputChannel); context.getBeanFactory().registerSingleton("outputChannel", outputChannel); - DefaultMessageBus messageBus = new DefaultMessageBus(); messageBus.setApplicationContext(context); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); ProxyFactory proxyFactory = new ProxyFactory(new SimpleAnnotatedEndpointSubclass()); @@ -252,15 +276,17 @@ public class MessagingAnnotationPostProcessorTests { @Test public void testMessageEndpointAnnotationInheritedFromInterface() { GenericApplicationContext context = new GenericApplicationContext(); + DefaultMessageBus messageBus = new DefaultMessageBus(); + context.getBeanFactory().registerSingleton( + MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); QueueChannel inputChannel = new QueueChannel(); QueueChannel outputChannel = new QueueChannel(); inputChannel.setBeanName("inputChannel"); outputChannel.setBeanName("outputChannel"); context.getBeanFactory().registerSingleton("inputChannel", inputChannel); context.getBeanFactory().registerSingleton("outputChannel", outputChannel); - DefaultMessageBus messageBus = new DefaultMessageBus(); messageBus.setApplicationContext(context); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); postProcessor.postProcessAfterInitialization(new SimpleAnnotatedEndpointImplementation(), "impl"); @@ -274,15 +300,17 @@ public class MessagingAnnotationPostProcessorTests { @Test public void testMessageEndpointAnnotationInheritedFromInterfaceWithAutoCreatedChannels() { GenericApplicationContext context = new GenericApplicationContext(); + DefaultMessageBus messageBus = new DefaultMessageBus(); + context.getBeanFactory().registerSingleton( + MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); QueueChannel inputChannel = new QueueChannel(); QueueChannel outputChannel = new QueueChannel(); inputChannel.setBeanName("inputChannel"); outputChannel.setBeanName("outputChannel"); context.getBeanFactory().registerSingleton("inputChannel", inputChannel); context.getBeanFactory().registerSingleton("outputChannel", outputChannel); - DefaultMessageBus messageBus = new DefaultMessageBus(); messageBus.setApplicationContext(context); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); postProcessor.postProcessAfterInitialization(new SimpleAnnotatedEndpointImplementation(), "impl"); @@ -296,15 +324,17 @@ public class MessagingAnnotationPostProcessorTests { @Test public void testMessageEndpointAnnotationInheritedFromInterfaceWithProxy() { GenericApplicationContext context = new GenericApplicationContext(); + DefaultMessageBus messageBus = new DefaultMessageBus(); + context.getBeanFactory().registerSingleton( + MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); QueueChannel inputChannel = new QueueChannel(); QueueChannel outputChannel = new QueueChannel(); inputChannel.setBeanName("inputChannel"); outputChannel.setBeanName("outputChannel"); context.getBeanFactory().registerSingleton("inputChannel", inputChannel); context.getBeanFactory().registerSingleton("outputChannel", outputChannel); - DefaultMessageBus messageBus = new DefaultMessageBus(); messageBus.setApplicationContext(context); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); ProxyFactory proxyFactory = new ProxyFactory(new SimpleAnnotatedEndpointImplementation()); @@ -320,12 +350,14 @@ public class MessagingAnnotationPostProcessorTests { @Test public void testEndpointWithPollerAnnotation() { GenericApplicationContext context = new GenericApplicationContext(); + DefaultMessageBus messageBus = new DefaultMessageBus(); + context.getBeanFactory().registerSingleton( + MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); QueueChannel testChannel = new QueueChannel(); testChannel.setBeanName("testChannel"); context.getBeanFactory().registerSingleton("testChannel", testChannel); - DefaultMessageBus messageBus = new DefaultMessageBus(); messageBus.setApplicationContext(context); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); AnnotatedEndpointWithPolledAnnotation bean = new AnnotatedEndpointWithPolledAnnotation(); @@ -344,8 +376,10 @@ public class MessagingAnnotationPostProcessorTests { public void testChannelAdapterAnnotation() throws InterruptedException { GenericApplicationContext context = new GenericApplicationContext(); DefaultMessageBus messageBus = new DefaultMessageBus(); + context.getBeanFactory().registerSingleton( + MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); messageBus.setApplicationContext(context); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); ChannelAdapterAnnotationTestBean testBean = new ChannelAdapterAnnotationTestBean(); @@ -379,7 +413,7 @@ public class MessagingAnnotationPostProcessorTests { DefaultMessageBus messageBus = new DefaultMessageBus(); context.getBeanFactory().registerSingleton(MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); messageBus.setApplicationContext(context); - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); TransformerAnnotationTestBean testBean = new TransformerAnnotationTestBean(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessorTests.java index af1d544932..817f8de3a8 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessorTests.java @@ -27,6 +27,7 @@ import org.springframework.integration.annotation.Router; import org.springframework.integration.bus.DefaultMessageBus; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.MessageBusParser; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; @@ -51,12 +52,14 @@ public class RouterAnnotationPostProcessorTests { outputChannel.setBeanName("output"); context.getBeanFactory().registerSingleton("input", inputChannel); context.getBeanFactory().registerSingleton("output", outputChannel); + context.getBeanFactory().registerSingleton( + MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); } @Test public void testRouter() { - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); TestRouter testRouter = new TestRouter(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessorTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessorTests.java index 3f2d3d210c..10dde19dcb 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessorTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessorTests.java @@ -29,6 +29,7 @@ import org.springframework.integration.annotation.Splitter; import org.springframework.integration.bus.DefaultMessageBus; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.MessageBusParser; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; @@ -52,13 +53,15 @@ public class SplitterAnnotationPostProcessorTests { outputChannel.setBeanName("output"); context.getBeanFactory().registerSingleton("input", inputChannel); context.getBeanFactory().registerSingleton("output", outputChannel); + context.getBeanFactory().registerSingleton( + MessageBusParser.MESSAGE_BUS_BEAN_NAME, messageBus); messageBus.setApplicationContext(context); } @Test public void testSplitterAnnotation() throws InterruptedException { - MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(messageBus); + MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor(); postProcessor.setBeanFactory(context.getBeanFactory()); postProcessor.afterPropertiesSet(); TestSplitter splitter = new TestSplitter(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/typeConvertingEndpointTests.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/typeConvertingEndpointTests.xml index 54b151983f..1b5f5cff02 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/typeConvertingEndpointTests.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/typeConvertingEndpointTests.xml @@ -7,7 +7,9 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - + + + @@ -17,8 +19,4 @@ - - - - diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/util/TestUtils.java b/org.springframework.integration/src/test/java/org/springframework/integration/util/TestUtils.java new file mode 100644 index 0000000000..27c89d1380 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/util/TestUtils.java @@ -0,0 +1,54 @@ +/* + * Copyright 2002-2008 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.util; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.util.Assert; + +/** + * @author Mark Fisher + */ +public abstract class TestUtils { + + public static Object getPropertyValue(Object root, String propertyPath) { + Object value = null; + DirectFieldAccessor accessor = new DirectFieldAccessor(root); + String[] tokens = propertyPath.split("\\."); + for (int i = 0; i < tokens.length; i++) { + value = accessor.getPropertyValue(tokens[i]); + if (value != null) { + accessor = new DirectFieldAccessor(value); + } + else if (i == tokens.length - 1) { + return null; + } + else { + throw new IllegalArgumentException( + "intermediate property '" + tokens[i] + "' is null"); + } + } + return value; + } + + @SuppressWarnings("unchecked") + public static T getPropertyValue(Object root, String propertyPath, Class type) { + Object value = getPropertyValue(root, propertyPath); + Assert.isAssignable(type, value.getClass()); + return (T) value; + } + +}