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

INT-3661: Fix the eager BF access from BPPs (P I)

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

Previously there were a lot of noise from the `PostProcessorRegistrationDelegate$BeanPostProcessorChecker` for early access for beans.
That may produce some side-effects when some of `BeanFactoryPostProcessor`s won't adjust those beans.

The issue is based on two facts:
1. Loading beans from `BPP`, e.g. `IntegrationEvaluationContextAwareBeanPostProcessor` (or `ChannelSecurityInterceptorBeanPostProcessor` - https://jira.spring.io/browse/INT-3663)
2. Loading beans from `setBeanFactory()/setApplicationContext()` container methods

* Move all code from `setBeanFactory()` with access to the `BeanFactory` (e.g. `this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);`)
to some other lazy-load methods like `getMessageBuilderFactory()`
* Fix parser tests to remove `messageBuilderFactory` tests since there is no activity for target components to lazy-load them
* Polish some test according the new lazy-load logic
* Rework `IntegrationEvaluationContextAwareBeanPostProcessor` to the `SmartInitializingSingleton` and make it `Ordered`
* Populate `beanFactory` for the internal instance of `connectionFactory` in the `TcpSyslogReceivingChannelAdapter`
* Populate `beanFactory` for the internal `UnicastReceivingChannelAdapter` in the `UdpSyslogReceivingChannelAdapter`
* Add `log.info` that `UdpSyslogReceivingChannelAdapter` overrides `outputChannel` for the provided `UnicastReceivingChannelAdapter`
* Change the internal `MessageChannel` in the `UdpSyslogReceivingChannelAdapter` to the `FixedSubscriberChannel` for better performance

* Fix `AbstractExpressionEvaluator`
* Add JavaDocs for the `IntegrationEvaluationContextAware`

Fix `MongoDbMessageStoreClaimCheckIntegrationTests`

Addressing PR comments
This commit is contained in:
Artem Bilan
2015-02-27 17:06:47 +02:00
committed by Gary Russell
parent 134c7c870b
commit 2bde14b742
45 changed files with 546 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -57,7 +57,6 @@ public class AmqpChannelParserTests {
TestUtils.getPropertyValue(channel, "dispatcher"), "maxSubscribers", Integer.class).intValue());
channel = context.getBean("pubSub", MessageChannel.class);
Object mbf = context.getBean(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
assertSame(mbf, TestUtils.getPropertyValue(channel, "dispatcher.messageBuilderFactory"));
assertSame(mbf, TestUtils.getPropertyValue(channel, "container.messageListener.messageBuilderFactory"));
assertTrue(TestUtils.getPropertyValue(channel, "container.missingQueuesFatal", Boolean.class));
assertFalse(TestUtils.getPropertyValue(channel, "container.transactional", Boolean.class));

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

@@ -43,6 +43,7 @@ import org.springframework.util.Assert;
* @author Mark Fisher
* @author Dave Syer
* @author Gary Russell
* @author Artem Bilan
* @since 2.0
*/
public abstract class AbstractAggregatingMessageGroupProcessor implements MessageGroupProcessor,
@@ -52,9 +53,23 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean messageBuilderFactorySet;
private BeanFactory beanFactory;
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);
this.beanFactory = beanFactory;
}
protected MessageBuilderFactory getMessageBuilderFactory() {
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
return this.messageBuilderFactory;
}
@Override
@@ -65,10 +80,10 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
Object payload = this.aggregatePayloads(group, headers);
AbstractIntegrationMessageBuilder<?> builder;
if (payload instanceof Message<?>) {
builder = this.messageBuilderFactory.fromMessage((Message<?>) payload).copyHeadersIfAbsent(headers);
builder = getMessageBuilderFactory().fromMessage((Message<?>) payload).copyHeadersIfAbsent(headers);
}
else {
builder = this.messageBuilderFactory.withPayload(payload).copyHeadersIfAbsent(headers);
builder = getMessageBuilderFactory().withPayload(payload).copyHeadersIfAbsent(headers);
}
return builder.popSequenceDetails().build();
@@ -89,7 +104,8 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
for (Entry<String, Object> entry : message.getHeaders().entrySet()) {
String key = entry.getKey();
if (MessageHeaders.ID.equals(key) || MessageHeaders.TIMESTAMP.equals(key)
|| IntegrationMessageHeaderAccessor.SEQUENCE_SIZE.equals(key) || IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER.equals(key)) {
|| IntegrationMessageHeaderAccessor.SEQUENCE_SIZE.equals(key)
|| IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER.equals(key)) {
continue;
}
Object value = entry.getValue();

12
spring-integration-core/src/main/java/org/springframework/integration/aggregator/MethodInvokingMessageListProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2015 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,14 +22,16 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Map;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.messaging.Message;
import org.springframework.integration.util.AbstractExpressionEvaluator;
import org.springframework.integration.util.MessagingMethodInvokerHelper;
/**
* A MessageListProcessor implementation that invokes a method on a target POJO.
*
*
* @author Dave Syer
* @author Artem Bilan
* @since 2.0
*/
public class MethodInvokingMessageListProcessor<T> extends AbstractExpressionEvaluator {
@@ -57,6 +59,12 @@ public class MethodInvokingMessageListProcessor<T> extends AbstractExpressionEva
delegate = new MessagingMethodInvokerHelper<T>(targetObject, annotationType, Object.class, true);
}
@Override
public void setBeanFactory(BeanFactory beanFactory) {
super.setBeanFactory(beanFactory);
this.delegate.setBeanFactory(beanFactory);
}
public String toString() {
return delegate.toString();
}

24
spring-integration-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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,7 +33,6 @@ import org.springframework.expression.EvaluationException;
import org.springframework.expression.Expression;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.ParseException;
import org.springframework.expression.spel.SpelParserConfiguration;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.core.MessagingTemplate;
@@ -63,9 +62,9 @@ public class MessagePublishingInterceptor implements MethodInterceptor, BeanFact
private final MessagingTemplate messagingTemplate = new MessagingTemplate();
private volatile PublisherMetadataSource metadataSource;
private final ExpressionParser parser = new SpelExpressionParser();
private final ExpressionParser parser = new SpelExpressionParser(new SpelParserConfiguration(true, true));
private volatile PublisherMetadataSource metadataSource;
private volatile DestinationResolver<MessageChannel> channelResolver;
@@ -75,6 +74,8 @@ public class MessagePublishingInterceptor implements MethodInterceptor, BeanFact
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean messageBuilderFactorySet;
private volatile String defaultChannelName;
public MessagePublishingInterceptor(PublisherMetadataSource metadataSource) {
@@ -114,7 +115,16 @@ public class MessagePublishingInterceptor implements MethodInterceptor, BeanFact
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
this.messagingTemplate.setBeanFactory(beanFactory);
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);
}
protected MessageBuilderFactory getMessageBuilderFactory() {
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
return this.messageBuilderFactory;
}
@Override
@@ -164,8 +174,8 @@ public class MessagePublishingInterceptor implements MethodInterceptor, BeanFact
Object result = expression.getValue(context);
if (result != null) {
AbstractIntegrationMessageBuilder<?> builder = (result instanceof Message<?>)
? this.messageBuilderFactory.fromMessage((Message<?>) result)
: this.messageBuilderFactory.withPayload(result);
? getMessageBuilderFactory().fromMessage((Message<?>) result)
: getMessageBuilderFactory().withPayload(result);
Map<String, Object> headers = this.evaluateHeaders(method, context);
if (headers != null) {
builder.copyHeaders(headers);

11
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java
View File

@@ -119,10 +119,9 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
}
@Override
public final void setBeanFactory(BeanFactory beanFactory) {
public void setBeanFactory(BeanFactory beanFactory) {
Assert.notNull(beanFactory, "'beanFactory' must not be null");
this.beanFactory = beanFactory;
this.integrationProperties = IntegrationContextUtils.getIntegrationProperties(this.beanFactory);
}
@Override
@@ -143,9 +142,15 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
@Override
public final void afterPropertiesSet() {
this.integrationProperties = IntegrationContextUtils.getIntegrationProperties(this.beanFactory);
try {
if (this.messageBuilderFactory == null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
else {
this.messageBuilderFactory = new DefaultMessageBuilderFactory();
}
}
this.onInit();
}

35
spring-integration-core/src/main/java/org/springframework/integration/core/MessagingTemplate.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.core;
import java.util.Properties;
@@ -22,6 +23,7 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.GenericMessagingTemplate;
@@ -35,6 +37,10 @@ import org.springframework.messaging.core.GenericMessagingTemplate;
*/
public class MessagingTemplate extends GenericMessagingTemplate {
private BeanFactory beanFactory;
private volatile boolean throwExceptionOnLateReplySet;
/**
* Create a MessagingTemplate with no default channel. Note, that one
* may be provided by invoking {@link #setDefaultChannel(MessageChannel)}.
@@ -55,10 +61,14 @@ public class MessagingTemplate extends GenericMessagingTemplate {
*/
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
super.setDestinationResolver(new BeanFactoryChannelResolver(beanFactory));
Properties integrationProperties = IntegrationContextUtils.getIntegrationProperties(beanFactory);
Boolean throwExceptionOnLateReply = Boolean.valueOf(integrationProperties.getProperty(IntegrationProperties.THROW_EXCEPTION_ON_LATE_REPLY));
this.setThrowExceptionOnLateReply(throwExceptionOnLateReply);
}
@Override
public void setThrowExceptionOnLateReply(boolean throwExceptionOnLateReply) {
super.setThrowExceptionOnLateReply(throwExceptionOnLateReply);
this.throwExceptionOnLateReplySet = true;
}
/**
@@ -70,4 +80,21 @@ public class MessagingTemplate extends GenericMessagingTemplate {
super.setDefaultDestination(channel);
}
@Override
public Message<?> sendAndReceive(MessageChannel destination, Message<?> requestMessage) {
if (!this.throwExceptionOnLateReplySet) {
synchronized (this) {
if (!this.throwExceptionOnLateReplySet) {
Properties integrationProperties =
IntegrationContextUtils.getIntegrationProperties(this.beanFactory);
Boolean throwExceptionOnLateReply = Boolean.valueOf(integrationProperties
.getProperty(IntegrationProperties.THROW_EXCEPTION_ON_LATE_REPLY));
super.setThrowExceptionOnLateReply(throwExceptionOnLateReply);
this.throwExceptionOnLateReplySet = true;
}
}
}
return super.sendAndReceive(destination, requestMessage);
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -44,6 +44,7 @@ import org.springframework.messaging.MessagingException;
* @author Iwein Fuld
* @author Gary Russell
* @author Oleg Zhurakousky
* @author Artem Bilan
*/
public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFactoryAware {
@@ -59,6 +60,10 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean messageBuilderFactorySet;
private BeanFactory beanFactory;
public BroadcastingDispatcher() {
this(null, false);
@@ -114,7 +119,17 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);
this.beanFactory = beanFactory;
}
protected MessageBuilderFactory getMessageBuilderFactory() {
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
return this.messageBuilderFactory;
}
@Override
@@ -127,7 +142,7 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa
}
int sequenceSize = handlers.size();
for (final MessageHandler handler : handlers) {
final Message<?> messageToSend = (!this.applySequence) ? message : this.messageBuilderFactory.fromMessage(message)
final Message<?> messageToSend = (!this.applySequence) ? message : getMessageBuilderFactory().fromMessage(message)
.pushSequenceDetails(message.getHeaders().getId(), sequenceNumber++, sequenceSize).build();
if (this.executor != null) {
this.executor.execute(new Runnable() {
@@ -174,5 +189,4 @@ public class BroadcastingDispatcher extends AbstractDispatcher implements BeanFa
}
}
}

13
spring-integration-core/src/main/java/org/springframework/integration/expression/IntegrationEvaluationContextAware.java
View File

@@ -19,8 +19,21 @@ package org.springframework.integration.expression;
import org.springframework.expression.EvaluationContext;
/**
* Interface to be implemented by beans that wish to be aware of their
* owning integration {@link EvaluationContext}, which is the result of
* {@link org.springframework.integration.config.IntegrationEvaluationContextFactoryBean}
* <p>
* The {@link #setIntegrationEvaluationContext} is invoked from
* the {@link IntegrationEvaluationContextAwareBeanPostProcessor#afterSingletonsInstantiated()},
* not during standard {@code postProcessBefore(After)Initialization} to avoid any
* {@code BeanFactory} early access during integration {@link EvaluationContext} retrieval.
* Therefore, if it is necessary to use {@link EvaluationContext} in the {@code afterPropertiesSet()},
* the {@code IntegrationContextUtils.getEvaluationContext(this.beanFactory)} should be used instead
* of this interface implementation.
*
* @author Artem Bilan
* @since 3.0
* @see IntegrationEvaluationContextAwareBeanPostProcessor
*/
public interface IntegrationEvaluationContextAware {

30
spring-integration-core/src/main/java/org/springframework/integration/expression/IntegrationEvaluationContextAwareBeanPostProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013 the original author or authors.
* Copyright 2013-2015 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,10 +16,15 @@
package org.springframework.integration.expression;
import java.util.ArrayList;
import java.util.List;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.core.Ordered;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.context.IntegrationContextUtils;
@@ -28,7 +33,11 @@ import org.springframework.integration.context.IntegrationContextUtils;
* @author Gary Russell
* @since 3.0
*/
public class IntegrationEvaluationContextAwareBeanPostProcessor implements BeanPostProcessor, BeanFactoryAware {
public class IntegrationEvaluationContextAwareBeanPostProcessor
implements BeanPostProcessor, Ordered, BeanFactoryAware, SmartInitializingSingleton {
private final List<IntegrationEvaluationContextAware> evaluationContextAwares =
new ArrayList<IntegrationEvaluationContextAware>();
private volatile BeanFactory beanFactory;
@@ -40,8 +49,7 @@ public class IntegrationEvaluationContextAwareBeanPostProcessor implements BeanP
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
if (bean instanceof IntegrationEvaluationContextAware) {
StandardEvaluationContext evaluationContext = IntegrationContextUtils.getEvaluationContext(this.beanFactory);
((IntegrationEvaluationContextAware) bean).setIntegrationEvaluationContext(evaluationContext);
this.evaluationContextAwares.add((IntegrationEvaluationContextAware) bean);
}
return bean;
}
@@ -50,4 +58,18 @@ public class IntegrationEvaluationContextAwareBeanPostProcessor implements BeanP
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
@Override
public void afterSingletonsInstantiated() {
StandardEvaluationContext evaluationContext = IntegrationContextUtils.getEvaluationContext(this.beanFactory);
for (IntegrationEvaluationContextAware evaluationContextAware : this.evaluationContextAwares) {
evaluationContextAware.setIntegrationEvaluationContext(evaluationContext);
}
}
@Override
public int getOrder() {
return LOWEST_PRECEDENCE;
}
}

7
spring-integration-core/src/main/java/org/springframework/integration/handler/MethodInvokingMessageHandler.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.handler;
import java.lang.reflect.Method;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.Lifecycle;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
@@ -47,6 +48,12 @@ public class MethodInvokingMessageHandler extends AbstractMessageHandler impleme
processor = new MethodInvokingMessageProcessor<Object>(object, methodName);
}
@Override
public void setBeanFactory(BeanFactory beanFactory) {
super.setBeanFactory(beanFactory);
this.processor.setBeanFactory(beanFactory);
}
public void setComponentType(String componentType) {
this.componentType = componentType;
}

22
spring-integration-core/src/main/java/org/springframework/integration/handler/advice/IdempotentReceiverInterceptor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014 the original author or authors.
* Copyright 2014-2015 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.
@@ -71,7 +71,11 @@ public class IdempotentReceiverInterceptor implements MethodInterceptor, BeanFac
private volatile boolean throwExceptionOnRejection;
private MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean messageBuilderFactorySet;
private BeanFactory beanFactory;
public IdempotentReceiverInterceptor(MessageSelector messageSelector) {
Assert.notNull(messageSelector);
@@ -124,7 +128,17 @@ public class IdempotentReceiverInterceptor implements MethodInterceptor, BeanFac
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);
this.beanFactory = beanFactory;
}
protected MessageBuilderFactory getMessageBuilderFactory() {
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
return this.messageBuilderFactory;
}
@Override
@@ -161,7 +175,7 @@ public class IdempotentReceiverInterceptor implements MethodInterceptor, BeanFac
}
if (!discarded) {
arguments[0] = this.messageBuilderFactory.fromMessage(message)
arguments[0] = getMessageBuilderFactory().fromMessage(message)
.setHeader(IntegrationMessageHeaderAccessor.DUPLICATE_MESSAGE, true).build();
}
else {

14
spring-integration-core/src/main/java/org/springframework/integration/store/AbstractMessageGroupStore.java
View File

@@ -33,9 +33,9 @@ import org.springframework.messaging.Message;
* @author Dave Syer
* @author Oleg Zhurakousky
* @author Gary Russell
* @author Artem Bilan
*
* @since 2.0
*
*/
@ManagedResource
@IntegrationManagedResource
@@ -52,6 +52,8 @@ public abstract class AbstractMessageGroupStore implements MessageGroupStore, It
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean messageBuilderFactorySet;
public AbstractMessageGroupStore() {
super();
}
@@ -59,11 +61,17 @@ public abstract class AbstractMessageGroupStore implements MessageGroupStore, It
@Override
public final void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
protected MessageBuilderFactory getMessageBuilderFactory() {
return messageBuilderFactory;
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
return this.messageBuilderFactory;
}
/**

43
spring-integration-core/src/main/java/org/springframework/integration/support/channel/BeanFactoryChannelResolver.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 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.
@@ -38,6 +38,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
*
* @see org.springframework.beans.factory.BeanFactory
*/
@@ -49,6 +50,8 @@ public class BeanFactoryChannelResolver implements DestinationResolver<MessageCh
private volatile HeaderChannelRegistry replyChannelRegistry;
private volatile boolean initialized;
/**
* Create a new instance of the {@link BeanFactoryChannelResolver} class.
* <p>The BeanFactory to access must be set via <code>setBeanFactory</code>.
@@ -66,30 +69,16 @@ public class BeanFactoryChannelResolver implements DestinationResolver<MessageCh
* replaced by the {@link BeanFactory} that creates it (c.f. the
* {@link BeanFactoryAware} contract). So only use this constructor if you
* are instantiating this object explicitly rather than defining a bean.
*
* @param beanFactory the bean factory to be used to lookup {@link MessageChannel}s.
*/
public BeanFactoryChannelResolver(BeanFactory beanFactory) {
Assert.notNull(beanFactory, "BeanFactory must not be null");
this.lookupHeaderChannelRegistry(beanFactory);
this.beanFactory = beanFactory;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
this.lookupHeaderChannelRegistry(beanFactory);
}
private void lookupHeaderChannelRegistry(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
try {
this.replyChannelRegistry = beanFactory.getBean(
IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME,
HeaderChannelRegistry.class);
}
catch (Exception e) {
logger.debug("No HeaderChannelRegistry found");
}
}
@Override
@@ -99,15 +88,31 @@ public class BeanFactoryChannelResolver implements DestinationResolver<MessageCh
return this.beanFactory.getBean(name, MessageChannel.class);
}
catch (BeansException e) {
if (!this.initialized) {
synchronized (this) {
if (!this.initialized) {
try {
this.replyChannelRegistry = this.beanFactory.getBean(
IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME,
HeaderChannelRegistry.class);
}
catch (Exception ex) {
logger.debug("No HeaderChannelRegistry found");
}
this.initialized = true;
}
}
}
if (this.replyChannelRegistry != null) {
MessageChannel channel = this.replyChannelRegistry.channelNameToChannel(name);
if (channel != null) {
return channel;
}
}
throw new DestinationResolutionException(
"failed to look up MessageChannel with name '" + name + "' in the BeanFactory"
+ (this.replyChannelRegistry == null ? " (and there is no HeaderChannelRegistry present)." : "."), e);
throw new DestinationResolutionException("failed to look up MessageChannel with name '" + name
+ "' in the BeanFactory"
+ (this.replyChannelRegistry == null ? " (and there is no HeaderChannelRegistry present)." : "."),
e);
}
}

