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

INT-3319 Fix BeanFactory Propagation

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

The `MessageBuilderFactory` abstraction relies on the bean factory
being propagated to all classes that create messages.

Some classes were missed in the initial PR.
This commit is contained in:
Gary Russell
2014-03-11 16:24:29 -04:00
committed by Artem Bilan
parent 1c9bcaccee
commit 75af72a77c
39 changed files with 364 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PublishSubscribeAmqpChannel.java
View File

@@ -99,7 +99,9 @@ public class PublishSubscribeAmqpChannel extends AbstractSubscribableAmqpChannel
@Override
protected AbstractDispatcher createDispatcher() {
return new BroadcastingDispatcher(true);
BroadcastingDispatcher broadcastingDispatcher = new BroadcastingDispatcher(true);
broadcastingDispatcher.setBeanFactory(this.getBeanFactory());
return broadcastingDispatcher;
}
@Override

3
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/config/AmqpChannelFactoryBean.java
View File

@@ -318,6 +318,9 @@ public class AmqpChannelFactoryBean extends AbstractFactoryBean<AbstractAmqpChan
if (this.maxSubscribers != null) {
pubsub.setMaxSubscribers(this.maxSubscribers);
}
if (this.getBeanFactory() != null) {
pubsub.setBeanFactory(this.getBeanFactory());
}
this.channel = pubsub;
}
else {

2
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests-context.xml
View File

@@ -19,4 +19,6 @@
<amqp:channel id="channelWithSubscriberLimit" max-subscribers="1" />
<amqp:publish-subscribe-channel id="pubSub" />
</beans>

8
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/config/AmqpChannelParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
package org.springframework.integration.amqp.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import java.util.List;
@@ -25,6 +26,7 @@ import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
@@ -51,6 +53,10 @@ public class AmqpChannelParserTests {
assertEquals(TestInterceptor.class, interceptorList.get(0).getClass());
assertEquals(Integer.MAX_VALUE, TestUtils.getPropertyValue(
TestUtils.getPropertyValue(channel, "dispatcher"), "maxSubscribers", Integer.class).intValue());
channel = context.getBean("pubSub", MessageChannel.class);
Object mbf = context.getBean(IntegrationContextUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
assertSame(mbf, TestUtils.getPropertyValue(channel, "dispatcher.messageBuilderFactory"));
assertSame(mbf, TestUtils.getPropertyValue(channel, "container.messageListener.messageBuilderFactory"));
}
@Test

13
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java
View File

@@ -21,7 +21,11 @@ import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
@@ -39,15 +43,16 @@ import org.springframework.util.Assert;
* @author Dave Syer
* @since 2.0
*/
public abstract class AbstractAggregatingMessageGroupProcessor implements MessageGroupProcessor {
public abstract class AbstractAggregatingMessageGroupProcessor implements MessageGroupProcessor,
BeanFactoryAware {
private final Log logger = LogFactory.getLog(this.getClass());
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
Assert.notNull(messageBuilderFactory, "'messageBuilderFactory' cannot be null");
this.messageBuilderFactory = messageBuilderFactory;
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(beanFactory);
}
@Override

5
spring-integration-core/src/main/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -30,6 +30,7 @@ import org.springframework.integration.store.MessageGroup;
*
* @author Alex Peters
* @author Dave Syer
* @author Gary Russell
*/
public class ExpressionEvaluatingMessageGroupProcessor extends AbstractAggregatingMessageGroupProcessor implements BeanFactoryAware {
@@ -40,7 +41,9 @@ public class ExpressionEvaluatingMessageGroupProcessor extends AbstractAggregati
processor = new ExpressionEvaluatingMessageListProcessor(expression);
}
@Override
public void setBeanFactory(BeanFactory beanFactory) {
super.setBeanFactory(beanFactory);
processor.setBeanFactory(beanFactory);
}

19
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,16 +22,17 @@ import java.util.Map;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.convert.ConversionService;
import org.springframework.messaging.Message;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.store.MessageGroup;
import org.springframework.messaging.Message;
/**
* MessageGroupProcessor that serves as an adapter for the invocation of a POJO method.
*
*
* @author Iwein Fuld
* @author Mark Fisher
* @author Dave Syer
* @author Gary Russell
* @since 2.0
*/
public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMessageGroupProcessor {
@@ -41,7 +42,7 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess
/**
* Creates a wrapper around the object passed in. This constructor will look for a method that can process
* a list of messages.
*
*
* @param target the object to wrap
*/
public MethodInvokingMessageGroupProcessor(Object target) {
@@ -51,7 +52,7 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess
/**
* Creates a wrapper around the object passed in. This constructor will look for a named method specifically and
* fail when it cannot find a method with the given name.
*
*
* @param target the object to wrap
* @param methodName the name of the method to invoke
*/
@@ -61,19 +62,21 @@ public class MethodInvokingMessageGroupProcessor extends AbstractAggregatingMess
/**
* Creates a wrapper around the object passed in.
*
*
* @param target the object to wrap
* @param method the method to invoke
*/
public MethodInvokingMessageGroupProcessor(Object target, Method method) {
this.processor = new MethodInvokingMessageListProcessor<Object>(target, method);
}
public void setConversionService(ConversionService conversionService) {
processor.setConversionService(conversionService);
}
@Override
public void setBeanFactory(BeanFactory beanFactory) {
super.setBeanFactory(beanFactory);
processor.setBeanFactory(beanFactory);
}

1
spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java
View File

@@ -164,6 +164,7 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel {
Integer maxSubscribers = this.getIntegrationProperty(IntegrationProperties.CHANNELS_MAX_BROADCAST_SUBSCRIBERS, Integer.class);
this.setMaxSubscribers(maxSubscribers);
}
this.dispatcher.setBeanFactory(this.getBeanFactory());
}
@Override

43
spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AggregatorAnnotationPostProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -20,9 +20,9 @@ import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import java.util.concurrent.atomic.AtomicReference;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.messaging.MessageChannel;
import org.springframework.integration.aggregator.AggregatingMessageHandler;
import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy;
import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor;
@@ -30,8 +30,9 @@ import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.CorrelationStrategy;
import org.springframework.integration.annotation.ReleaseStrategy;
import org.springframework.messaging.MessageHandler;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
@@ -44,14 +45,18 @@ import org.springframework.util.StringUtils;
*/
public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor<Aggregator> {
private final BeanFactory beanFactory;
public AggregatorAnnotationPostProcessor(ListableBeanFactory beanFactory) {
super(beanFactory);
this.beanFactory = beanFactory;
}
@Override
protected MessageHandler createHandler(Object bean, Method method, Aggregator annotation) {
MethodInvokingMessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(bean, method);
processor.setBeanFactory(this.beanFactory);
MethodInvokingReleaseStrategy releaseStrategy = getReleaseStrategy(bean);
MethodInvokingCorrelationStrategy correlationStrategy = getCorrelationStrategy(bean);
AggregatingMessageHandler handler = new AggregatingMessageHandler(processor, new SimpleMessageStore(), correlationStrategy, releaseStrategy);
@@ -73,29 +78,31 @@ public class AggregatorAnnotationPostProcessor extends AbstractMethodAnnotationP
}
private MethodInvokingReleaseStrategy getReleaseStrategy(final Object bean) {
final AtomicReference<MethodInvokingReleaseStrategy> reference = new AtomicReference<MethodInvokingReleaseStrategy>();
final AtomicReference<MethodInvokingReleaseStrategy> reference = new AtomicReference<MethodInvokingReleaseStrategy>();
ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
@Override
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, ReleaseStrategy.class);
if (annotation != null) {
reference.set(new MethodInvokingReleaseStrategy(bean, method));
reference.set(new MethodInvokingReleaseStrategy(bean, method));
}
}
});
return reference.get();
return reference.get();
}
private MethodInvokingCorrelationStrategy getCorrelationStrategy(final Object bean) {
final AtomicReference<MethodInvokingCorrelationStrategy> reference = new AtomicReference<MethodInvokingCorrelationStrategy>();
ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, CorrelationStrategy.class);
if (annotation != null) {
reference.set(new MethodInvokingCorrelationStrategy(bean, method));
}
}
});
return reference.get();
}
private MethodInvokingCorrelationStrategy getCorrelationStrategy(final Object bean) {
final AtomicReference<MethodInvokingCorrelationStrategy> reference = new AtomicReference<MethodInvokingCorrelationStrategy>();
ReflectionUtils.doWithMethods(bean.getClass(), new ReflectionUtils.MethodCallback() {
@Override
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, CorrelationStrategy.class);
if (annotation != null) {
reference.set(new MethodInvokingCorrelationStrategy(bean, method));
}
}
});
return reference.get();
}
}

