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

Added support for the @Transactional annotation within the @Poller annotation (INT-399). Simplified AbstractMessageBarrierConsumer somewhat (to reuse base class methods for sending replies and discarding incomplete messages). Added TestUtils with convenient property-path accessor methods. Refactored annotation post-processors such that the BeanFactory is passed to the individual method post-processors rather than passing the MessageBus. The base class still detects the MessageBus bean and provides a protected 'channelRegistry' reference.

Browse Source
This commit is contained in:
Mark Fisher
2008-10-08 00:19:20 +00:00
parent ea220cac6c
commit 6c4245f3ca
21 changed files with 331 additions and 177 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierConsumer.java
View File

@@ -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<Object, MessageBarrier> barriers =
new ConcurrentHashMap<Object, MessageBarrier>();
@@ -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);

79
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
View File

@@ -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<T extends Annotation
private static final String OUTPUT_CHANNEL_ATTRIBUTE = "outputChannel";
private final MessageBus messageBus;
private final BeanFactory beanFactory;
protected final ChannelRegistry channelRegistry;
public AbstractMethodAnnotationPostProcessor(MessageBus messageBus) {
Assert.notNull(messageBus, "MessageBus must not be null");
this.messageBus = messageBus;
public AbstractMethodAnnotationPostProcessor(BeanFactory beanFactory) {
Assert.notNull(beanFactory, "BeanFactory must not be null");
this.beanFactory = beanFactory;
this.channelRegistry = (ChannelRegistry) this.beanFactory.getBean(
MessageBusParser.MESSAGE_BUS_BEAN_NAME);
}
protected ChannelRegistry getChannelRegistry() {
return this.messageBus;
}
public Object postProcess(Object bean, String beanName, Method method, T annotation) {
MessageConsumer consumer = this.createConsumer(bean, method, annotation);
if (consumer instanceof ChannelRegistryAware) {
((ChannelRegistryAware) consumer).setChannelRegistry(this.getChannelRegistry());
((ChannelRegistryAware) consumer).setChannelRegistry(this.channelRegistry);
}
Poller pollerAnnotation = AnnotationUtils.findAnnotation(method, Poller.class);
MessageEndpoint endpoint = this.createEndpoint(consumer, annotation, pollerAnnotation);
@@ -93,7 +101,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
MessageEndpoint endpoint = null;
String inputChannelName = (String) AnnotationUtils.getValue(annotation, INPUT_CHANNEL_ATTRIBUTE);
if (StringUtils.hasText(inputChannelName)) {
MessageChannel inputChannel = this.messageBus.lookupChannel(inputChannelName);
MessageChannel inputChannel = this.channelRegistry.lookupChannel(inputChannelName);
Assert.notNull(inputChannel, "unable to resolve inputChannel '" + inputChannelName + "'");
if (consumer instanceof AbstractMessageConsumer) {
if (inputChannel instanceof PollableChannel) {
@@ -106,6 +114,15 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
trigger.setFixedRate(pollerAnnotation.fixedRate());
pollingEndpoint.setTrigger(trigger);
pollingEndpoint.setMaxMessagesPerPoll(pollerAnnotation.maxMessagesPerPoll());
if (StringUtils.hasText(pollerAnnotation.transactionManager())) {
String txManagerRef = pollerAnnotation.transactionManager();
Assert.isTrue(this.beanFactory.containsBean(txManagerRef), "no such bean '" + txManagerRef + "'");
PlatformTransactionManager txManager = (PlatformTransactionManager)
this.beanFactory.getBean(txManagerRef, PlatformTransactionManager.class);
pollingEndpoint.setTransactionManager(txManager);
Transactional txAnnotation = pollerAnnotation.transactionAttributes();
pollingEndpoint.setTransactionDefinition(this.parseTransactionAnnotation(txAnnotation));
}
}
endpoint = pollingEndpoint;
}
@@ -122,7 +139,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
if (consumer instanceof AbstractReplyProducingMessageConsumer) {
String outputChannelName = (String) AnnotationUtils.getValue(annotation, OUTPUT_CHANNEL_ATTRIBUTE);
if (StringUtils.hasText(outputChannelName)) {
MessageChannel outputChannel = this.messageBus.lookupChannel(outputChannelName);
MessageChannel outputChannel = this.channelRegistry.lookupChannel(outputChannelName);
Assert.notNull(outputChannel, "unable to resolve outputChannel '" + outputChannelName + "'");
((AbstractReplyProducingMessageConsumer) consumer).setOutputChannel(outputChannel);
}
@@ -131,6 +148,44 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
return endpoint;
}
@SuppressWarnings("unchecked")
private TransactionAttribute parseTransactionAnnotation(Transactional annotation) {
if (annotation == null) {
return null;
}
RuleBasedTransactionAttribute rbta = new RuleBasedTransactionAttribute();
rbta.setPropagationBehavior(annotation.propagation().value());
rbta.setIsolationLevel(annotation.isolation().value());
rbta.setTimeout(annotation.timeout());
rbta.setReadOnly(annotation.readOnly());
ArrayList<RollbackRuleAttribute> rollBackRules = new ArrayList<RollbackRuleAttribute>();
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);
}

8
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java
View File

@@ -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<Aggregator> {
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);
}

2
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AnnotationDrivenParser.java
View File

@@ -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);