21
spring-integration-core/src/main/java/org/springframework/integration/support/converter/MapMessageConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.support.converter;
import java.util.Arrays;
@@ -32,7 +33,9 @@ import org.springframework.util.Assert;
/**
* Converts to/from a Map with 2 keys ('headers' and 'payload').
*
* @author Gary Russell
* @author Artem Bilan
* @since 3.0
*
*/
@@ -46,15 +49,22 @@ public class MapMessageConverter implements MessageConverter, BeanFactoryAware {
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean messageBuilderFactorySet;
@Override
public final void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
protected MessageBuilderFactory getMessageBuilderFactory() {
return messageBuilderFactory;
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
return this.messageBuilderFactory;
}
/**
@@ -87,7 +97,7 @@ public class MapMessageConverter implements MessageConverter, BeanFactoryAware {
Map<String, ?> map = (Map<String, ?>) object;
Object payload = map.get("payload");
Assert.notNull(payload, "'payload' entry cannot be null");
AbstractIntegrationMessageBuilder<?> messageBuilder = this.messageBuilderFactory.withPayload(payload);
AbstractIntegrationMessageBuilder<?> messageBuilder = getMessageBuilderFactory().withPayload(payload);
@SuppressWarnings("unchecked")
Map<String, ?> headers = (Map<String, ?>) map.get("headers");
if (headers != null) {
@@ -96,8 +106,7 @@ public class MapMessageConverter implements MessageConverter, BeanFactoryAware {
}
messageBuilder.copyHeaders(headers);
}
Message<?> convertedMessage = messageBuilder.build();
return convertedMessage;
return messageBuilder.build();
}
@Override

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -32,6 +32,7 @@ import org.springframework.messaging.converter.MessageConverter;
/**
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
* @since 2.0
*/
@SuppressWarnings({"unchecked", "rawtypes"})
@@ -43,37 +44,60 @@ public class SimpleMessageConverter implements MessageConverter, BeanFactoryAwar
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean messageBuilderFactorySet;
private BeanFactory beanFactory;
public SimpleMessageConverter() {
this(null, null);
}
public SimpleMessageConverter(InboundMessageMapper<?> inboundMessageMapper) {
this(inboundMessageMapper,
(inboundMessageMapper instanceof OutboundMessageMapper ? (OutboundMessageMapper<?>) inboundMessageMapper : null));
(inboundMessageMapper instanceof OutboundMessageMapper
? (OutboundMessageMapper<?>) inboundMessageMapper
: null));
}
public SimpleMessageConverter(OutboundMessageMapper<?> outboundMessageMapper) {
this(outboundMessageMapper instanceof InboundMessageMapper ? (InboundMessageMapper<?>) outboundMessageMapper : null,
this(outboundMessageMapper instanceof InboundMessageMapper
? (InboundMessageMapper<?>) outboundMessageMapper
: null,
outboundMessageMapper);
}
public SimpleMessageConverter(InboundMessageMapper<?> inboundMessageMapper, OutboundMessageMapper<?> outboundMessageMapper) {
public SimpleMessageConverter(InboundMessageMapper<?> inboundMessageMapper,
OutboundMessageMapper<?> outboundMessageMapper) {
this.setInboundMessageMapper(inboundMessageMapper);
this.setOutboundMessageMapper(outboundMessageMapper);
}
public void setInboundMessageMapper(InboundMessageMapper<?> inboundMessageMapper) {
this.inboundMessageMapper = (inboundMessageMapper != null) ? inboundMessageMapper : new DefaultInboundMessageMapper();
this.inboundMessageMapper = (inboundMessageMapper != null)
? inboundMessageMapper
: new DefaultInboundMessageMapper();
}
public void setOutboundMessageMapper(OutboundMessageMapper<?> outboundMessageMapper) {
this.outboundMessageMapper = (outboundMessageMapper != null) ? outboundMessageMapper : new DefaultOutboundMessageMapper();
this.outboundMessageMapper = (outboundMessageMapper != null
? outboundMessageMapper
: new DefaultOutboundMessageMapper());
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);
this.beanFactory = beanFactory;
}
protected MessageBuilderFactory getMessageBuilderFactory() {
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
return this.messageBuilderFactory;
}
@Override
@@ -107,8 +131,9 @@ public class SimpleMessageConverter implements MessageConverter, BeanFactoryAwar
if (object instanceof Message<?>) {
return (Message<?>) object;
}
return messageBuilderFactory.withPayload(object).build();
return getMessageBuilderFactory().withPayload(object).build();
}
}
@@ -118,6 +143,7 @@ public class SimpleMessageConverter implements MessageConverter, BeanFactoryAwar
public Object fromMessage(Message<?> message) throws Exception {
return (message != null) ? message.getPayload() : null;
}
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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,17 +42,27 @@ abstract class AbstractJacksonJsonMessageParser<P> implements JsonInboundMessage
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private BeanFactory beanFactory;
private volatile boolean messageBuilderFactorySet;
protected AbstractJacksonJsonMessageParser(JsonObjectMapper<?, P> objectMapper) {
this.objectMapper = objectMapper;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);
this.beanFactory = beanFactory;
}
protected MessageBuilderFactory getMessageBuilderFactory() {
return messageBuilderFactory;
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
return this.messageBuilderFactory;
}
@Override
@@ -64,7 +74,7 @@ abstract class AbstractJacksonJsonMessageParser<P> implements JsonInboundMessage
if (messageMapper.isMapToPayload()) {
Object payload = this.readPayload(parser, jsonMessage);
return this.messageBuilderFactory.withPayload(payload).build();
return getMessageBuilderFactory().withPayload(payload).build();
}
else {
return this.parseWithHeaders(parser, jsonMessage);

28
spring-integration-core/src/main/java/org/springframework/integration/transformer/AbstractMessageProcessingTransformer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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,8 +19,6 @@ package org.springframework.integration.transformer;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.context.Lifecycle;
import org.springframework.core.convert.ConversionService;
import org.springframework.integration.handler.AbstractMessageProcessor;
import org.springframework.integration.handler.MessageProcessor;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
@@ -41,25 +39,31 @@ public abstract class AbstractMessageProcessingTransformer
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean messageBuilderFactorySet;
private BeanFactory beanFactory;
protected AbstractMessageProcessingTransformer(MessageProcessor<?> messageProcessor) {
Assert.notNull(messageProcessor, "messageProcessor must not be null");
this.messageProcessor = messageProcessor;
}
protected MessageBuilderFactory getMessageBuilderFactory() {
return messageBuilderFactory;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
if (this.messageProcessor instanceof BeanFactoryAware) {
((BeanFactoryAware) this.messageProcessor).setBeanFactory(beanFactory);
}
ConversionService conversionService = IntegrationUtils.getConversionService(beanFactory);
if (conversionService != null && this.messageProcessor instanceof AbstractMessageProcessor) {
((AbstractMessageProcessor<?>) this.messageProcessor).setConversionService(conversionService);
}
protected MessageBuilderFactory getMessageBuilderFactory() {
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);
return this.messageBuilderFactory;
}
@Override
@@ -90,7 +94,7 @@ public abstract class AbstractMessageProcessingTransformer
if (result instanceof Message<?>) {
return (Message<?>) result;
}
return this.messageBuilderFactory.withPayload(result).copyHeaders(message.getHeaders()).build();
return getMessageBuilderFactory().withPayload(result).copyHeaders(message.getHeaders()).build();
}
}