12
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java
View File

@@ -23,8 +23,6 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
@@ -55,7 +53,6 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
private volatile MessageHandler theOneHandler;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
/**
* Set the maximum subscribers allowed by this dispatcher.
* @param maxSubscribers The maximum number of subscribers allowed.
@@ -74,15 +71,6 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
return handlers.asUnmodifiableSet();
}
protected MessageBuilderFactory getMessageBuilderFactory() {
return messageBuilderFactory;
}
public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
Assert.notNull(messageBuilderFactory, "'messageBuilderFactory' cannot be null");
this.messageBuilderFactory = messageBuilderFactory;
}
/**
* Add the handler to the internal Set.
*

18
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java
View File

@@ -19,7 +19,13 @@ package org.springframework.integration.dispatcher;
import java.util.Collection;
import java.util.concurrent.Executor;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
@@ -39,7 +45,7 @@ import org.springframework.messaging.MessagingException;
* @author Gary Russell
* @author Oleg Zhurakousky
*/
public class BroadcastingDispatcher extends AbstractDispatcher {
public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFactoryAware {
private final boolean requireSubscribers;
@@ -51,6 +57,9 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
private volatile int minSubscribers;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
public BroadcastingDispatcher() {
this(null, false);
}
@@ -103,6 +112,11 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
this.minSubscribers = minSubscribers;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(beanFactory);
}
@Override
public boolean dispatch(Message<?> message) {
int dispatched = 0;
@@ -113,7 +127,7 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
}
int sequenceSize = handlers.size();
for (final MessageHandler handler : handlers) {
final Message<?> messageToSend = (!this.applySequence) ? message : this.getMessageBuilderFactory().fromMessage(message)
final Message<?> messageToSend = (!this.applySequence) ? message : this.messageBuilderFactory.fromMessage(message)
.pushSequenceDetails(message.getHeaders().getId(), sequenceNumber++, sequenceSize).build();
if (this.executor != null) {
this.executor.execute(new Runnable() {

1
spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java
View File

@@ -180,6 +180,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
if (this.requestMapper instanceof DefaultRequestMapper) {
((DefaultRequestMapper) this.requestMapper).setMessageBuilderFactory(this.getMessageBuilderFactory());
}
this.messageConverter.setBeanFactory(this.getBeanFactory());
}
this.initialized = true;
}

11
spring-integration-core/src/main/java/org/springframework/integration/support/converter/SimpleMessageConverter.java
View File

@@ -16,6 +16,10 @@
package org.springframework.integration.support.converter;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.mapping.InboundMessageMapper;
import org.springframework.integration.mapping.OutboundMessageMapper;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
@@ -31,7 +35,7 @@ import org.springframework.messaging.converter.MessageConverter;
* @since 2.0
*/
@SuppressWarnings({"unchecked", "rawtypes"})
public class SimpleMessageConverter implements MessageConverter {
public class SimpleMessageConverter implements MessageConverter, BeanFactoryAware {
private volatile InboundMessageMapper inboundMessageMapper;
@@ -67,8 +71,9 @@ public class SimpleMessageConverter implements MessageConverter {
this.outboundMessageMapper = (outboundMessageMapper != null) ? outboundMessageMapper : new DefaultOutboundMessageMapper();
}
public final void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
this.messageBuilderFactory = messageBuilderFactory;
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(beanFactory);
}
@Override

14
spring-integration-core/src/main/java/org/springframework/integration/support/json/AbstractJacksonJsonMessageParser.java
View File

@@ -18,10 +18,13 @@ package org.springframework.integration.support.json;
import java.lang.reflect.Type;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
* Base {@link JsonInboundMessageMapper.JsonMessageParser} implementation for Jackson processors.
@@ -30,7 +33,8 @@ import org.springframework.util.Assert;
* @since 3.0
*
*/
abstract class AbstractJacksonJsonMessageParser<P> implements JsonInboundMessageMapper.JsonMessageParser<P> {
abstract class AbstractJacksonJsonMessageParser<P> implements JsonInboundMessageMapper.JsonMessageParser<P>,
BeanFactoryAware {
private final JsonObjectMapper<?, P> objectMapper;
@@ -42,9 +46,9 @@ abstract class AbstractJacksonJsonMessageParser<P> implements JsonInboundMessage
this.objectMapper = objectMapper;
}
public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
Assert.notNull(messageBuilderFactory, "'messageBuilderFactory' cannot be null");
this.messageBuilderFactory = messageBuilderFactory;
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(beanFactory);
}
protected MessageBuilderFactory getMessageBuilderFactory() {

10
spring-integration-core/src/main/java/org/springframework/integration/util/AbstractExpressionEvaluator.java
View File

@@ -32,7 +32,6 @@ import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.util.Assert;
/**
* @author Mark Fisher
@@ -60,6 +59,7 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, I
/**
* Specify a BeanFactory in order to enable resolution via <code>@beanName</code> in the expression.
*/
@Override
public void setBeanFactory(final BeanFactory beanFactory) {
if (beanFactory != null) {
this.beanFactory = beanFactory;
@@ -67,6 +67,7 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, I
if (this.evaluationContext != null && this.evaluationContext.getBeanResolver() == null) {
this.evaluationContext.setBeanResolver(new BeanFactoryResolver(beanFactory));
}
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(beanFactory);
}
}
@@ -76,11 +77,6 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, I
}
}
public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
Assert.notNull(messageBuilderFactory, "'messageBuilderFactory' cannot be null");
this.messageBuilderFactory = messageBuilderFactory;
}
protected MessageBuilderFactory getMessageBuilderFactory() {
if (this.messageBuilderFactory == null) {
this.messageBuilderFactory = new DefaultMessageBuilderFactory();
@@ -92,7 +88,7 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, I
public void afterPropertiesSet() throws Exception {
getEvaluationContext();
if (this.messageBuilderFactory == null) {
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(beanFactory);
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(this.beanFactory);
}
}

21
spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -37,7 +37,6 @@ import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.aggregator.AggregatingMessageHandler;
import org.springframework.integration.aggregator.CorrelationStrategy;
@@ -46,6 +45,7 @@ import org.springframework.integration.aggregator.ExpressionEvaluatingReleaseStr
import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor;
import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
import org.springframework.integration.aggregator.ReleaseStrategy;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
@@ -53,6 +53,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
@@ -63,6 +64,7 @@ import org.springframework.messaging.SubscribableChannel;
* @author Oleg Zhurakousky
* @author Artem Bilan
* @author Gunnar Hillert
* @author Gary Russell
*/
public class AggregatorParserTests {
@@ -88,6 +90,10 @@ public class AggregatorParserTests {
.size());
Message<?> aggregatedMessage = aggregatorBean.getAggregatedMessages().get("id1");
assertEquals("The aggregated message payload is not correct", "123456789", aggregatedMessage.getPayload());
Object mbf = context.getBean(IntegrationContextUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
Object handler = context.getBean("aggregatorWithReference.handler");
assertSame(mbf, TestUtils.getPropertyValue(handler, "outputProcessor.messageBuilderFactory"));
assertSame(mbf, TestUtils.getPropertyValue(handler, "outputProcessor.processor.messageBuilderFactory"));
}
@Test
@@ -96,6 +102,7 @@ public class AggregatorParserTests {
SubscribableChannel outputChannel = (SubscribableChannel) context.getBean("aggregatorWithExpressionsOutput");
final AtomicReference<Message<?>> aggregatedMessage = new AtomicReference<Message<?>>();
outputChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessageRejectedException, MessageHandlingException,
MessageDeliveryException {
aggregatedMessage.set(message);
@@ -110,6 +117,10 @@ public class AggregatorParserTests {
}
assertEquals("The aggregated message payload is not correct", "[123]", aggregatedMessage.get().getPayload()
.toString());
Object mbf = context.getBean(IntegrationContextUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
Object handler = context.getBean("aggregatorWithExpressions.handler");
assertSame(mbf, TestUtils.getPropertyValue(handler, "outputProcessor.messageBuilderFactory"));
assertSame(mbf, TestUtils.getPropertyValue(handler, "outputProcessor.processor.messageBuilderFactory"));
}
@Test
@@ -158,6 +169,10 @@ public class AggregatorParserTests {
PollableChannel outputChannel = (PollableChannel) context.getBean("outputChannel");
Message<?> response = outputChannel.receive(10);
Assert.assertEquals(6l, response.getPayload());
Object mbf = context.getBean(IntegrationContextUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
Object handler = context.getBean("aggregatorWithReferenceAndMethod.handler");
assertSame(mbf, TestUtils.getPropertyValue(handler, "outputProcessor.messageBuilderFactory"));
assertSame(mbf, TestUtils.getPropertyValue(handler, "outputProcessor.processor.messageBuilderFactory"));
}
@Test(expected = BeanCreationException.class)
@@ -230,6 +245,8 @@ public class AggregatorParserTests {
EventDrivenConsumer aggregatorConsumer = (EventDrivenConsumer) context.getBean("aggregatorWithExpressionsAndPojoAggregator");
AggregatingMessageHandler aggregatingMessageHandler = (AggregatingMessageHandler) TestUtils.getPropertyValue(aggregatorConsumer, "handler");
MethodInvokingMessageGroupProcessor messageGroupProcessor = (MethodInvokingMessageGroupProcessor) TestUtils.getPropertyValue(aggregatingMessageHandler, "outputProcessor");
Object mbf = context.getBean(IntegrationContextUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
assertSame(mbf, TestUtils.getPropertyValue(messageGroupProcessor, "messageBuilderFactory"));
Object messageGroupProcessorTargetObject = TestUtils.getPropertyValue(messageGroupProcessor, "processor.delegate.targetObject");
assertSame(context.getBean("aggregatorBean"), messageGroupProcessorTargetObject);
ReleaseStrategy releaseStrategy = (ReleaseStrategy) TestUtils.getPropertyValue(aggregatingMessageHandler, "releaseStrategy");

14
spring-integration-core/src/test/java/org/springframework/integration/config/PublishSubscribeChannelParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.Executor;
@@ -29,12 +30,14 @@ import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.util.ErrorHandler;
/**
* @author Mark Fisher
* @author Gary Russell
*/
public class PublishSubscribeChannelParserTests {
@@ -51,6 +54,9 @@ public class PublishSubscribeChannelParserTests {
assertNull(dispatcherAccessor.getPropertyValue("executor"));
assertFalse((Boolean) dispatcherAccessor.getPropertyValue("ignoreFailures"));
assertFalse((Boolean) dispatcherAccessor.getPropertyValue("applySequence"));
Object mbf = context.getBean(IntegrationContextUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
assertSame(mbf, dispatcherAccessor.getPropertyValue("messageBuilderFactory"));
context.close();
}
@Test
@@ -63,6 +69,7 @@ public class PublishSubscribeChannelParserTests {
BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
accessor.getPropertyValue("dispatcher");
assertTrue((Boolean) new DirectFieldAccessor(dispatcher).getPropertyValue("ignoreFailures"));
context.close();
}
@Test
@@ -75,6 +82,7 @@ public class PublishSubscribeChannelParserTests {
BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
accessor.getPropertyValue("dispatcher");
assertTrue((Boolean) new DirectFieldAccessor(dispatcher).getPropertyValue("applySequence"));
context.close();
}
@Test
@@ -93,6 +101,7 @@ public class PublishSubscribeChannelParserTests {
DirectFieldAccessor executorAccessor = new DirectFieldAccessor(executor);
Executor innerExecutor = (Executor) executorAccessor.getPropertyValue("executor");
assertEquals(context.getBean("pool"), innerExecutor);
context.close();
}
@Test
@@ -112,6 +121,7 @@ public class PublishSubscribeChannelParserTests {
DirectFieldAccessor executorAccessor = new DirectFieldAccessor(executor);
Executor innerExecutor = (Executor) executorAccessor.getPropertyValue("executor");
assertEquals(context.getBean("pool"), innerExecutor);
context.close();
}
@Test
@@ -131,6 +141,7 @@ public class PublishSubscribeChannelParserTests {
DirectFieldAccessor executorAccessor = new DirectFieldAccessor(executor);
Executor innerExecutor = (Executor) executorAccessor.getPropertyValue("executor");
assertEquals(context.getBean("pool"), innerExecutor);
context.close();
}
@Test
@@ -143,6 +154,7 @@ public class PublishSubscribeChannelParserTests {
ErrorHandler errorHandler = (ErrorHandler) accessor.getPropertyValue("errorHandler");
assertNotNull(errorHandler);
assertEquals(context.getBean("testErrorHandler"), errorHandler);
context.close();
}
}

51
spring-integration-core/src/test/java/org/springframework/integration/gateway/GatewayInterfaceTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -21,6 +21,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
@@ -28,6 +29,7 @@ import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import org.hamcrest.Matchers;
@@ -35,12 +37,14 @@ import org.junit.Test;
import org.mockito.Mockito;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.annotation.Header;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
@@ -55,7 +59,7 @@ public class GatewayInterfaceTests {
@Test
public void testWithServiceSuperclassAnnotatedMethod() throws Exception {
ApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
DirectChannel channel = ac.getBean("requestChannelFoo", DirectChannel.class);
final Method fooMethod = Foo.class.getMethod("foo", String.class);
final AtomicBoolean called = new AtomicBoolean();
@@ -76,11 +80,16 @@ public class GatewayInterfaceTests {
Bar bar = ac.getBean(Bar.class);
bar.foo("hello");
assertTrue(called.get());
Map<?,?> gateways = TestUtils.getPropertyValue(ac.getBean("&sampleGateway"), "gatewayMap", Map.class);
Object mbf = ac.getBean(IntegrationContextUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
assertSame(mbf, TestUtils.getPropertyValue(gateways.values().iterator().next(),
"messageConverter.messageBuilderFactory"));
ac.close();
}
@Test
public void testWithServiceSuperclassAnnotatedMethodOverridePE() throws Exception {
ApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests2-context.xml", this.getClass());
ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests2-context.xml", this.getClass());
DirectChannel channel = ac.getBean("requestChannelFoo", DirectChannel.class);
final Method fooMethod = Foo.class.getMethod("foo", String.class);
final AtomicBoolean called = new AtomicBoolean();
@@ -101,22 +110,24 @@ public class GatewayInterfaceTests {
Bar bar = ac.getBean(Bar.class);
bar.foo("hello");
assertTrue(called.get());
ac.close();
}
@Test
public void testWithServiceAnnotatedMethod() {
ApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
DirectChannel channel = ac.getBean("requestChannelBar", DirectChannel.class);
MessageHandler handler = mock(MessageHandler.class);
channel.subscribe(handler);
Bar bar = ac.getBean(Bar.class);
bar.bar("hello");
verify(handler, times(1)).handleMessage(Mockito.any(Message.class));
ac.close();
}
@Test
public void testWithServiceSuperclassUnAnnotatedMethod() throws Exception {
ApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
DirectChannel channel = ac.getBean("requestChannelBaz", DirectChannel.class);
final Method bazMethod = Foo.class.getMethod("baz", String.class);
final AtomicBoolean called = new AtomicBoolean();
@@ -137,11 +148,12 @@ public class GatewayInterfaceTests {
Bar bar = ac.getBean(Bar.class);
bar.baz("hello");
assertTrue(called.get());
ac.close();
}
@Test
public void testWithServiceUnAnnotatedMethodGlobalHeaderDoesntOverride() throws Exception {
ApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
DirectChannel channel = ac.getBean("requestChannelBaz", DirectChannel.class);
final Method quxMethod = Bar.class.getMethod("qux", String.class, String.class);
final AtomicBoolean called = new AtomicBoolean();
@@ -162,55 +174,60 @@ public class GatewayInterfaceTests {
Bar bar = ac.getBean(Bar.class);
bar.qux("hello", "arg1");
assertTrue(called.get());
ac.close();
}
@Test
public void testWithServiceCastAsSuperclassAnnotatedMethod() {
ApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
DirectChannel channel = ac.getBean("requestChannelFoo", DirectChannel.class);
MessageHandler handler = mock(MessageHandler.class);
channel.subscribe(handler);
Foo foo = ac.getBean(Foo.class);
foo.foo("hello");
verify(handler, times(1)).handleMessage(Mockito.any(Message.class));
ac.close();
}
@Test
public void testWithServiceCastAsSuperclassUnAnnotatedMethod() {
ApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
DirectChannel channel = ac.getBean("requestChannelBaz", DirectChannel.class);
MessageHandler handler = mock(MessageHandler.class);
channel.subscribe(handler);
Foo foo = ac.getBean(Foo.class);
foo.baz("hello");
verify(handler, times(1)).handleMessage(Mockito.any(Message.class));
ac.close();
}
@Test
public void testWithServiceHashcode() throws Exception {
ApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
DirectChannel channel = ac.getBean("requestChannelBaz", DirectChannel.class);
MessageHandler handler = mock(MessageHandler.class);
channel.subscribe(handler);
Bar bar = ac.getBean(Bar.class);
assertEquals(bar.hashCode(), ac.getBean(Bar.class).hashCode());
verify(handler, times(0)).handleMessage(Mockito.any(Message.class));
ac.close();
}
@Test
public void testWithServiceToString() {
ApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
DirectChannel channel = ac.getBean("requestChannelBaz", DirectChannel.class);
MessageHandler handler = mock(MessageHandler.class);
channel.subscribe(handler);
Bar bar = ac.getBean(Bar.class);
bar.toString();
verify(handler, times(0)).handleMessage(Mockito.any(Message.class));
ac.close();
}
@Test
public void testWithServiceEquals() throws Exception {
ApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
DirectChannel channel = ac.getBean("requestChannelBaz", DirectChannel.class);
MessageHandler handler = mock(MessageHandler.class);
channel.subscribe(handler);
@@ -225,17 +242,19 @@ public class GatewayInterfaceTests {
fb.afterPropertiesSet();
assertFalse(bar.equals(fb.getObject()));
verify(handler, times(0)).handleMessage(Mockito.any(Message.class));
ac.close();
}
@Test
public void testWithServiceGetClass() {
ApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
DirectChannel channel = ac.getBean("requestChannelBaz", DirectChannel.class);
MessageHandler handler = mock(MessageHandler.class);
channel.subscribe(handler);
Bar bar = ac.getBean(Bar.class);
bar.getClass();
verify(handler, times(0)).handleMessage(Mockito.any(Message.class));
ac.close();
}
@Test(expected=IllegalArgumentException.class)
@@ -245,7 +264,7 @@ public class GatewayInterfaceTests {
@Test
public void testWithCustomMapper() {
ApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
DirectChannel channel = ac.getBean("requestChannelBaz", DirectChannel.class);
final AtomicBoolean called = new AtomicBoolean();
MessageHandler handler = new MessageHandler() {
@@ -260,11 +279,12 @@ public class GatewayInterfaceTests {
Baz baz = ac.getBean(Baz.class);
baz.baz("hello");
assertTrue(called.get());
ac.close();
}
@Test
public void testLateReply() throws Exception {
ApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
ConfigurableApplicationContext ac = new ClassPathXmlApplicationContext("GatewayInterfaceTests-context.xml", this.getClass());
Bar baz = ac.getBean(Bar.class);
String reply = baz.lateReply("hello");
assertNull(reply);
@@ -273,6 +293,7 @@ public class GatewayInterfaceTests {
assertNotNull(receive);
MessagingException messagingException = (MessagingException) receive.getPayload();
assertThat(messagingException.getMessage(), Matchers.startsWith("Reply message received but the receiving thread has exited due to a timeout"));
ac.close();
}

9
spring-integration-file/src/main/java/org/springframework/integration/file/config/FileToByteArrayTransformerParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,16 +16,19 @@
package org.springframework.integration.file.config;
import org.springframework.integration.file.transformer.FileToByteArrayTransformer;
/**
* Parser for the &lt;file-to-bytes-transformer&gt; element.
*
*
* @author Mark Fisher
* @author Gary Russell
*/
public class FileToByteArrayTransformerParser extends AbstractFilePayloadTransformerParser {
@Override
protected String getTransformerClassName() {
return "org.springframework.integration.file.transformer.FileToByteArrayTransformer";
return FileToByteArrayTransformer.class.getName();
}
}

8
spring-integration-file/src/main/java/org/springframework/integration/file/config/FileToStringTransformerParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -21,17 +21,19 @@ import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.file.transformer.FileToStringTransformer;
/**
* Parser for the &lt;file-to-string-transformer&gt; element.
*
*
* @author Mark Fisher
* @author Gary Russell
*/
public class FileToStringTransformerParser extends AbstractFilePayloadTransformerParser {
@Override
protected String getTransformerClassName() {
return "org.springframework.integration.file.transformer.FileToStringTransformer";
return FileToStringTransformer.class.getName();
}
@Override

12
spring-integration-file/src/main/java/org/springframework/integration/file/transformer/AbstractFilePayloadTransformer.java
View File

@@ -21,6 +21,10 @@ import java.io.File;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
@@ -34,7 +38,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public abstract class AbstractFilePayloadTransformer<T> implements Transformer {
public abstract class AbstractFilePayloadTransformer<T> implements Transformer, BeanFactoryAware {
private final Log logger = LogFactory.getLog(this.getClass());
@@ -52,9 +56,9 @@ public abstract class AbstractFilePayloadTransformer<T> implements Transformer {
this.deleteFiles = deleteFiles;
}
public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
Assert.notNull(messageBuilderFactory, "'messageBuilderFactory' cannot be null");
this.messageBuilderFactory = messageBuilderFactory;
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(beanFactory);
}
@Override

14
spring-integration-file/src/test/java/org/springframework/integration/file/config/FileToStringTransformerParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2009 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,21 +16,25 @@
package org.springframework.integration.file.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.file.transformer.FileToStringTransformer;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.transformer.MessageTransformingHandler;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.junit.Assert.assertEquals;
/**
* @author Mark Fisher
* @author Gary Russell
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@@ -40,6 +44,9 @@ public class FileToStringTransformerParserTests {
@Qualifier("transformer")
EventDrivenConsumer endpoint;
@Autowired
MessageBuilderFactory messageBuilderFactory;
@Test
public void checkDeleteFilesValue() {
DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(endpoint);
@@ -50,6 +57,7 @@ public class FileToStringTransformerParserTests {
handlerAccessor.getPropertyValue("transformer");
DirectFieldAccessor transformerAccessor = new DirectFieldAccessor(transformer);
assertEquals(Boolean.TRUE, transformerAccessor.getPropertyValue("deleteFiles"));
assertSame(this.messageBuilderFactory, transformerAccessor.getPropertyValue("messageBuilderFactory"));
}
}

3
spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java
View File

@@ -27,7 +27,6 @@ import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.DefaultTcpNetSSLSocketFactorySupport;
@@ -134,7 +133,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
@Override
protected AbstractConnectionFactory createInstance() throws Exception {
if (!this.mapperSet) {
mapper.setMessageBuilderFactory(IntegrationContextUtils.getMessageBuilderFactory(this.beanFactory));
mapper.setBeanFactory(this.beanFactory);
}
if (this.usingNio) {
if ("server".equals(this.type)) {

11
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java
View File

@@ -93,6 +93,8 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
private volatile TcpMessageMapper mapper = new TcpMessageMapper();
private volatile boolean mapperSet;
private volatile boolean singleUse;
private volatile boolean active;
@@ -359,6 +361,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
*/
public void setMapper(TcpMessageMapper mapper) {
this.mapper = mapper;
this.mapperSet = true;
}
/**
@@ -409,6 +412,14 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
this.nioHarvestInterval = nioHarvestInterval;
}
@Override
protected void onInit() throws Exception {
super.onInit();
if (!this.mapperSet) {
this.mapper.setBeanFactory(this.getBeanFactory());
}
}
@Override
public void start() {
if (logger.isInfoEnabled()) {

12
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java
View File

@@ -21,6 +21,10 @@ import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.mapping.InboundMessageMapper;
import org.springframework.integration.mapping.OutboundMessageMapper;
@@ -46,7 +50,8 @@ import org.springframework.messaging.MessageHandlingException;
*/
public class TcpMessageMapper implements
InboundMessageMapper<TcpConnection>,
OutboundMessageMapper<Object> {
OutboundMessageMapper<Object>,
BeanFactoryAware {
protected final Log logger = LogFactory.getLog(this.getClass());
@@ -81,8 +86,9 @@ public class TcpMessageMapper implements
this.applySequence = applySequence;
}
public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
this.messageBuilderFactory = messageBuilderFactory;
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(beanFactory);
}
protected MessageBuilderFactory getMessageBuilderFactory() {

16
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java
View File

@@ -23,6 +23,10 @@ import java.util.UUID;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.ip.util.RegexUtils;
import org.springframework.integration.mapping.InboundMessageMapper;
@@ -57,7 +61,8 @@ import org.springframework.util.Assert;
* @author Dave Syer
* @since 2.0
*/
public class DatagramPacketMessageMapper implements InboundMessageMapper<DatagramPacket>, OutboundMessageMapper<DatagramPacket> {
public class DatagramPacketMessageMapper implements InboundMessageMapper<DatagramPacket>, OutboundMessageMapper<DatagramPacket>,
BeanFactoryAware {
private volatile String charset = "UTF-8";
@@ -77,10 +82,6 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
RegexUtils.escapeRegexSpecials(MessageHeaders.ID) +
"=" + "([^;]*);");
public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
this.messageBuilderFactory = messageBuilderFactory;
}
public void setCharset(String charset) {
this.charset = charset;
}
@@ -104,6 +105,11 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
this.lookupHost = lookupHost;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(beanFactory);
}
/**
* Raw byte[] from message, possibly with a length field up front.
*/

2
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java
View File

@@ -75,7 +75,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
@Override
protected void onInit() {
super.onInit();
this.mapper.setMessageBuilderFactory(this.getMessageBuilderFactory());
this.mapper.setBeanFactory(this.getBeanFactory());
}
@Override

113
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/UnicastSendingMessageHandler.java
View File

@@ -32,11 +32,11 @@ import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
@@ -298,6 +298,65 @@ public class UnicastSendingMessageHandler extends
return this.socket;
}
/**
* @see java.net.Socket#setReceiveBufferSize(int)
* @see DatagramSocket#setReceiveBufferSize(int)
*/
@Override
public void setSoReceiveBufferSize(int size) {
this.soReceiveBufferSize = size;
}
@Override
public void setLocalAddress(String localAddress) {
this.localAddress = localAddress;
}
public void setTaskExecutor(Executor taskExecutor) {
Assert.notNull(taskExecutor, "'taskExecutor' cannot be null");
this.taskExecutor = taskExecutor;
this.taskExecutorSet = true;
}
/**
* @param ackCounter the ackCounter to set
*/
public void setAckCounter(int ackCounter) {
this.ackCounter = ackCounter;
}
@Override
public String getComponentType(){
return "ip:udp-outbound-channel-adapter";
}
/**
* @return the acknowledge
*/
public boolean isAcknowledge() {
return acknowledge;
}
/**
* @return the ackPort
*/
public int getAckPort() {
return ackPort;
}
/**
* @return the soReceiveBufferSize
*/
public int getSoReceiveBufferSize() {
return soReceiveBufferSize;
}
@Override
protected void onInit() throws Exception {
super.onInit();
this.mapper.setBeanFactory(this.getBeanFactory());
}
protected void setSocketAttributes(DatagramSocket socket) throws SocketException {
if (this.getSoTimeout() >= 0) {
socket.setSoTimeout(this.getSoTimeout());
@@ -365,56 +424,4 @@ public class UnicastSendingMessageHandler extends
}
}
/**
* @see java.net.Socket#setReceiveBufferSize(int)
* @see DatagramSocket#setReceiveBufferSize(int)
*/
@Override
public void setSoReceiveBufferSize(int size) {
this.soReceiveBufferSize = size;
}
@Override
public void setLocalAddress(String localAddress) {
this.localAddress = localAddress;
}
public void setTaskExecutor(Executor taskExecutor) {
Assert.notNull(taskExecutor, "'taskExecutor' cannot be null");
this.taskExecutor = taskExecutor;
this.taskExecutorSet = true;
}
/**
* @param ackCounter the ackCounter to set
*/
public void setAckCounter(int ackCounter) {
this.ackCounter = ackCounter;
}
@Override
public String getComponentType(){
return "ip:udp-outbound-channel-adapter";
}
/**
* @return the acknowledge
*/
public boolean isAcknowledge() {
return acknowledge;
}
/**
* @return the ackPort
*/
public int getAckPort() {
return ackPort;
}
/**
* @return the soReceiveBufferSize
*/
public int getSoReceiveBufferSize() {
return soReceiveBufferSize;
}
}

19
spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -42,6 +42,7 @@ import org.springframework.core.serializer.Serializer;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
@@ -70,12 +71,12 @@ import org.springframework.integration.ip.udp.MulticastReceivingChannelAdapter;
import org.springframework.integration.ip.udp.MulticastSendingMessageHandler;
import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter;
import org.springframework.integration.ip.udp.UnicastSendingMessageHandler;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -274,6 +275,9 @@ public class ParserUnitTests {
@Autowired
QueueChannel eventChannel;
@Autowired
MessageBuilderFactory messageBuilderFactory;
private static volatile int adviceCalled;
@Test
@@ -295,6 +299,7 @@ public class ParserUnitTests {
assertFalse((Boolean)mapperAccessor.getPropertyValue("lookupHost"));
assertFalse(TestUtils.getPropertyValue(udpIn, "autoStartup", Boolean.class));
assertEquals(1234, dfa.getPropertyValue("phase"));
assertSame(this.messageBuilderFactory, mapperAccessor.getPropertyValue("messageBuilderFactory"));
}
@Test
@@ -313,12 +318,14 @@ public class ParserUnitTests {
DatagramPacketMessageMapper mapper = (DatagramPacketMessageMapper) dfa.getPropertyValue("mapper");
DirectFieldAccessor mapperAccessor = new DirectFieldAccessor(mapper);
assertTrue((Boolean)mapperAccessor.getPropertyValue("lookupHost"));
assertSame(this.messageBuilderFactory, mapperAccessor.getPropertyValue("messageBuilderFactory"));
}
@Test
public void testInTcp() {
DirectFieldAccessor dfa = new DirectFieldAccessor(tcpIn);
assertSame(cfS1, dfa.getPropertyValue("serverConnectionFactory"));
assertSame(this.messageBuilderFactory, TestUtils.getPropertyValue(cfS1, "mapper.messageBuilderFactory"));
assertEquals("testInTcp",tcpIn.getComponentName());
assertEquals("ip:tcp-inbound-channel-adapter", tcpIn.getComponentType());
assertEquals(errorChannel, dfa.getPropertyValue("errorChannel"));
@@ -345,8 +352,8 @@ public class ParserUnitTests {
@Test
public void testInTcpNioSSLDefaultConfig() {
assertFalse(cfS1Nio.isLookupHost());
assertTrue((Boolean) TestUtils.getPropertyValue(
TestUtils.getPropertyValue(cfS1Nio, "mapper"), "applySequence"));
assertTrue((Boolean) TestUtils.getPropertyValue(cfS1Nio, "mapper.applySequence"));
assertSame(this.messageBuilderFactory, TestUtils.getPropertyValue(cfS1Nio, "mapper.messageBuilderFactory"));
Object connectionSupport = TestUtils.getPropertyValue(cfS1Nio, "tcpNioConnectionSupport");
assertTrue(connectionSupport instanceof DefaultTcpNioSSLConnectionSupport);
assertNotNull(TestUtils.getPropertyValue(connectionSupport, "sslContext"));
@@ -374,6 +381,7 @@ public class ParserUnitTests {
assertEquals(23, dfa.getPropertyValue("order"));
assertEquals("testOutUdp",udpOut.getComponentName());
assertEquals("ip:udp-outbound-channel-adapter", udpOut.getComponentType());
assertSame(this.messageBuilderFactory, TestUtils.getPropertyValue(mapper, "messageBuilderFactory"));
}
@Test
@@ -395,6 +403,7 @@ public class ParserUnitTests {
assertEquals(54, dfa.getPropertyValue("soTimeout"));
assertEquals(55, dfa.getPropertyValue("timeToLive"));
assertEquals(12, dfa.getPropertyValue("order"));
assertSame(this.messageBuilderFactory, TestUtils.getPropertyValue(mapper, "messageBuilderFactory"));
}
@Test

4
spring-integration-jms/src/main/java/org/springframework/integration/jms/SubscribableJmsChannel.java
View File

@@ -102,7 +102,9 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
private void configureDispatcher(boolean isPubSub) {
if (isPubSub) {
this.dispatcher = new BroadcastingDispatcher(true);
BroadcastingDispatcher broadcastingDispatcher = new BroadcastingDispatcher(true);
broadcastingDispatcher.setBeanFactory(this.getBeanFactory());
this.dispatcher = broadcastingDispatcher;
}
else {
UnicastingDispatcher unicastingDispatcher = new UnicastingDispatcher();

3
spring-integration-jms/src/main/java/org/springframework/integration/jms/config/JmsChannelFactoryBean.java
View File

@@ -337,6 +337,9 @@ public class JmsChannelFactoryBean extends AbstractFactoryBean<AbstractJmsChanne
this.container = this.createContainer();
SubscribableJmsChannel subscribableJmsChannel = new SubscribableJmsChannel(this.container, this.jmsTemplate);
subscribableJmsChannel.setMaxSubscribers(this.maxSubscribers);
if (this.getBeanFactory() != null) {
subscribableJmsChannel.setBeanFactory(this.getBeanFactory());
}
this.channel = subscribableJmsChannel;
}
else {

9
spring-integration-jms/src/test/java/org/springframework/integration/jms/config/JmsChannelParserTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.jms.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import java.util.List;
@@ -36,6 +37,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.integration.jms.PollableJmsChannel;
import org.springframework.integration.jms.SubscribableJmsChannel;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.listener.AbstractMessageListenerContainer;
@@ -116,6 +118,9 @@ public class JmsChannelParserTests {
@Autowired
private AbstractApplicationContext context;
@Autowired
private MessageBuilderFactory messageBuilderFactory;
@After
public void closeContext() {
@@ -172,8 +177,12 @@ public class JmsChannelParserTests {
AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) accessor.getPropertyValue("container");
assertEquals(topic, jmsTemplate.getDefaultDestination());
assertEquals(topic, container.getDestination());
assertSame(this.messageBuilderFactory, TestUtils.getPropertyValue(channel, "dispatcher.messageBuilderFactory"));
assertSame(this.messageBuilderFactory,
TestUtils.getPropertyValue(channel, "container.messageListener.messageBuilderFactory"));
}
@Test
public void topicNameChannel() {
assertEquals(SubscribableJmsChannel.class, topicNameChannel.getClass());

9
spring-integration-redis/src/main/java/org/springframework/integration/redis/channel/SubscribableRedisChannel.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.redis.channel;
import java.util.concurrent.Executor;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
@@ -33,7 +34,6 @@ import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.AbstractDispatcher;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.support.converter.SimpleMessageConverter;
@@ -60,7 +60,7 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements
private final RedisTemplate redisTemplate;
private final String topicName;
private final AbstractDispatcher dispatcher = new BroadcastingDispatcher(true);
private final BroadcastingDispatcher dispatcher = new BroadcastingDispatcher(true);
private volatile Integer maxSubscribers;
@@ -84,6 +84,7 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements
this.taskExecutor = taskExecutor;
}
@Override
public void setMessageConverter(MessageConverter messageConverter) {
Assert.notNull(messageConverter, "'messageConverter' must not be null");
this.messageConverter = messageConverter;
@@ -135,6 +136,9 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements
if (this.messageConverter == null){
this.messageConverter = new SimpleMessageConverter();
}
if (this.messageConverter instanceof BeanFactoryAware) {
((BeanFactoryAware) this.messageConverter).setBeanFactory(this.getBeanFactory());
}
this.container.setConnectionFactory(connectionFactory);
if (!(this.taskExecutor instanceof ErrorHandlingTaskExecutor)) {
ErrorHandler errorHandler = new MessagePublishingErrorHandler(
@@ -147,6 +151,7 @@ public class SubscribableRedisChannel extends AbstractMessageChannel implements
adapter.afterPropertiesSet();
this.container.addMessageListener(adapter, new ChannelTopic(topicName));
this.container.afterPropertiesSet();
this.dispatcher.setBeanFactory(this.getBeanFactory());
this.initialized = true;
}

6
spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/RedisInboundChannelAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2007-2013 the original author or authors
* Copyright 2007-2014 the original author or authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -19,6 +19,7 @@ package org.springframework.integration.redis.inbound;
import java.util.ArrayList;
import java.util.List;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.PatternTopic;
@@ -95,6 +96,9 @@ public class RedisInboundChannelAdapter extends MessageProducerSupport {
}
Assert.state(hasTopics || hasPatterns, "at least one topic or topic pattern is required for subscription.");
if (this.messageConverter instanceof BeanFactoryAware) {
((BeanFactoryAware) this.messageConverter).setBeanFactory(this.getBeanFactory());
}
MessageListenerDelegate delegate = new MessageListenerDelegate();
MessageListenerAdapter adapter = new MessageListenerAdapter(delegate);
adapter.setSerializer(this.serializer);

4
spring-integration-redis/src/main/java/org/springframework/integration/redis/outbound/RedisPublishingMessageHandler.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.redis.outbound;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
@@ -94,6 +95,9 @@ public class RedisPublishingMessageHandler extends AbstractMessageHandler implem
@Override
protected void onInit() throws Exception {
Assert.notNull(topicExpression, "'topicExpression' must not be null.");
if (this.messageConverter instanceof BeanFactoryAware) {
((BeanFactoryAware) this.messageConverter).setBeanFactory(this.getBeanFactory());
}
}
@Override

25
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisChannelParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
package org.springframework.integration.redis.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.CountDownLatch;
@@ -28,14 +29,15 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.redis.rules.RedisAvailable;
import org.springframework.integration.redis.rules.RedisAvailableTests;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;
/**
* @author Oleg Zhurakousky
@@ -59,9 +61,11 @@ public class RedisChannelParserTests extends RedisAvailableTests{
assertEquals(Integer.MAX_VALUE, TestUtils.getPropertyValue(
TestUtils.getPropertyValue(redisChannel, "dispatcher"), "maxSubscribers", Integer.class).intValue());
redisChannel = context.getBean("redisChannelWithSubLimit", SubscribableChannel.class);
assertEquals(1, TestUtils.getPropertyValue(
TestUtils.getPropertyValue(redisChannel, "dispatcher"), "maxSubscribers", Integer.class).intValue());
context.stop();
assertEquals(1, TestUtils.getPropertyValue(redisChannel, "dispatcher.maxSubscribers", Integer.class).intValue());
Object mbf = context.getBean(IntegrationContextUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
assertSame(mbf, TestUtils.getPropertyValue(redisChannel, "dispatcher.messageBuilderFactory"));
assertSame(mbf, TestUtils.getPropertyValue(redisChannel, "messageBuilderFactory"));
context.close();
}
@Test
@@ -76,6 +80,7 @@ public class RedisChannelParserTests extends RedisAvailableTests{
final CountDownLatch latch = new CountDownLatch(1);
redisChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
assertEquals(m.getPayload(), message.getPayload());
latch.countDown();
@@ -84,7 +89,7 @@ public class RedisChannelParserTests extends RedisAvailableTests{
redisChannel.send(m);
assertTrue(latch.await(5, TimeUnit.SECONDS));
context.stop();
context.close();
}
}

5
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisInboundChannelAdapterParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -33,6 +33,7 @@ import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.redis.inbound.RedisInboundChannelAdapter;
import org.springframework.integration.redis.rules.RedisAvailable;
import org.springframework.integration.redis.rules.RedisAvailableTests;
@@ -77,6 +78,8 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests {
Object bean = context.getBean("withoutSerializer.adapter");
assertNotNull(bean);
assertNull(TestUtils.getPropertyValue(bean, "serializer"));
Object mbf = context.getBean(IntegrationContextUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
assertSame(mbf, TestUtils.getPropertyValue(bean, "messageConverter.messageBuilderFactory"));
}
@Test

4
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisOutboundChannelAdapterParserTests.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.redis.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -28,6 +29,7 @@ import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.expression.Expression;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.redis.inbound.RedisInboundChannelAdapter;
import org.springframework.integration.redis.outbound.RedisPublishingMessageHandler;
@@ -73,6 +75,8 @@ public class RedisOutboundChannelAdapterParserTests extends RedisAvailableTests
Object converterBean = context.getBean("testConverter");
assertEquals(converterBean, accessor.getPropertyValue("messageConverter"));
assertEquals(context.getBean("serializer"), accessor.getPropertyValue("serializer"));
Object mbf = context.getBean(IntegrationContextUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
assertSame(mbf, TestUtils.getPropertyValue(handler, "messageConverter.messageBuilderFactory"));
}
@Test

5
spring-integration-redis/src/test/java/org/springframework/integration/redis/config/RedisStoreInboundChannelAdapterParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,6 +22,7 @@ import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.parsing.BeanDefinitionParsingException;
import org.springframework.context.ApplicationContext;
@@ -70,7 +71,7 @@ public class RedisStoreInboundChannelAdapterParserTests {
@Test(expected=BeanDefinitionParsingException.class)
public void testTemplateAndCfMutualExclusivity(){
new ClassPathXmlApplicationContext("inbound-template-cf-fail.xml", this.getClass());
new ClassPathXmlApplicationContext("inbound-template-cf-fail.xml", this.getClass()).close();
}
}