15
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java
View File

@@ -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<ChannelAdapter> {
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);

22
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java
View File

@@ -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<Class<? extends Annotation>, 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 {

8
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessor.java
View File

@@ -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<Router> {
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);
}

6
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ServiceActivatorAnnotationPostProcessor.java
View File

@@ -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<ServiceActivator> {
public ServiceActivatorAnnotationPostProcessor(MessageBus messageBus) {
super(messageBus);
public ServiceActivatorAnnotationPostProcessor(BeanFactory beanFactory) {
super(beanFactory);
}

6
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessor.java
View File

@@ -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<Splitter> {
public SplitterAnnotationPostProcessor(MessageBus messageBus) {
super(messageBus);
public SplitterAnnotationPostProcessor(BeanFactory beanFactory) {
super(beanFactory);
}

6
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/TransformerAnnotationPostProcessor.java
View File

@@ -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<Transformer> {
public TransformerAnnotationPostProcessor(MessageBus messageBus) {
super(messageBus);
public TransformerAnnotationPostProcessor(BeanFactory beanFactory) {
super(beanFactory);
}

20
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractReplyProducingMessageConsumer.java
View File

@@ -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);
}

7
org.springframework.integration/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java
View File

@@ -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();

3
org.springframework.integration/src/test/java/org/springframework/integration/config/AggregatorParserTests.java
View File

@@ -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"));

88
org.springframework.integration/src/test/java/org/springframework/integration/config/ResequencerParserTests.java
View File

@@ -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 <T> Message<T> createMessage(T payload, Object correlationId,
int sequenceSize, int sequenceNumber, MessageChannel outputChannel) {
return MessageBuilder.withPayload(payload)

2
org.springframework.integration/src/test/java/org/springframework/integration/config/ServiceActivatorAnnotationPostProcessorTests.java
View File

@@ -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();

63
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java
View File

@@ -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) {

78
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java
View File

@@ -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();

5
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/RouterAnnotationPostProcessorTests.java
View File

@@ -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();

5
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/SplitterAnnotationPostProcessorTests.java
View File

@@ -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();

8
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/typeConvertingEndpointTests.xml
View File

@@ -7,7 +7,9 @@
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<bean id="bus" class="org.springframework.integration.bus.DefaultMessageBus"/>
<integration:message-bus/>
<integration:annotation-driven/>
<integration:channel id="inputChannel"/>
@@ -17,8 +19,4 @@
<bean id="endpoint" class="org.springframework.integration.config.annotation.TypeConvertingTestEndpoint"/>
<bean class="org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor">
<constructor-arg ref="bus"/>
</bean>
</beans>

54
org.springframework.integration/src/test/java/org/springframework/integration/util/TestUtils.java Normal file
View File

@@ -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> T getPropertyValue(Object root, String propertyPath, Class<T> type) {
Object value = getPropertyValue(root, propertyPath);
Assert.isAssignable(type, value.getClass());
return (T) value;
}
}