13
spring-integration-core/src/main/java/org/springframework/integration/transformer/support/MessageProcessingHeaderValueMessageProcessor.java
View File

@@ -16,6 +16,9 @@
package org.springframework.integration.transformer.support;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.handler.MessageProcessor;
import org.springframework.integration.handler.MethodInvokingMessageProcessor;
import org.springframework.messaging.Message;
@@ -25,7 +28,8 @@ import org.springframework.messaging.Message;
* @author Artem Bilan
* @since 3.0
*/
public class MessageProcessingHeaderValueMessageProcessor extends AbstractHeaderValueMessageProcessor<Object> {
public class MessageProcessingHeaderValueMessageProcessor extends AbstractHeaderValueMessageProcessor<Object>
implements BeanFactoryAware {
private final MessageProcessor<?> targetProcessor;
@@ -41,6 +45,13 @@ public class MessageProcessingHeaderValueMessageProcessor extends AbstractHeader
this.targetProcessor = new MethodInvokingMessageProcessor<Object>(targetObject, method);
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
if (this.targetProcessor instanceof BeanFactoryAware) {
((BeanFactoryAware) this.targetProcessor).setBeanFactory(beanFactory);
}
}
public Object processMessage(Message<?> message) {
return this.targetProcessor.processMessage(message);
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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
@@ -54,7 +54,7 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, I
private volatile BeanFactory beanFactory;
private volatile MessageBuilderFactory messageBuilderFactory;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
/**
* Specify a BeanFactory in order to enable resolution via <code>@beanName</code> in the expression.
@@ -67,7 +67,6 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, I
if (this.evaluationContext != null && this.evaluationContext.getBeanResolver() == null) {
this.evaluationContext.setBeanResolver(new BeanFactoryResolver(beanFactory));
}
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);
}
}
@@ -82,16 +81,13 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, I
}
protected MessageBuilderFactory getMessageBuilderFactory() {
if (this.messageBuilderFactory == null) {
this.messageBuilderFactory = new DefaultMessageBuilderFactory();
}
return this.messageBuilderFactory;
}
@Override
public void afterPropertiesSet() throws Exception {
getEvaluationContext();
if (this.messageBuilderFactory == null) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
}
@@ -114,6 +110,12 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, I
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.beanFactory);
}
this.evaluationContext.setTypeConverter(this.typeConverter);
if (this.beanFactory != null) {
ConversionService conversionService = IntegrationUtils.getConversionService(beanFactory);
if (conversionService != null) {
this.typeConverter.setConversionService(conversionService);
}
}
}
return this.evaluationContext;
}

47
spring-integration-core/src/main/java/org/springframework/integration/util/MessagingMethodInvokerHelper.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -112,6 +112,14 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
private final boolean canProcessMessageList;
private Class<? extends Annotation> annotationType;
private volatile boolean initialized;
private String methodName;
private Method method;
public MessagingMethodInvokerHelper(Object targetObject, Method method, Class<?> expectedType,
boolean canProcessMessageList) {
@@ -184,6 +192,7 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
Method method, Class<?> expectedType, boolean canProcessMessageList) {
this.canProcessMessageList = canProcessMessageList;
Assert.notNull(method, "method must not be null");
this.method = method;
this.expectedType = expectedType;
this.requiresReply = expectedType != null;
if (expectedType != null) {
@@ -201,12 +210,13 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
this.handlerMethods = null;
this.handlerMessageMethods = null;
this.handlerMethodsList = null;
this.prepareEvaluationContext(this.getEvaluationContext(false), method, annotationType);
this.setDisplayString(targetObject, method);
}
private MessagingMethodInvokerHelper(Object targetObject, Class<? extends Annotation> annotationType,
String methodName, Class<?> expectedType, boolean canProcessMessageList) {
this.annotationType = annotationType;
this.methodName = methodName;
this.canProcessMessageList = canProcessMessageList;
Assert.notNull(targetObject, "targetObject must not be null");
this.expectedType = expectedType;
@@ -238,7 +248,6 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
this.handlerMethodsList.add(this.handlerMethods);
this.handlerMethodsList.add(this.handlerMessageMethods);
}
this.prepareEvaluationContext(this.getEvaluationContext(false), methodName, annotationType);
this.setDisplayString(targetObject, methodName);
}
@@ -253,26 +262,26 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
this.displayString = sb.toString() + "]";
}
private void prepareEvaluationContext(StandardEvaluationContext context, Object method,
Class<? extends Annotation> annotationType) {
private void prepareEvaluationContext() {
StandardEvaluationContext context = getEvaluationContext(false);
Class<?> targetType = AopUtils.getTargetClass(this.targetObject);
if (method instanceof Method) {
context.registerMethodFilter(targetType, new FixedMethodFilter((Method) method));
if (expectedType != null) {
if (this.method != null) {
context.registerMethodFilter(targetType, new FixedMethodFilter(this.method));
if (this.expectedType != null) {
Assert.state(context.getTypeConverter()
.canConvert(TypeDescriptor.valueOf(((Method) method).getReturnType()),
TypeDescriptor.valueOf(expectedType)),
"Cannot convert to expected type (" + expectedType + ") from " + method);
.canConvert(TypeDescriptor.valueOf((this.method).getReturnType()),
TypeDescriptor.valueOf(this.expectedType)),
"Cannot convert to expected type (" + this.expectedType + ") from " + this.method);
}
}
else if (method == null || method instanceof String) {
AnnotatedMethodFilter filter = new AnnotatedMethodFilter(annotationType, (String) method,
else {
AnnotatedMethodFilter filter = new AnnotatedMethodFilter(this.annotationType, this.methodName,
this.requiresReply);
Assert.state(canReturnExpectedType(filter, targetType, context.getTypeConverter()),
"Cannot convert to expected type (" + expectedType + ") from " + method);
"Cannot convert to expected type (" + this.expectedType + ") from " + this.method);
context.registerMethodFilter(targetType, filter);
}
context.setVariable("target", targetObject);
context.setVariable("target", this.targetObject);
}
private boolean canReturnExpectedType(AnnotatedMethodFilter filter, Class<?> targetType,
@@ -291,6 +300,14 @@ public class MessagingMethodInvokerHelper<T> extends AbstractExpressionEvaluator
}
private T processInternal(ParametersWrapper parameters) throws Exception {
if (!this.initialized) {
synchronized (this) {
if (!this.initialized) {
prepareEvaluationContext();
this.initialized = true;
}
}
}
HandlerMethod candidate = this.findHandlerMethodForParameters(parameters);
Assert.notNull(candidate, "No candidate methods found for messages.");
Expression expression = candidate.getExpression();

17
spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingReleaseStrategyTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 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,6 +16,8 @@
package org.springframework.integration.aggregator;
import static org.mockito.Mockito.mock;
import java.util.ArrayList;
import java.util.Date;
import java.util.LinkedList;
@@ -23,16 +25,18 @@ import java.util.List;
import org.junit.Assert;
import org.junit.Test;
import org.springframework.core.convert.ConversionFailedException;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.SimpleMessageGroup;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
/**
* @author Marius Bogoevici
* @author Dave Syer
* @author Artem Bilan
*/
public class MethodInvokingReleaseStrategyTests {
@@ -248,13 +252,18 @@ public class MethodInvokingReleaseStrategyTests {
@Test(expected = IllegalStateException.class)
public void testWrongReturnTypeUsingMethodObject() throws SecurityException, NoSuchMethodException {
class TestReleaseStrategy {
@SuppressWarnings("unused")
public int wrongReturnType(List<Message<?>> message) {
return 0;
}
}
new MethodInvokingReleaseStrategy(new TestReleaseStrategy(), TestReleaseStrategy.class.getMethod(
"wrongReturnType", new Class<?>[] { List.class }));
MethodInvokingReleaseStrategy wrongReturnType =
new MethodInvokingReleaseStrategy(new TestReleaseStrategy(), TestReleaseStrategy.class.getMethod(
"wrongReturnType", new Class<?>[] {List.class}));
wrongReturnType.canRelease(mock(MessageGroup.class));
}
private static MessageGroup createListOfMessages(int size) {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -94,7 +94,6 @@ public class AggregatorParserTests {
Object mbf = context.getBean(IntegrationUtils.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
@@ -121,7 +120,6 @@ public class AggregatorParserTests {
Object mbf = context.getBean(IntegrationUtils.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"));
assertTrue(TestUtils.getPropertyValue(handler, "expireGroupsUponTimeout", Boolean.class));
}
@@ -175,7 +173,6 @@ public class AggregatorParserTests {
Object mbf = context.getBean(IntegrationUtils.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)
@@ -248,8 +245,6 @@ 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(IntegrationUtils.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");
@@ -271,4 +266,5 @@ public class AggregatorParserTests {
return MessageBuilder.withPayload(payload).setCorrelationId(correlationId).setSequenceSize(sequenceSize)
.setSequenceNumber(sequenceNumber).setReplyChannel(outputChannel).build();
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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,11 +33,16 @@ import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.ErrorHandler;
/**
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
*/
public class PublishSubscribeChannelParserTests {
@@ -50,10 +55,20 @@ public class PublishSubscribeChannelParserTests {
DirectFieldAccessor accessor = new DirectFieldAccessor(channel);
BroadcastingDispatcher dispatcher = (BroadcastingDispatcher)
accessor.getPropertyValue("dispatcher");
dispatcher.setApplySequence(true);
dispatcher.addHandler(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
}
});
dispatcher.dispatch(new GenericMessage<String>("foo"));
DirectFieldAccessor dispatcherAccessor = new DirectFieldAccessor(dispatcher);
assertNull(dispatcherAccessor.getPropertyValue("executor"));
assertFalse((Boolean) dispatcherAccessor.getPropertyValue("ignoreFailures"));
assertFalse((Boolean) dispatcherAccessor.getPropertyValue("applySequence"));
assertTrue((Boolean) dispatcherAccessor.getPropertyValue("applySequence"));
Object mbf = context.getBean(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
assertSame(mbf, dispatcherAccessor.getPropertyValue("messageBuilderFactory"));
context.close();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
@@ -133,9 +134,6 @@ public class GatewayInterfaceTests {
bar.foo("hello");
assertTrue(called.get());
Map<?,?> gateways = TestUtils.getPropertyValue(ac.getBean("&sampleGateway"), "gatewayMap", Map.class);
Object mbf = ac.getBean(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
assertSame(mbf, TestUtils.getPropertyValue(gateways.values().iterator().next(),
"messageConverter.messageBuilderFactory"));
ac.close();
}

1
spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java
View File

@@ -151,7 +151,6 @@ public abstract class AbstractInboundFileSynchronizer<F> implements InboundFileS
@Override
public final void afterPropertiesSet() {
Assert.notNull(this.remoteDirectory, "remoteDirectory must not be null");
Assert.notNull(this.evaluationContext, "evaluationContext must not be null");
}
protected final List<F> filterFiles(F[] files) {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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,6 +37,7 @@ import org.springframework.util.Assert;
* Base class for transformers that convert a File payload.
*
* @author Mark Fisher
* @author Artem Bilan
*/
public abstract class AbstractFilePayloadTransformer<T> implements Transformer, BeanFactoryAware {
@@ -46,6 +47,10 @@ public abstract class AbstractFilePayloadTransformer<T> implements Transformer,
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private boolean messageBuilderFactorySet;
private volatile BeanFactory beanFactory;
/**
* Specify whether to delete the File after transformation.
* Default is <em>false</em>.
@@ -58,7 +63,17 @@ public abstract class AbstractFilePayloadTransformer<T> implements Transformer,
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);
this.beanFactory = beanFactory;
}
protected MessageBuilderFactory getMessageBuilderFactory() {
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
return this.messageBuilderFactory;
}
@Override
@@ -66,11 +81,11 @@ public abstract class AbstractFilePayloadTransformer<T> implements Transformer,
try {
Assert.notNull(message, "Message must not be null");
Object payload = message.getPayload();
Assert.notNull(payload, "Mesasge payload must not be null");
Assert.notNull(payload, "Message payload must not be null");
Assert.isInstanceOf(File.class, payload, "Message payload must be of type [java.io.File]");
File file = (File) payload;
T result = this.transformFile(file);
Message<?> transformedMessage = this.messageBuilderFactory.withPayload(result)
Message<?> transformedMessage = getMessageBuilderFactory().withPayload(result)
.copyHeaders(message.getHeaders())
.setHeaderIfAbsent(FileHeaders.ORIGINAL_FILE, file)
.setHeaderIfAbsent(FileHeaders.FILENAME, file.getName())

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

@@ -17,6 +17,7 @@
package org.springframework.integration.file.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertSame;
import org.junit.Test;
@@ -44,9 +45,6 @@ public class FileToStringTransformerParserTests {
@Qualifier("transformer")
EventDrivenConsumer endpoint;
@Autowired
MessageBuilderFactory messageBuilderFactory;
@Test
public void checkDeleteFilesValue() {
DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(endpoint);
@@ -57,7 +55,6 @@ public class FileToStringTransformerParserTests {
handlerAccessor.getPropertyValue("transformer");
DirectFieldAccessor transformerAccessor = new DirectFieldAccessor(transformer);
assertEquals(Boolean.TRUE, transformerAccessor.getPropertyValue("deleteFiles"));
assertSame(this.messageBuilderFactory, transformerAccessor.getPropertyValue("messageBuilderFactory"));
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -44,7 +44,9 @@ import org.springframework.messaging.MessageHandlingException;
* to correlate which connection to send a reply. If applySequence is set, adds
* standard correlationId/sequenceNumber headers allowing for downstream (unbounded)
* resequencing.
* *
* @author Gary Russell
* @author Artem Bilan
* @since 2.0
*
*/
@@ -63,6 +65,10 @@ public class TcpMessageMapper implements
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean messageBuilderFactorySet;
private BeanFactory beanFactory;
/**
* @param charset the charset to set
*/
@@ -88,11 +94,17 @@ public class TcpMessageMapper implements
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);
this.beanFactory = beanFactory;
}
protected MessageBuilderFactory getMessageBuilderFactory() {
return messageBuilderFactory;
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
return this.messageBuilderFactory;
}
@Override
@@ -100,7 +112,7 @@ public class TcpMessageMapper implements
Message<Object> message = null;
Object payload = connection.getPayload();
if (payload != null) {
AbstractIntegrationMessageBuilder<Object> messageBuilder = this.messageBuilderFactory.withPayload(payload);
AbstractIntegrationMessageBuilder<Object> messageBuilder = getMessageBuilderFactory().withPayload(payload);
this.addStandardHeaders(connection, messageBuilder);
this.addCustomHeaders(connection, messageBuilder);
message = messageBuilder.build();

23
spring-integration-ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.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.
@@ -59,6 +59,7 @@ import org.springframework.util.Assert;
* @author Mark Fisher
* @author Gary Russell
* @author Dave Syer
* @author Artem Bilan
* @since 2.0
*/
public class DatagramPacketMessageMapper implements InboundMessageMapper<DatagramPacket>, OutboundMessageMapper<DatagramPacket>,
@@ -76,12 +77,16 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean messageBuilderFactorySet;
private static Pattern udpHeadersPattern =
Pattern.compile(RegexUtils.escapeRegexSpecials(IpHeaders.ACK_ADDRESS) +
"=" + "([^;]*);" +
RegexUtils.escapeRegexSpecials(MessageHeaders.ID) +
"=" + "([^;]*);");
private BeanFactory beanFactory;
public void setCharset(String charset) {
this.charset = charset;
}
@@ -107,7 +112,17 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);
this.beanFactory = beanFactory;
}
protected MessageBuilderFactory getMessageBuilderFactory() {
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
return this.messageBuilderFactory;
}
/**
@@ -213,7 +228,7 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
length = length - matcher.end();
payload = new byte[length];
System.arraycopy(packet.getData(), offset + matcher.end(), payload, 0, length);
message = this.messageBuilderFactory.withPayload(payload)
message = getMessageBuilderFactory().withPayload(payload)
.setHeader(IpHeaders.ACK_ID, UUID.fromString(matcher.group(2)))
.setHeader(IpHeaders.ACK_ADDRESS, matcher.group(1))
.setHeader(IpHeaders.HOSTNAME, hostName)
@@ -230,7 +245,7 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
payload = new byte[length];
System.arraycopy(packet.getData(), offset, payload, 0, length);
if (payload.length > 0) {
message = this.messageBuilderFactory.withPayload(payload)
message = getMessageBuilderFactory().withPayload(payload)
.setHeader(IpHeaders.HOSTNAME, hostName)
.setHeader(IpHeaders.IP_ADDRESS, hostAddress)
.setHeader(IpHeaders.PORT, port)

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -275,9 +275,6 @@ public class ParserUnitTests {
@Autowired
QueueChannel eventChannel;
@Autowired
MessageBuilderFactory messageBuilderFactory;
private static volatile int adviceCalled;
@Test
@@ -299,7 +296,6 @@ 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
@@ -318,14 +314,12 @@ 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"));
@@ -353,7 +347,6 @@ public class ParserUnitTests {
public void testInTcpNioSSLDefaultConfig() {
assertFalse(cfS1Nio.isLookupHost());
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"));
@@ -381,7 +374,6 @@ 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
@@ -403,7 +395,6 @@ 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

5
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 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.
@@ -166,9 +166,6 @@ public class TcpNioConnectionReadTests {
done.countDown();
}
/**
* Test method for {@link org.springframework.integration.ip.tcp.NioSocketReader}.
*/
@SuppressWarnings("unchecked")
@Test
public void testReadCrLf() throws Exception {

7
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/store/JdbcChannelMessageStore.java
View File

@@ -157,6 +157,8 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
private boolean priorityEnabled;
private BeanFactory beanFactory;
/**
* Convenient constructor for configuration use.
*/
@@ -362,7 +364,7 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);
this.beanFactory = beanFactory;
}
/**
@@ -392,6 +394,9 @@ public class JdbcChannelMessageStore implements PriorityCapableChannelMessageSto
logger.warn("The jdbcTemplate's fetchsize is not 1. This may cause FIFO issues with Oracle databases.");
}
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.jdbcTemplate.afterPropertiesSet();
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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,8 @@
package org.springframework.integration.jms.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import java.util.List;
@@ -168,7 +170,6 @@ 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"));
}

16
spring-integration-mail/src/main/java/org/springframework/integration/mail/MailReceivingMessageSource.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -44,6 +44,7 @@ import org.springframework.util.Assert;
* @author Mark Fisher
* @author Gary Russell
* @author Oleg Zhurakousky
* @author Artem Bilan
*/
public class MailReceivingMessageSource implements MessageSource<javax.mail.Message>,
BeanFactoryAware, BeanNameAware, NamedComponent {
@@ -58,6 +59,8 @@ public class MailReceivingMessageSource implements MessageSource<javax.mail.Mess
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean messageBuilderFactorySet;
private volatile String beanName;
@@ -69,7 +72,6 @@ public class MailReceivingMessageSource implements MessageSource<javax.mail.Mess
@Override
public final void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
protected BeanFactory getBeanFactory() {
@@ -77,7 +79,13 @@ public class MailReceivingMessageSource implements MessageSource<javax.mail.Mess
}
protected MessageBuilderFactory getMessageBuilderFactory() {
return messageBuilderFactory;
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
return this.messageBuilderFactory;
}
@Override
@@ -110,7 +118,7 @@ public class MailReceivingMessageSource implements MessageSource<javax.mail.Mess
if (logger.isDebugEnabled()) {
logger.debug("received mail message [" + mailMessage + "]");
}
return this.messageBuilderFactory.withPayload(mailMessage).build();
return getMessageBuilderFactory().withPayload(mailMessage).build();
}
}
catch (Exception e) {

12
spring-integration-mail/src/main/java/org/springframework/integration/mail/transformer/AbstractMailMessageTransformer.java
View File

@@ -44,6 +44,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
*/
public abstract class AbstractMailMessageTransformer<T> implements Transformer,
BeanFactoryAware {
@@ -54,15 +55,22 @@ public abstract class AbstractMailMessageTransformer<T> implements Transformer,
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean messageBuilderFactorySet;
@Override
public final void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
protected MessageBuilderFactory getMessageBuilderFactory() {
return messageBuilderFactory;
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
return this.messageBuilderFactory;
}
@Override

22
spring-integration-mongodb/src/main/java/org/springframework/integration/mongodb/store/AbstractConfigurableMongoDbMessageStore.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014 the original author or authors.
* Copyright 2014-2015 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.
@@ -23,8 +23,6 @@ import java.util.Map;
import java.util.Set;
import java.util.UUID;
import com.mongodb.DB;
import com.mongodb.MongoException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -61,6 +59,9 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.Assert;
import com.mongodb.DB;
import com.mongodb.MongoException;
/**
* The abstract MongoDB {@link BasicMessageGroupStore} implementation to provide configuration for common options
* for implementations of this class.
@@ -111,7 +112,8 @@ public abstract class AbstractConfigurableMongoDbMessageStore implements BasicMe
this(mongoDbFactory, null, collectionName);
}
public AbstractConfigurableMongoDbMessageStore(MongoDbFactory mongoDbFactory, MappingMongoConverter mappingMongoConverter, String collectionName) {
public AbstractConfigurableMongoDbMessageStore(MongoDbFactory mongoDbFactory,
MappingMongoConverter mappingMongoConverter, String collectionName) {
Assert.notNull("'mongoDbFactory' must not be null");
Assert.hasText("'collectionName' must not be empty");
this.collectionName = collectionName;
@@ -122,7 +124,6 @@ public abstract class AbstractConfigurableMongoDbMessageStore implements BasicMe
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.applicationContext);
}
@Override
@@ -142,7 +143,9 @@ public abstract class AbstractConfigurableMongoDbMessageStore implements BasicMe
this.mongoTemplate.setApplicationContext(this.applicationContext);
}
}
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.applicationContext);
IndexOperations indexOperations = this.mongoTemplate.indexOps(this.collectionName);
indexOperations.ensureIndex(new Index(MessageDocumentFields.MESSAGE_ID, Sort.Direction.ASC));
@@ -201,13 +204,16 @@ public abstract class AbstractConfigurableMongoDbMessageStore implements BasicMe
}
}
final long createdDate = document.getCreatedTime() == 0 ? System.currentTimeMillis() : document.getCreatedTime();
final long createdDate = document.getCreatedTime() == 0
? System.currentTimeMillis()
: document.getCreatedTime();
Message<?> result = messageBuilderFactory.fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE)
.setHeader(CREATED_DATE_KEY, createdDate).build();
@SuppressWarnings("unchecked")
Map<String, Object> innerMap = (Map<String, Object>) new DirectFieldAccessor(result.getHeaders()).getPropertyValue("headers");
Map<String, Object> innerMap = (Map<String, Object>) new DirectFieldAccessor(result.getHeaders())
.getPropertyValue("headers");
// using reflection to set ID since it is immutable through MessageHeaders
innerMap.put(MessageHeaders.ID, message.getHeaders().get(MessageHeaders.ID));
innerMap.put(MessageHeaders.TIMESTAMP, message.getHeaders().get(MessageHeaders.TIMESTAMP));

15
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/store/MongoDbMessageStoreClaimCheckIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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,22 +16,25 @@
package org.springframework.integration.mongodb.store;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import java.io.Serializable;
import com.mongodb.MongoClient;
import org.junit.Test;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.data.mongodb.MongoDbFactory;
import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
import org.springframework.integration.mongodb.rules.MongoDbAvailable;
import org.springframework.integration.mongodb.rules.MongoDbAvailableTests;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.transformer.ClaimCheckInTransformer;
import org.springframework.integration.transformer.ClaimCheckOutTransformer;
import org.springframework.messaging.Message;
import com.mongodb.MongoClient;
/**
* @author Mark Fisher
* @author Artem Bilan
@@ -81,6 +84,9 @@ public class MongoDbMessageStoreClaimCheckIntegrationTests extends MongoDbAvaila
public void stringPayloadConfigurable() throws Exception {
MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new MongoClient(), "test");
ConfigurableMongoDbMessageStore messageStore = new ConfigurableMongoDbMessageStore(mongoDbFactory);
GenericApplicationContext testApplicationContext = TestUtils.createTestApplicationContext();
testApplicationContext.refresh();
messageStore.setApplicationContext(testApplicationContext);
messageStore.afterPropertiesSet();
ClaimCheckInTransformer checkin = new ClaimCheckInTransformer(messageStore);
ClaimCheckOutTransformer checkout = new ClaimCheckOutTransformer(messageStore);
@@ -98,6 +104,9 @@ public class MongoDbMessageStoreClaimCheckIntegrationTests extends MongoDbAvaila
public void objectPayloadConfigurable() throws Exception {
MongoDbFactory mongoDbFactory = new SimpleMongoDbFactory(new MongoClient(), "test");
ConfigurableMongoDbMessageStore messageStore = new ConfigurableMongoDbMessageStore(mongoDbFactory);
GenericApplicationContext testApplicationContext = TestUtils.createTestApplicationContext();
testApplicationContext.refresh();
messageStore.setApplicationContext(testApplicationContext);
messageStore.afterPropertiesSet();
ClaimCheckInTransformer checkin = new ClaimCheckInTransformer(messageStore);
ClaimCheckOutTransformer checkout = new ClaimCheckOutTransformer(messageStore);

6
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.mqtt.inbound;
import java.util.LinkedHashSet;
@@ -240,7 +241,10 @@ public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessagePro
protected void onInit() {
super.onInit();
if (this.converter == null) {
this.converter = new DefaultPahoMessageConverter();
DefaultPahoMessageConverter pahoMessageConverter = new DefaultPahoMessageConverter();
pahoMessageConverter.setBeanFactory(getBeanFactory());
this.converter = pahoMessageConverter;
}
}

19
spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/support/DefaultPahoMessageConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.mqtt.support;
import org.eclipse.paho.client.mqttv3.MqttMessage;
@@ -33,6 +34,7 @@ import org.springframework.util.Assert;
* Default implementation for mapping to/from Messages.
*
* @author Gary Russell
* @author Artem Bilan
* @since 4.0
*
*/
@@ -50,6 +52,8 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean messageBuilderFactorySet;
/**
* Construct a converter with default options (qos=0, retain=false, charset=UTF-8).
@@ -60,7 +64,7 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa
/**
* Construct a converter to create outbound messages with the supplied default qos and retain settings and
* a UTF-8 charset for converting outbound String paylaods to {@code byte[]} and inbound
* a UTF-8 charset for converting outbound String payloads to {@code byte[]} and inbound
* {@code byte[]} to String (unless {@link #setPayloadAsBytes(boolean) payloadAdBytes} is true).
* @param defaultQos the default qos.
* @param defaultRetain the default retain.
@@ -97,7 +101,6 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa
@Override
public final void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
protected BeanFactory getBeanFactory() {
@@ -105,7 +108,13 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa
}
protected MessageBuilderFactory getMessageBuilderFactory() {
return messageBuilderFactory;
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
return this.messageBuilderFactory;
}
/**
@@ -130,7 +139,7 @@ public class DefaultPahoMessageConverter implements MqttMessageConverter, BeanFa
@Override
public Message<?> toMessage(String topic, MqttMessage mqttMessage) {
try {
AbstractIntegrationMessageBuilder<Object> messageBuilder = this.messageBuilderFactory
AbstractIntegrationMessageBuilder<Object> messageBuilder = getMessageBuilderFactory()
.withPayload(mqttBytesToPayload(mqttMessage))
.setHeader(MqttHeaders.QOS, mqttMessage.getQos())
.setHeader(MqttHeaders.DUPLICATE, mqttMessage.isDuplicate())

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -62,7 +62,6 @@ public class RedisChannelParserTests extends RedisAvailableTests {
redisChannel = context.getBean("redisChannelWithSubLimit", SubscribableChannel.class);
assertEquals(1, TestUtils.getPropertyValue(redisChannel, "dispatcher.maxSubscribers", Integer.class).intValue());
Object mbf = context.getBean(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
assertSame(mbf, TestUtils.getPropertyValue(redisChannel, "dispatcher.messageBuilderFactory"));
assertSame(mbf, TestUtils.getPropertyValue(redisChannel, "messageBuilderFactory"));
context.close();
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -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.assertNotSame;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
@@ -78,8 +79,6 @@ public class RedisInboundChannelAdapterParserTests extends RedisAvailableTests {
Object bean = context.getBean("withoutSerializer.adapter");
assertNotNull(bean);
assertNull(TestUtils.getPropertyValue(bean, "serializer"));
Object mbf = context.getBean(IntegrationUtils.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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -77,8 +77,6 @@ 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(IntegrationUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
assertSame(mbf, TestUtils.getPropertyValue(handler, "messageConverter.messageBuilderFactory"));
Object endpointHandler = TestUtils.getPropertyValue(adapter, "handler");

18
spring-integration-syslog/src/main/java/org/springframework/integration/syslog/DefaultMessageConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -49,8 +49,12 @@ public class DefaultMessageConverter implements MessageConverter, BeanFactoryAwa
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private volatile boolean messageBuilderFactorySet;
private volatile boolean asMap = true;
private BeanFactory beanFactory;
/**
* Set false will leave the payload as the original complete syslog.
* @param asMap boolean flag.
@@ -65,11 +69,17 @@ public class DefaultMessageConverter implements MessageConverter, BeanFactoryAwa
@Override
public final void setBeanFactory(BeanFactory beanFactory) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(beanFactory);
this.beanFactory = beanFactory;
}
protected MessageBuilderFactory getMessageBuilderFactory() {
return messageBuilderFactory;
if (!this.messageBuilderFactorySet) {
if (this.beanFactory != null) {
this.messageBuilderFactory = IntegrationUtils.getMessageBuilderFactory(this.beanFactory);
}
this.messageBuilderFactorySet = true;
}
return this.messageBuilderFactory;
}
@Override
@@ -82,7 +92,7 @@ public class DefaultMessageConverter implements MessageConverter, BeanFactoryAwa
out.put(SyslogHeaders.PREFIX + entry.getKey(), entry.getValue());
}
}
return this.messageBuilderFactory.withPayload(this.asMap ? map : message.getPayload())
return getMessageBuilderFactory().withPayload(this.asMap ? map : message.getPayload())
.copyHeaders(out)
.build();
}

6
spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/TcpSyslogReceivingChannelAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.syslog.inbound;
@@ -28,6 +29,7 @@ import org.springframework.messaging.Message;
* TCP implementation of a syslog inbound channel adapter.
*
* @author Gary Russell
* @author Artem Bilan
* @since 3.0
*
*/
@@ -61,9 +63,11 @@ public class TcpSyslogReceivingChannelAdapter extends SyslogReceivingChannelAdap
if (this.connectionFactory == null) {
this.connectionFactory = new TcpNioServerConnectionFactory(this.getPort());
this.connectionFactory.setDeserializer(new ByteArrayLfSerializer());
this.connectionFactory.setBeanFactory(getBeanFactory());
if (this.applicationEventPublisher != null) {
this.connectionFactory.setApplicationEventPublisher(this.applicationEventPublisher);
}
this.connectionFactory.afterPropertiesSet();
}
this.connectionFactory.registerListener(this);
}

26
spring-integration-syslog/src/main/java/org/springframework/integration/syslog/inbound/UdpSyslogReceivingChannelAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 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.
@@ -13,9 +13,10 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.syslog.inbound;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
@@ -25,6 +26,7 @@ import org.springframework.messaging.MessagingException;
* UDP implementation of a syslog inbound channel adapter.
*
* @author Gary Russell
* @author Artem Bilan
* @since 3.0
*
*/
@@ -32,8 +34,11 @@ public class UdpSyslogReceivingChannelAdapter extends SyslogReceivingChannelAdap
private volatile UnicastReceivingChannelAdapter udpAdapter;
private volatile boolean udpAdapterSet;
public void setUdpAdapter(UnicastReceivingChannelAdapter udpAdpter) {
this.udpAdapter = udpAdpter;
this.udpAdapterSet = true;
}
@Override
@@ -43,18 +48,27 @@ public class UdpSyslogReceivingChannelAdapter extends SyslogReceivingChannelAdap
@Override
protected void onInit() {
super.onInit();
if (this.udpAdapter == null) {
this.udpAdapter = new UnicastReceivingChannelAdapter(this.getPort());
this.udpAdapter.setBeanFactory(getBeanFactory());
}
DirectChannel outputChannel = new DirectChannel();
outputChannel.subscribe(new MessageHandler() {
else {
logger.info("The 'UdpSyslogReceivingChannelAdapter' overrides an 'outputChannel' " +
"of the provided 'UnicastReceivingChannelAdapter' to support Syslog conversion " +
"for the incoming UDP packets");
}
this.udpAdapter.setOutputChannel(new FixedSubscriberChannel(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
convertAndSend(message);
}
});
this.udpAdapter.setOutputChannel(outputChannel);
}));
if (!this.udpAdapterSet) {
this.udpAdapter.afterPropertiesSet();
}
}
@Override