INT-3309 Pluggable MessageBuilder

This is still a work in process.

There are a bunch of TODOs in classes
that are not managed by Spring and so
need to have the MessageBuilderFactory
injected.

But I am looking for feedback on the
approach.

INT-3309 Resolve TODOs

Provide access to the MessageBuilderFactory in all classes.

INT-3309 Polishing + Tests

* Fallback to 'fromMessage()' if mutating and inbound message is not MutableMessage
* Add 'alwaysMutate' boolean to MutableMessageBuilderFactory - coerces 'fromMessage' calls to 'mutateMessage'
* Add tests

INT-3309 Polishing; PR Comments

Also add tests to parent/child contexts where the parent has the
default message builder and the child has a mutable message builder.

INT-3309 More Polish; PR Comments

Also fix removeHeader in MMB.
This commit is contained in:
Gary Russell
2014-03-06 19:09:40 +02:00
committed by Artem Bilan
parent 4c775bce9d
commit ff845b5069
115 changed files with 1763 additions and 426 deletions

View File

@@ -35,7 +35,7 @@ import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.dispatcher.AbstractDispatcher;
import org.springframework.integration.dispatcher.MessageDispatcher;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHandler;
@@ -129,7 +129,8 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
? ((RabbitTemplate) this.getAmqpTemplate()).getMessageConverter()
: new SimpleMessageConverter();
MessageListener listener = new DispatchingMessageListener(converter,
this.dispatcher, this, this.isPubSub);
this.dispatcher, this, this.isPubSub,
this.getMessageBuilderFactory());
this.container.setMessageListener(listener);
if (!this.container.isActive()) {
this.container.afterPropertiesSet();
@@ -153,14 +154,18 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
private final boolean isPubSub;
private final MessageBuilderFactory messageBuilderFactory;
private DispatchingMessageListener(MessageConverter converter,
MessageDispatcher dispatcher, AbstractSubscribableAmqpChannel channel, boolean isPubSub) {
MessageDispatcher dispatcher, AbstractSubscribableAmqpChannel channel, boolean isPubSub,
MessageBuilderFactory messageBuilderFactory) {
Assert.notNull(converter, "MessageConverter must not be null");
Assert.notNull(dispatcher, "MessageDispatcher must not be null");
this.converter = converter;
this.dispatcher = dispatcher;
this.channel = channel;
this.isPubSub = isPubSub;
this.messageBuilderFactory = messageBuilderFactory;
}
@@ -171,7 +176,7 @@ abstract class AbstractSubscribableAmqpChannel extends AbstractAmqpChannel imple
Object converted = this.converter.fromMessage(message);
if (converted != null) {
messageToSend = (converted instanceof Message<?>) ? (Message<?>) converted
: MessageBuilder.withPayload(converted).build();
: this.messageBuilderFactory.withPayload(converted).build();
this.dispatcher.dispatch(messageToSend);
}
else if (this.logger.isWarnEnabled()) {

View File

@@ -21,7 +21,6 @@ import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.util.Assert;
@@ -106,7 +105,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable
replyMessage = (Message<?>) object;
}
else {
replyMessage = MessageBuilder.withPayload(object).build();
replyMessage = this.getMessageBuilderFactory().withPayload(object).build();
}
return this.getInterceptors().postReceive(replyMessage, this) ;
}

View File

@@ -27,7 +27,6 @@ import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
/**
@@ -73,7 +72,7 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport implements
public void onMessage(Message message) {
Object payload = messageConverter.fromMessage(message);
Map<String, ?> headers = headerMapper.toHeadersFromRequest(message.getMessageProperties());
sendMessage(MessageBuilder.withPayload(payload).copyHeaders(headers).build());
sendMessage(AmqpInboundChannelAdapter.this.getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build());
}
});
this.messageListenerContainer.afterPropertiesSet();

View File

@@ -31,7 +31,6 @@ import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -40,7 +39,7 @@ import org.springframework.util.StringUtils;
* Spring Integration Messages, and sends the results to a Message Channel.
* If a reply Message is received, it will be converted and sent back to
* the AMQP 'replyTo'.
*
*
* @author Mark Fisher
* @since 2.1
*/
@@ -83,7 +82,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
Object payload = amqpMessageConverter.fromMessage(message);
Map<String, ?> headers = headerMapper.toHeadersFromRequest(message.getMessageProperties());
org.springframework.messaging.Message<?> request =
MessageBuilder.withPayload(payload).copyHeaders(headers).build();
AmqpInboundGateway.this.getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build();
final org.springframework.messaging.Message<?> reply = sendAndReceiveMessage(request);
if (reply != null) {
// TODO: fallback to a reply address property of this gateway

View File

@@ -28,14 +28,14 @@ import org.springframework.expression.Expression;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.SpelParserConfiguration;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
/**
@@ -240,9 +240,9 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
return null;
}
Object replyObject = converter.fromMessage(amqpReplyMessage);
MessageBuilder<?> builder = (replyObject instanceof Message)
? MessageBuilder.fromMessage((Message<?>) replyObject)
: MessageBuilder.withPayload(replyObject);
AbstractIntegrationMessageBuilder<?> builder = (replyObject instanceof Message)
? this.getMessageBuilderFactory().fromMessage((Message<?>) replyObject)
: this.getMessageBuilderFactory().withPayload(replyObject);
Map<String, ?> headers = this.headerMapper.toHeadersFromReply(amqpReplyMessage.getMessageProperties());
builder.copyHeadersIfAbsent(headers);
return builder.build();
@@ -253,7 +253,7 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
if (correlationData instanceof CorrelationDataWrapper) {
userCorrelationData = ((CorrelationDataWrapper) correlationData).getUserData();
}
Message<Object> confirmMessage = MessageBuilder.withPayload(userCorrelationData)
Message<Object> confirmMessage = this.getMessageBuilderFactory().withPayload(userCorrelationData)
.setHeader(AmqpHeaders.PUBLISH_CONFIRM, ack)
.build();
if (ack && this.confirmAckChannel != null) {
@@ -291,9 +291,9 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler
// safe to cast; we asserted we have a RabbitTemplate in doInit()
MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
Object returnedObject = converter.fromMessage(message);
MessageBuilder<?> builder = (returnedObject instanceof Message)
? MessageBuilder.fromMessage((Message<?>) returnedObject)
: MessageBuilder.withPayload(returnedObject);
AbstractIntegrationMessageBuilder<?> builder = (returnedObject instanceof Message)
? this.getMessageBuilderFactory().fromMessage((Message<?>) returnedObject)
: this.getMessageBuilderFactory().withPayload(returnedObject);
Map<String, ?> headers = this.headerMapper.toHeadersFromReply(message.getMessageProperties());
builder.copyHeadersIfAbsent(headers)
.setHeader(AmqpHeaders.RETURN_REPLY_CODE, replyCode)

View File

@@ -23,7 +23,9 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.Assert;
@@ -41,18 +43,25 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag
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 final Object processMessageGroup(MessageGroup group) {
Assert.notNull(group, "MessageGroup must not be null");
Map<String, Object> headers = this.aggregateHeaders(group);
Object payload = this.aggregatePayloads(group, headers);
MessageBuilder<?> builder;
AbstractIntegrationMessageBuilder<?> builder;
if (payload instanceof Message<?>) {
builder = MessageBuilder.fromMessage((Message<?>) payload).copyHeadersIfAbsent(headers);
builder = this.messageBuilderFactory.fromMessage((Message<?>) payload).copyHeadersIfAbsent(headers);
}
else {
builder = MessageBuilder.withPayload(payload).copyHeadersIfAbsent(headers);
builder = this.messageBuilderFactory.withPayload(payload).copyHeadersIfAbsent(headers);
}
return builder.popSequenceDetails().build();

View File

@@ -22,6 +22,7 @@ import java.util.Map;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
@@ -35,12 +36,15 @@ 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.context.IntegrationContextUtils;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.core.DestinationResolver;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -68,6 +72,7 @@ public class MessagePublishingInterceptor implements MethodInterceptor, BeanFact
private final ParameterNameDiscoverer parameterNameDiscoverer = new LocalVariableTableParameterNameDiscoverer();
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
public MessagePublishingInterceptor(PublisherMetadataSource metadataSource) {
Assert.notNull(metadataSource, "metadataSource must not be null");
@@ -92,6 +97,7 @@ public class MessagePublishingInterceptor implements MethodInterceptor, BeanFact
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
this.messagingTemplate.setBeanFactory(beanFactory);
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(beanFactory);
}
public final Object invoke(final MethodInvocation invocation) throws Throwable {
@@ -139,9 +145,9 @@ public class MessagePublishingInterceptor implements MethodInterceptor, BeanFact
Expression expression = this.parser.parseExpression(payloadExpressionString);
Object result = expression.getValue(context);
if (result != null) {
MessageBuilder<?> builder = (result instanceof Message<?>)
? MessageBuilder.fromMessage((Message<?>) result)
: MessageBuilder.withPayload(result);
AbstractIntegrationMessageBuilder<?> builder = (result instanceof Message<?>)
? this.messageBuilderFactory.fromMessage((Message<?>) result)
: this.messageBuilderFactory.withPayload(result);
Map<String, Object> headers = this.evaluateHeaders(method, context);
if (headers != null) {
builder.copyHeaders(headers);

View File

@@ -26,7 +26,6 @@ import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.history.MessageHistory;
import org.springframework.integration.history.TrackableComponent;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.converter.DefaultDatatypeChannelMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -243,7 +242,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
Assert.notNull(message, "message must not be null");
Assert.notNull(message.getPayload(), "message payload must not be null");
if (this.shouldTrack) {
message = MessageHistory.write(message, this);
message = MessageHistory.write(message, this, this.getMessageBuilderFactory());
}
try {
if (this.datatypes.length > 0) {
@@ -282,7 +281,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
return (Message<?>) converted;
}
else {
return MessageBuilder.withPayload(converted).copyHeaders(message.getHeaders()).build();
return this.getMessageBuilderFactory().withPayload(converted).copyHeaders(message.getHeaders()).build();
}
}
}

View File

@@ -48,6 +48,7 @@ import org.springframework.integration.config.annotation.MessagingAnnotationPost
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.IntegrationProperties;
import org.springframework.integration.expression.IntegrationEvaluationContextAwareBeanPostProcessor;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.converter.DefaultDatatypeChannelMessageConverter;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;
@@ -91,6 +92,7 @@ public class IntegrationRegistrar implements ImportBeanDefinitionRegistrar, Bean
this.registerMessagingAnnotationPostProcessors(importingClassMetadata, registry);
}
this.registerIntegrationConfigurationBeanFactoryPostProcessor(registry);
this.registerMessageBuilderFactory(registry);
}
/**
@@ -353,4 +355,24 @@ public class IntegrationRegistrar implements ImportBeanDefinitionRegistrar, Bean
}
private void registerMessageBuilderFactory(BeanDefinitionRegistry registry) {
boolean alreadyRegistered = false;
if (registry instanceof ListableBeanFactory) {
alreadyRegistered = ((ListableBeanFactory) registry)
.containsBean(IntegrationContextUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
}
else {
alreadyRegistered = registry
.isBeanNameInUse(IntegrationContextUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME);
}
if (!alreadyRegistered) {
BeanDefinitionBuilder mbfBuilder = BeanDefinitionBuilder
.genericBeanDefinition(DefaultMessageBuilderFactory.class);
registry.registerBeanDefinition(
IntegrationContextUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME,
mbfBuilder.getBeanDefinition());
}
}
}

View File

@@ -18,11 +18,16 @@ package org.springframework.integration.context;
import java.util.Properties;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.core.convert.ConversionService;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.config.IntegrationConfigUtils;
import org.springframework.integration.metadata.MetadataStore;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.MessageChannel;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
@@ -37,6 +42,8 @@ import org.springframework.util.Assert;
*/
public abstract class IntegrationContextUtils {
private static final Log logger = LogFactory.getLog(IntegrationContextUtils.class);
public static final String TASK_SCHEDULER_BEAN_NAME = "taskScheduler";
public static final String ERROR_CHANNEL_BEAN_NAME = "errorChannel";
@@ -72,7 +79,7 @@ public abstract class IntegrationContextUtils {
public static final String INTEGRATION_DATATYPE_CHANNEL_MESSAGE_CONVERTER_BEAN_NAME = "datatypeChannelMessageConverter";
public static final String INTEGRATION_FIXED_SUBSCRIBER_CHANNEL_BPP_BEAN_NAME = "fixedSubscriberChannelBeanFactoryPostProcessor";
public static final String INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME = "messageBuilderFactory";
/**
* @param beanFactory BeanFactory for lookup, must not be null.
* @return The {@link MetadataStore} bean whose name is "metadataStore".
@@ -155,4 +162,38 @@ public abstract class IntegrationContextUtils {
return properties;
}
/**
* Returns the context-wide `messageBuilderFactory` bean from the beanFactory,
* or a {@link DefaultMessageBuilderFactory} if not found or the beanFactory is null.
* @param beanFactory The bean factory.
* @return The message builder factory.
*/
public static MessageBuilderFactory getMessageBuilderFactory(BeanFactory beanFactory) {
MessageBuilderFactory messageBuilderFactory = null;
if (beanFactory != null) {
try {
messageBuilderFactory = beanFactory.getBean(
IntegrationContextUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME, MessageBuilderFactory.class);
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
logger.warn("No MessageBuilderFactory with name '"
+ IntegrationContextUtils.INTEGRATION_MESSAGE_BUILDER_FACTORY_BEAN_NAME
+ "' found: " + e.getMessage()
+ ", using default.");
}
}
}
else {
if (logger.isWarnEnabled()) {
logger.warn("No 'beanFactory' supplied; cannot find MessageBuilderFactory"
+ ", using default.");
}
}
if (messageBuilderFactory == null) {
messageBuilderFactory = new DefaultMessageBuilderFactory();
}
return messageBuilderFactory;
}
}

View File

@@ -31,6 +31,8 @@ import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
@@ -76,6 +78,8 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
private volatile ApplicationContext applicationContext;
private volatile MessageBuilderFactory messageBuilderFactory;
@Override
public final void setBeanName(String beanName) {
this.beanName = beanName;
@@ -122,6 +126,9 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
@Override
public final void afterPropertiesSet() {
try {
if (this.messageBuilderFactory == null) {
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(this.beanFactory);
}
this.onInit();
}
catch (Exception e) {
@@ -192,6 +199,18 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
return this.integrationProperties;
}
protected MessageBuilderFactory getMessageBuilderFactory() {
if (this.messageBuilderFactory == null) {
this.messageBuilderFactory = new DefaultMessageBuilderFactory();
}
return this.messageBuilderFactory;
}
public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
this.messageBuilderFactory = messageBuilderFactory;
}
/**
* @param key Integration property.
* @param tClass the class to convert a value of Integration property.

View File

@@ -23,6 +23,8 @@ 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;
@@ -53,6 +55,7 @@ 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.
@@ -71,6 +74,15 @@ 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.
*

View File

@@ -20,7 +20,6 @@ import java.util.Collection;
import java.util.concurrent.Executor;
import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
@@ -114,7 +113,7 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
}
int sequenceSize = handlers.size();
for (final MessageHandler handler : handlers) {
final Message<?> messageToSend = (!this.applySequence) ? message : MessageBuilder.fromMessage(message)
final Message<?> messageToSend = (!this.applySequence) ? message : this.getMessageBuilderFactory().fromMessage(message)
.pushSequenceDetails(message.getHeaders().getId(), sequenceNumber++, sequenceSize).build();
if (this.executor != null) {
this.executor.execute(new Runnable() {

View File

@@ -22,7 +22,7 @@ import java.util.Map;
import org.springframework.expression.Expression;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.util.AbstractExpressionEvaluator;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
@@ -58,7 +58,7 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
}
if (!CollectionUtils.isEmpty(headers)) {
// create a new Message from this one in order to apply headers
MessageBuilder<T> builder = MessageBuilder.fromMessage(message);
AbstractIntegrationMessageBuilder<T> builder = this.getMessageBuilderFactory().fromMessage(message);
builder.copyHeaders(headers);
message = builder.build();
}
@@ -71,7 +71,7 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
catch (Exception e) {
throw new MessagingException("MessageSource returned unexpected type.", e);
}
MessageBuilder<T> builder = MessageBuilder.withPayload(payload);
AbstractIntegrationMessageBuilder<T> builder = this.getMessageBuilderFactory().withPayload(payload);
if (!CollectionUtils.isEmpty(headers)) {
builder.copyHeaders(headers);
}

View File

@@ -17,13 +17,13 @@
package org.springframework.integration.endpoint;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.history.MessageHistory;
import org.springframework.integration.history.TrackableComponent;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessagingException;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.util.Assert;
@@ -45,6 +45,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
private final MessagingTemplate messagingTemplate = new MessagingTemplate();
@Override
public void setOutputChannel(MessageChannel outputChannel) {
this.outputChannel = outputChannel;
}
@@ -57,6 +58,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
this.messagingTemplate.setSendTimeout(sendTimeout);
}
@Override
public void setShouldTrack(boolean shouldTrack) {
this.shouldTrack = shouldTrack;
}
@@ -90,7 +92,7 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements
throw new MessagingException("cannot send a null message");
}
if (this.shouldTrack) {
message = MessageHistory.write(message, this);
message = MessageHistory.write(message, this, this.getMessageBuilderFactory());
}
try {
this.messagingTemplate.send(this.outputChannel, message);

View File

@@ -104,7 +104,7 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint
@Override
protected void handleMessage(Message<?> message) {
if (this.shouldTrack) {
message = MessageHistory.write(message, this);
message = MessageHistory.write(message, this, this.getMessageBuilderFactory());
}
try {
this.messagingTemplate.send(this.outputChannel, message);

View File

@@ -35,15 +35,17 @@ import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.integration.annotation.Header;
import org.springframework.integration.annotation.Headers;
import org.springframework.integration.annotation.Payload;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.mapping.InboundMessageMapper;
import org.springframework.integration.mapping.MessageMappingException;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
@@ -96,16 +98,19 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
private volatile BeanFactory beanFactory;
private final MessageBuilderFactory messageBuilderFactory;
public GatewayMethodInboundMessageMapper(Method method) {
this(method, null);
}
public GatewayMethodInboundMessageMapper(Method method, Map<String, Expression> headerExpressions) {
this(method, headerExpressions, null, null);
this(method, headerExpressions, null, null, null);
}
public GatewayMethodInboundMessageMapper(Method method, Map<String, Expression> headerExpressions,
Map<String, Expression> globalHeaderExpressions, MethodArgsMessageMapper mapper) {
Map<String, Expression> globalHeaderExpressions, MethodArgsMessageMapper mapper,
MessageBuilderFactory messageBuilderFactory) {
Assert.notNull(method, "method must not be null");
this.method = method;
this.headerExpressions = headerExpressions;
@@ -118,6 +123,12 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
else {
this.argsMapper = mapper;
}
if (messageBuilderFactory == null) {
this.messageBuilderFactory = new DefaultMessageBuilderFactory();
}
else {
this.messageBuilderFactory = messageBuilderFactory;
}
}
@@ -327,9 +338,9 @@ class GatewayMethodInboundMessageMapper implements InboundMessageMapper<Object[]
}
}
Assert.isTrue(messageOrPayload != null, "unable to determine a Message or payload parameter on method [" + method + "]");
MessageBuilder<?> builder = (messageOrPayload instanceof Message)
? MessageBuilder.fromMessage((Message<?>) messageOrPayload)
: MessageBuilder.withPayload(messageOrPayload);
AbstractIntegrationMessageBuilder<?> builder = (messageOrPayload instanceof Message)
? GatewayMethodInboundMessageMapper.this.messageBuilderFactory.fromMessage((Message<?>) messageOrPayload)
: GatewayMethodInboundMessageMapper.this.messageBuilderFactory.withPayload(messageOrPayload);
builder.copyHeadersIfAbsent(headers);
// Explicit headers in XML override any @Header annotations...
if (!CollectionUtils.isEmpty(GatewayMethodInboundMessageMapper.this.headerExpressions)) {

View File

@@ -447,7 +447,7 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint implements Trackab
}
GatewayMethodInboundMessageMapper messageMapper = new GatewayMethodInboundMessageMapper(method, headerExpressions,
this.globalMethodMetadata != null ? this.globalMethodMetadata.getHeaderExpressions() : null,
this.argsMapper);
this.argsMapper, this.getMessageBuilderFactory());
if (StringUtils.hasText(payloadExpression)) {
messageMapper.setPayloadExpression(payloadExpression);
}

View File

@@ -25,7 +25,8 @@ import org.springframework.integration.history.HistoryWritingMessagePostProcesso
import org.springframework.integration.history.TrackableComponent;
import org.springframework.integration.mapping.InboundMessageMapper;
import org.springframework.integration.mapping.OutboundMessageMapper;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.converter.SimpleMessageConverter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -173,8 +174,12 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
@Override
protected void onInit() throws Exception {
this.historyWritingPostProcessor.setTrackableComponent(this);
this.historyWritingPostProcessor.setMessageBuilderFactory(this.getMessageBuilderFactory());
if (this.getBeanFactory() != null) {
this.messagingTemplate.setBeanFactory(this.getBeanFactory());
if (this.requestMapper instanceof DefaultRequestMapper) {
((DefaultRequestMapper) this.requestMapper).setMessageBuilderFactory(this.getMessageBuilderFactory());
}
}
this.initialized = true;
}
@@ -335,12 +340,18 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement
private static class DefaultRequestMapper implements InboundMessageMapper<Object> {
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
this.messageBuilderFactory = messageBuilderFactory;
}
@Override
public Message<?> toMessage(Object object) throws Exception {
if (object instanceof Message<?>) {
return (Message<?>) object;
}
return (object != null) ? MessageBuilder.withPayload(object).build() : null;
return (object != null) ? this.messageBuilderFactory.withPayload(object).build() : null;
}
}

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.
@@ -17,14 +17,14 @@
package org.springframework.integration.handler;
import org.springframework.core.Ordered;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.context.Orderable;
import org.springframework.messaging.MessageHandler;
import org.springframework.integration.history.MessageHistory;
import org.springframework.integration.history.TrackableComponent;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
/**
@@ -35,6 +35,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Gary Russell
*/
public abstract class AbstractMessageHandler extends IntegrationObjectSupport implements MessageHandler, TrackableComponent, Orderable {
@@ -43,10 +44,12 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
private volatile int order = Ordered.LOWEST_PRECEDENCE;
@Override
public void setOrder(int order) {
this.order = order;
}
@Override
public int getOrder() {
return this.order;
}
@@ -56,10 +59,12 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
return "message-handler";
}
@Override
public void setShouldTrack(boolean shouldTrack) {
this.shouldTrack = shouldTrack;
}
@Override
public final void handleMessage(Message<?> message) {
Assert.notNull(message, "Message must not be null");
Assert.notNull(message.getPayload(), "Message payload must not be null");
@@ -68,7 +73,7 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
}
try {
if (message != null && this.shouldTrack) {
message = MessageHistory.write(message, this);
message = MessageHistory.write(message, this, this.getMessageBuilderFactory());
}
this.handleMessageInternal(message);
}

View File

@@ -25,7 +25,7 @@ import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
@@ -206,18 +206,18 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
}
private Message<?> createReplyMessage(Object reply, MessageHeaders requestHeaders) {
MessageBuilder<?> builder = null;
AbstractIntegrationMessageBuilder<?> builder = null;
if (reply instanceof Message<?>) {
if (!this.shouldCopyRequestHeaders()) {
return (Message<?>) reply;
}
builder = MessageBuilder.fromMessage((Message<?>) reply);
builder = this.getMessageBuilderFactory().fromMessage((Message<?>) reply);
}
else if (reply instanceof MessageBuilder<?>) {
builder = (MessageBuilder<?>) reply;
else if (reply instanceof AbstractIntegrationMessageBuilder) {
builder = (AbstractIntegrationMessageBuilder<?>) reply;
}
else {
builder = MessageBuilder.withPayload(reply);
builder = this.getMessageBuilderFactory().withPayload(reply);
}
if (this.shouldCopyRequestHeaders()) {
builder.copyHeadersIfAbsent(requestHeaders);
@@ -270,7 +270,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
private boolean shouldSplitReply(Iterable<?> reply) {
for (Object next : reply) {
if (next instanceof Message<?> || next instanceof MessageBuilder<?>) {
if (next instanceof Message<?> || next instanceof AbstractIntegrationMessageBuilder<?>) {
return true;
}
}

View File

@@ -38,7 +38,6 @@ import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
@@ -320,7 +319,7 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement
}
else {
messageWrapper = new DelayedMessageWrapper(message, System.currentTimeMillis());
delayedMessage = MessageBuilder.withPayload(messageWrapper).copyHeaders(message.getHeaders()).build();
delayedMessage = this.getMessageBuilderFactory().withPayload(messageWrapper).copyHeaders(message.getHeaders()).build();
this.messageStore.addMessageToGroup(this.messageGroupId, delayedMessage);
}

View File

@@ -114,4 +114,4 @@ public class ExpressionCommandMessageProcessor extends AbstractMessageProcessor<
}
}
}
}

View File

@@ -16,6 +16,8 @@
package org.springframework.integration.history;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.core.MessagePostProcessor;
import org.springframework.util.Assert;
@@ -30,6 +32,7 @@ public class HistoryWritingMessagePostProcessor implements MessagePostProcessor
private volatile boolean shouldTrack;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
public HistoryWritingMessagePostProcessor() {
}
@@ -39,6 +42,10 @@ public class HistoryWritingMessagePostProcessor implements MessagePostProcessor
this.trackableComponent = trackableComponent;
}
public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
Assert.notNull(messageBuilderFactory, "'messageBuilderFactory' cannot be null");
this.messageBuilderFactory = messageBuilderFactory;
}
public void setTrackableComponent(TrackableComponent trackableComponent) {
this.trackableComponent = trackableComponent;
@@ -48,9 +55,10 @@ public class HistoryWritingMessagePostProcessor implements MessagePostProcessor
this.shouldTrack = shouldTrack;
}
@Override
public Message<?> postProcessMessage(Message<?> message) {
if (this.shouldTrack && this.trackableComponent != null) {
return MessageHistory.write(message, this.trackableComponent);
return MessageHistory.write(message, this.trackableComponent, this.messageBuilderFactory);
}
return message;
}

View File

@@ -25,9 +25,10 @@ import java.util.List;
import java.util.ListIterator;
import java.util.Properties;
import org.springframework.messaging.Message;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -46,6 +47,8 @@ public class MessageHistory implements List<Properties>, Serializable {
public static final String TIMESTAMP_PROPERTY = "timestamp";
private static final MessageBuilderFactory mesageBuilderFactory = new DefaultMessageBuilderFactory();
private final List<Properties> components;
@@ -56,6 +59,11 @@ public class MessageHistory implements List<Properties>, Serializable {
}
public static <T> Message<T> write(Message<T> message, NamedComponent component) {
return write(message, component, mesageBuilderFactory);
}
public static <T> Message<T> write(Message<T> message, NamedComponent component,
MessageBuilderFactory messageBuilderFactory) {
Assert.notNull(message, "Message must not be null");
Assert.notNull(component, "Component must not be null");
Properties metadata = extractMetadata(component);
@@ -65,7 +73,7 @@ public class MessageHistory implements List<Properties>, Serializable {
new ArrayList<Properties>(previousHistory) : new ArrayList<Properties>();
components.add(metadata);
MessageHistory history = new MessageHistory(components);
message = MessageBuilder.fromMessage(message).setHeader(HEADER_NAME, history).build();
message = messageBuilderFactory.fromMessage(message).setHeader(HEADER_NAME, history).build();
}
return message;
}
@@ -77,58 +85,72 @@ public class MessageHistory implements List<Properties>, Serializable {
}
@Override
public int size() {
return this.components.size();
}
@Override
public boolean isEmpty() {
return this.components.isEmpty();
}
@Override
public boolean contains(Object o) {
return this.components.contains(o);
}
@Override
public boolean containsAll(Collection<?> c) {
return this.components.containsAll(c);
}
@Override
public Properties get(int index) {
return this.components.get(index);
}
@Override
public Iterator<Properties> iterator() {
return Collections.unmodifiableList(this.components).iterator();
}
@Override
public ListIterator<Properties> listIterator() {
return Collections.unmodifiableList(this.components).listIterator();
}
@Override
public ListIterator<Properties> listIterator(int index) {
return Collections.unmodifiableList(this.components).listIterator(index);
}
@Override
public List<Properties> subList(int fromIndex, int toIndex) {
return Collections.unmodifiableList(this.components).subList(fromIndex, toIndex);
}
@Override
public Object[] toArray() {
return this.components.toArray();
}
@Override
public <T> T[] toArray(T[] a) {
return this.components.toArray(a);
}
@Override
public int indexOf(Object o) {
return this.components.indexOf(o);
}
@Override
public int lastIndexOf(Object o) {
return this.components.lastIndexOf(o);
}
@Override
public String toString() {
List<String> names = new ArrayList<String>();
for (Properties p : this.components) {
@@ -145,42 +167,52 @@ public class MessageHistory implements List<Properties>, Serializable {
* Unsupported Operations
*/
@Override
public boolean add(Properties e) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
}
@Override
public void add(int index, Properties element) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
}
@Override
public boolean addAll(Collection<? extends Properties> c) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
}
@Override
public boolean addAll(int index, Collection<? extends Properties> c) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
}
@Override
public Properties set(int index, Properties element) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
}
@Override
public Properties remove(int index) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
}
@Override
public boolean remove(Object o) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
}
@Override
public boolean removeAll(Collection<?> c) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
}
@Override
public boolean retainAll(Collection<?> c) {
throw new UnsupportedOperationException("MessageHistory is immutable.");
}
@Override
public void clear() {
throw new UnsupportedOperationException("MessageHistory is immutable.");
}

View File

@@ -18,7 +18,7 @@ package org.springframework.integration.json;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.integration.mapping.support.JsonHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.json.JacksonJsonObjectMapperProvider;
import org.springframework.integration.support.json.JsonObjectMapper;
import org.springframework.integration.transformer.AbstractTransformer;
@@ -74,7 +74,7 @@ public class JsonToObjectTransformer extends AbstractTransformer implements Bean
}
else {
Object result = this.jsonObjectMapper.fromJson(message.getPayload(), message.getHeaders());
MessageBuilder<Object> messageBuilder = MessageBuilder.withPayload(result)
AbstractIntegrationMessageBuilder<Object> messageBuilder = this.getMessageBuilderFactory().withPayload(result)
.copyHeaders(message.getHeaders())
.removeHeaders(JsonHeaders.HEADERS.toArray(new String[3]));
return messageBuilder.build();

View File

@@ -15,7 +15,7 @@
*/
package org.springframework.integration.json;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.json.JacksonJsonObjectMapperProvider;
import org.springframework.integration.support.json.JsonObjectMapper;
import org.springframework.integration.transformer.AbstractTransformer;
@@ -91,7 +91,7 @@ public class ObjectToJsonTransformer extends AbstractTransformer {
Object payload = ResultType.STRING.equals(this.resultType)
? this.jsonObjectMapper.toJson(message.getPayload())
: this.jsonObjectMapper.toJsonNode(message.getPayload());
MessageBuilder<Object> messageBuilder = MessageBuilder.withPayload(payload);
AbstractIntegrationMessageBuilder<Object> messageBuilder = this.getMessageBuilderFactory().withPayload(payload);
LinkedCaseInsensitiveMap<Object> headers = new LinkedCaseInsensitiveMap<Object>();
headers.putAll(message.getHeaders());

View File

@@ -0,0 +1,114 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.message;
import java.util.Map;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.ObjectUtils;
/**
* An implementation of {@link Message} with a generic payload. Unlike
* {@link GenericMessage}, this message (or its headers) can be modified
* after creation. Great care must be taken, when mutating messages, that
* some other element/thread is not concurrently using the message. Also note
* that any in-memory stores (such as {@link SimpleMessageStore}) may have
* a reference to the message and changes will be reflected there too.
*
* @author Gary Russell
* @since 4.0
*
*/
public class MutableMessage<T> implements Message<T> {
private T payload;
private final MessageHeaders headers;
private final Map<String, Object> rawHeaders;
public MutableMessage(T payload) {
this(payload, null);
}
@SuppressWarnings("unchecked")
public MutableMessage(T payload, MessageHeaders headers) {
this.payload = payload;
this.headers = new MessageHeaders(headers);
// Needs SPR-11468 to avoid DFA and header manipulation
rawHeaders = (Map<String, Object>) new DirectFieldAccessor(this.headers)
.getPropertyValue("headers");
if (headers != null) {
this.rawHeaders.put(MessageHeaders.ID, headers.get(MessageHeaders.ID));
this.rawHeaders.put(MessageHeaders.TIMESTAMP, headers.get(MessageHeaders.TIMESTAMP));
}
}
@Override
public MessageHeaders getHeaders() {
return this.headers;
}
@Override
public T getPayload() {
return this.payload;
}
public void setPayload(T payload) {
this.payload = payload;
}
public Map<String, Object> getRawHeaders() {
return this.rawHeaders;
}
@Override
public String toString() {
StringBuilder sb = new StringBuilder();
if (this.payload instanceof byte[]) {
sb.append("[Payload byte[").append(((byte[]) this.payload).length).append("]]");
}
else {
sb.append("[Payload ").append(this.payload.getClass().getSimpleName());
sb.append(" content=").append(this.payload).append("]");
}
sb.append("[Headers=").append(this.headers).append("]");
return sb.toString();
}
@Override
public int hashCode() {
return this.headers.hashCode() * 23 + ObjectUtils.nullSafeHashCode(this.payload);
}
@Override
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (obj != null && obj instanceof MutableMessage<?>) {
MutableMessage<?> other = (MutableMessage<?>) obj;
return (this.headers.getId().equals(other.headers.getId()) &&
this.headers.equals(other.headers) && this.payload.equals(other.payload));
}
return false;
}
}

View File

@@ -23,7 +23,6 @@ import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -149,7 +148,7 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler {
int sequenceSize = results.size();
int sequenceNumber = 1;
for (MessageChannel channel : results) {
final Message<?> messageToSend = (!this.applySequence) ? message : MessageBuilder.fromMessage(message)
final Message<?> messageToSend = (!this.applySequence) ? message : this.getMessageBuilderFactory().fromMessage(message)
.pushSequenceDetails(message.getHeaders().getId(), sequenceNumber++, sequenceSize).build();
if (channel != null) {
try {

View File

@@ -21,7 +21,7 @@ import java.util.Collection;
import java.util.List;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.CollectionUtils;
@@ -57,7 +57,7 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
}
MessageHeaders headers = message.getHeaders();
Object correlationId = headers.getId();
List<MessageBuilder<?>> messageBuilders = new ArrayList<MessageBuilder<?>>();
List<AbstractIntegrationMessageBuilder<?>> messageBuilders = new ArrayList<AbstractIntegrationMessageBuilder<?>>();
if (result instanceof Collection) {
Collection<?> items = (Collection<?>) result;
int sequenceNumber = 0;
@@ -81,14 +81,14 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess
}
@SuppressWarnings( { "unchecked", "rawtypes" })
private MessageBuilder createBuilder(Object item, MessageHeaders headers, Object correlationId, int sequenceNumber,
private AbstractIntegrationMessageBuilder createBuilder(Object item, MessageHeaders headers, Object correlationId, int sequenceNumber,
int sequenceSize) {
MessageBuilder builder;
AbstractIntegrationMessageBuilder builder;
if (item instanceof Message) {
builder = MessageBuilder.fromMessage((Message) item);
builder = this.getMessageBuilderFactory().fromMessage((Message) item);
}
else {
builder = MessageBuilder.withPayload(item);
builder = this.getMessageBuilderFactory().withPayload(item);
builder.copyHeaders(headers);
}
if (this.applySequence) {

View File

@@ -25,10 +25,9 @@ import java.util.Set;
import java.util.UUID;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.util.Assert;
/**
@@ -47,6 +46,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
// MessageStore methods
@Override
public Message<?> getMessage(UUID id) {
Message<?> message = this.getRawMessage(id);
if (message != null){
@@ -55,6 +55,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
return null;
}
@Override
@SuppressWarnings("unchecked")
public <T> Message<T> addMessage(Message<T> message) {
Assert.notNull(message, "'message' must not be null");
@@ -63,6 +64,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
return (Message<T>) this.getRawMessage(messageId);
}
@Override
public Message<?> removeMessage(UUID id) {
Assert.notNull(id, "'id' must not be null");
Object message = this.doRemove(MESSAGE_KEY_PREFIX + id);
@@ -75,6 +77,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
return null;
}
@Override
@ManagedAttribute
public long getMessageCount() {
Collection<?> messageIds = this.doListKeys(MESSAGE_KEY_PREFIX + "*");
@@ -87,6 +90,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
/**
* Will create a new instance of SimpleMessageGroup if necessary.
*/
@Override
public MessageGroup getMessageGroup(Object groupId) {
return this.buildMessageGroup(groupId, false);
}
@@ -95,6 +99,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
/**
* Add a Message to the group with the provided group ID.
*/
@Override
public MessageGroup addMessageToGroup(Object groupId, Message<?> message) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(message, "'message' must not be null");
@@ -124,6 +129,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
/**
* Remove a Message from the group with the provided group ID.
*/
@Override
public MessageGroup removeMessageFromGroup(Object groupId, Message<?> messageToRemove) {
Assert.notNull(groupId, "'groupId' must not be null");
Assert.notNull(messageToRemove, "'messageToRemove' must not be null");
@@ -149,6 +155,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
}
@Override
public void completeGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
SimpleMessageGroup messageGroup = this.buildMessageGroup(groupId, true);
@@ -160,6 +167,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
/**
* Remove the MessageGroup with the provided group ID.
*/
@Override
public void removeMessageGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Object mgm = this.doRemove(MESSAGE_GROUP_KEY_PREFIX + groupId);
@@ -174,6 +182,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
}
}
@Override
public void setLastReleasedSequenceNumberForGroup(Object groupId, int sequenceNumber) {
Assert.notNull(groupId, "'groupId' must not be null");
SimpleMessageGroup messageGroup = this.buildMessageGroup(groupId, true);
@@ -182,6 +191,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
this.doStore(MESSAGE_GROUP_KEY_PREFIX + groupId, new MessageGroupMetadata(messageGroup));
}
@Override
public Message<?> pollMessageFromGroup(Object groupId) {
Assert.notNull(groupId, "'groupId' must not be null");
Object mgm = this.doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId);
@@ -200,6 +210,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
return null;
}
@Override
@SuppressWarnings("unchecked")
public Iterator<MessageGroup> iterator() {
final Iterator<?> idIterator = this.normalizeKeys(
@@ -223,6 +234,7 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
return normalizedKeys;
}
@Override
public int messageGroupSize(Object groupId) {
Object mgm = this.doRetrieve(MESSAGE_GROUP_KEY_PREFIX + groupId);
if (mgm != null) {
@@ -243,7 +255,9 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
@SuppressWarnings({ "rawtypes", "unchecked" })
private Message<?> normalizeMessage(Message<?> message){
Message<?> normalizedMessage = MessageBuilder.fromMessage(message).removeHeader("CREATED_DATE").build();
Message<?> normalizedMessage = this.getMessageBuilderFactory().fromMessage(message)
.removeHeader("CREATED_DATE")
.build();
Map innerMap = (Map) new DirectFieldAccessor(normalizedMessage.getHeaders()).getPropertyValue("headers");
innerMap.put(MessageHeaders.ID, message.getHeaders().getId());
innerMap.put(MessageHeaders.TIMESTAMP, message.getHeaders().getTimestamp());
@@ -255,7 +269,9 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
*/
@SuppressWarnings({ "rawtypes", "unchecked" })
private Message<?> enrichMessage(Message<?> message){
Message<?> enrichedMessage = MessageBuilder.fromMessage(message).setHeader(CREATED_DATE, System.currentTimeMillis()).build();
Message<?> enrichedMessage = this.getMessageBuilderFactory().fromMessage(message)
.setHeader(CREATED_DATE, System.currentTimeMillis())
.build();
Map innerMap = (Map) new DirectFieldAccessor(enrichedMessage.getHeaders()).getPropertyValue("headers");
innerMap.put(MessageHeaders.ID, message.getHeaders().getId());
innerMap.put(MessageHeaders.TIMESTAMP, message.getHeaders().getTimestamp());
@@ -323,15 +339,18 @@ public abstract class AbstractKeyValueMessageStore extends AbstractMessageGroupS
this.idIterator = idIterator;
}
@Override
public boolean hasNext() {
return idIterator.hasNext();
}
@Override
public MessageGroup next() {
Object messageGroupId = idIterator.next();
return getMessageGroup(messageGroupId);
}
@Override
public void remove() {
throw new UnsupportedOperationException();
}

View File

@@ -19,16 +19,23 @@ import java.util.LinkedHashSet;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
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.jmx.export.annotation.ManagedAttribute;
/**
* @author Dave Syer
* @author Oleg Zhurakousky
* @author Gary Russell
*
* @since 2.0
*
*/
public abstract class AbstractMessageGroupStore implements MessageGroupStore, Iterable<MessageGroup> {
public abstract class AbstractMessageGroupStore implements MessageGroupStore, Iterable<MessageGroup>,
BeanFactoryAware {
protected final Log logger = LogFactory.getLog(getClass());
@@ -36,10 +43,24 @@ public abstract class AbstractMessageGroupStore implements MessageGroupStore, It
private volatile boolean timeoutOnIdle;
private volatile BeanFactory beanFactory;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
public AbstractMessageGroupStore() {
super();
}
@Override
public final void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(this.beanFactory);
}
protected MessageBuilderFactory getMessageBuilderFactory() {
return messageBuilderFactory;
}
/**
* Convenient injection point for expiry callbacks in the message store. Each of the callbacks provided will simply
* be registered with the store using {@link #registerMessageGroupExpiryCallback(MessageGroupCallback)}.

View File

@@ -0,0 +1,199 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.support;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Map;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.Assert;
/**
* @author Gary Russell
* @since 4.0
*
*/
public abstract class AbstractIntegrationMessageBuilder<T> {
/**
* Set the value for the given header name. If the provided value is <code>null</code>, the header will be removed.
*
* @param headerName The header name.
* @param headerValue The header value.
* @return this.
*/
public abstract AbstractIntegrationMessageBuilder<T> setHeader(String headerName, Object headerValue);
/**
* Set the value for the given header name only if the header name is not already associated with a value.
*
* @param headerName The header name.
* @param headerValue The header value.
* @return this.
*/
public abstract AbstractIntegrationMessageBuilder<T> setHeaderIfAbsent(String headerName, Object headerValue);
/**
* Removes all headers provided via array of 'headerPatterns'. As the name suggests the array
* may contain simple matching patterns for header names. Supported pattern styles are:
* "xxx*", "*xxx", "*xxx*" and "xxx*yyy".
*
* @param headerPatterns The header patterns.
* @return this.
*/
public abstract AbstractIntegrationMessageBuilder<T> removeHeaders(String... headerPatterns);
/**
* Remove the value for the given header name.
* @param headerName The header name.
* @return this.
*/
public abstract AbstractIntegrationMessageBuilder<T> removeHeader(String headerName);
/**
* Copy the name-value pairs from the provided Map. This operation will overwrite any existing values. Use {
* {@link #copyHeadersIfAbsent(Map)} to avoid overwriting values. Note that the 'id' and 'timestamp' header values
* will never be overwritten.
*
* @param headersToCopy The headers to copy.
* @return this.
*
* @see MessageHeaders#ID
* @see MessageHeaders#TIMESTAMP
*/
public abstract AbstractIntegrationMessageBuilder<T> copyHeaders(Map<String, ?> headersToCopy);
/**
* Copy the name-value pairs from the provided Map. This operation will <em>not</em> overwrite any existing values.
*
* @param headersToCopy The headers to copy.
* @return this.
*/
public abstract AbstractIntegrationMessageBuilder<T> copyHeadersIfAbsent(Map<String, ?> headersToCopy);
public AbstractIntegrationMessageBuilder<T> setExpirationDate(Long expirationDate) {
return this.setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, expirationDate);
}
public AbstractIntegrationMessageBuilder<T> setExpirationDate(Date expirationDate) {
if (expirationDate != null) {
return this.setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, expirationDate.getTime());
}
else {
return this.setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, null);
}
}
public AbstractIntegrationMessageBuilder<T> setCorrelationId(Object correlationId) {
return this.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, correlationId);
}
public AbstractIntegrationMessageBuilder<T> pushSequenceDetails(Object correlationId, int sequenceNumber,
int sequenceSize) {
Object incomingCorrelationId = this.getCorrelationId();
List<List<Object>> incomingSequenceDetails = this.getSequenceDetails();
if (incomingCorrelationId != null) {
if (incomingSequenceDetails == null) {
incomingSequenceDetails = new ArrayList<List<Object>>();
}
else {
incomingSequenceDetails = new ArrayList<List<Object>>(incomingSequenceDetails);
}
incomingSequenceDetails.add(Arrays.asList(incomingCorrelationId,
this.getSequenceNumber(), this.getSequenceSize()));
incomingSequenceDetails = Collections.unmodifiableList(incomingSequenceDetails);
}
if (incomingSequenceDetails != null) {
this.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS, incomingSequenceDetails);
}
return setCorrelationId(correlationId).setSequenceNumber(sequenceNumber).setSequenceSize(sequenceSize);
}
public AbstractIntegrationMessageBuilder<T> popSequenceDetails() {
List<List<Object>> incomingSequenceDetails = this.getSequenceDetails();
if (incomingSequenceDetails == null) {
return this;
}
else {
incomingSequenceDetails = new ArrayList<List<Object>>(incomingSequenceDetails);
}
List<Object> sequenceDetails = incomingSequenceDetails.remove(incomingSequenceDetails.size() - 1);
Assert.state(sequenceDetails.size() == 3, "Wrong sequence details (not created by MessageBuilder?): "
+ sequenceDetails);
setCorrelationId(sequenceDetails.get(0));
Integer sequenceNumber = (Integer) sequenceDetails.get(1);
Integer sequenceSize = (Integer) sequenceDetails.get(2);
if (sequenceNumber != null) {
setSequenceNumber(sequenceNumber);
}
if (sequenceSize != null) {
setSequenceSize(sequenceSize);
}
if (!incomingSequenceDetails.isEmpty()) {
this.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS, incomingSequenceDetails);
}
else {
this.removeHeader(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS);
}
return this;
}
protected abstract List<List<Object>> getSequenceDetails();
protected abstract Object getCorrelationId();
protected abstract Object getSequenceNumber();
protected abstract Object getSequenceSize();
public AbstractIntegrationMessageBuilder<T> setReplyChannel(MessageChannel replyChannel) {
return this.setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel);
}
public AbstractIntegrationMessageBuilder<T> setReplyChannelName(String replyChannelName) {
return this.setHeader(MessageHeaders.REPLY_CHANNEL, replyChannelName);
}
public AbstractIntegrationMessageBuilder<T> setErrorChannel(MessageChannel errorChannel) {
return this.setHeader(MessageHeaders.ERROR_CHANNEL, errorChannel);
}
public AbstractIntegrationMessageBuilder<T> setErrorChannelName(String errorChannelName) {
return this.setHeader(MessageHeaders.ERROR_CHANNEL, errorChannelName);
}
public AbstractIntegrationMessageBuilder<T> setSequenceNumber(Integer sequenceNumber) {
return this.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, sequenceNumber);
}
public AbstractIntegrationMessageBuilder<T> setSequenceSize(Integer sequenceSize) {
return this.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, sequenceSize);
}
public AbstractIntegrationMessageBuilder<T> setPriority(Integer priority) {
return this.setHeader(IntegrationMessageHeaderAccessor.PRIORITY, priority);
}
public abstract Message<T> build();
}

View File

@@ -0,0 +1,37 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.support;
import org.springframework.messaging.Message;
/**
* @author Gary Russell
* @since 4.0
*
*/
public class DefaultMessageBuilderFactory implements MessageBuilderFactory {
@Override
public <T> MessageBuilder<T> fromMessage(Message<T> message) {
return MessageBuilder.fromMessage(message);
}
@Override
public <T> MessageBuilder<T> withPayload(T payload) {
return MessageBuilder.withPayload(payload);
}
}

View File

@@ -64,4 +64,4 @@ public class IdGenerators {
}
}
}
}

View File

@@ -16,9 +16,6 @@
package org.springframework.integration.support;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Map;
@@ -32,12 +29,17 @@ import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.Assert;
/**
* The default message builder; creates immutable {@link GenericMessage}s.
* Named MessageBuilder instead of DefaultMessageBuilder for backwards
* compatibility.
*
* @author Arjen Poutsma
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Dave Syer
* @author Gary Russell
*/
public final class MessageBuilder<T> {
public final class MessageBuilder<T> extends AbstractIntegrationMessageBuilder<T> {
private final T payload;
@@ -93,6 +95,7 @@ public final class MessageBuilder<T> {
* @param headerValue The header value.
* @return this MessageBuilder.
*/
@Override
public MessageBuilder<T> setHeader(String headerName, Object headerValue) {
this.headerAccessor.setHeader(headerName, headerValue);
return this;
@@ -105,6 +108,7 @@ public final class MessageBuilder<T> {
* @param headerValue The header value.
* @return this MessageBuilder.
*/
@Override
public MessageBuilder<T> setHeaderIfAbsent(String headerName, Object headerValue) {
this.headerAccessor.setHeaderIfAbsent(headerName, headerValue);
return this;
@@ -118,6 +122,7 @@ public final class MessageBuilder<T> {
* @param headerPatterns The header patterns.
* @return this MessageBuilder.
*/
@Override
public MessageBuilder<T> removeHeaders(String... headerPatterns) {
this.headerAccessor.removeHeaders(headerPatterns);
return this;
@@ -127,6 +132,7 @@ public final class MessageBuilder<T> {
* @param headerName The header name.
* @return this MessageBuilder.
*/
@Override
public MessageBuilder<T> removeHeader(String headerName) {
this.headerAccessor.removeHeader(headerName);
return this;
@@ -143,6 +149,7 @@ public final class MessageBuilder<T> {
* @see MessageHeaders#ID
* @see MessageHeaders#TIMESTAMP
*/
@Override
public MessageBuilder<T> copyHeaders(Map<String, ?> headersToCopy) {
this.headerAccessor.copyHeaders(headersToCopy);
return this;
@@ -154,107 +161,111 @@ public final class MessageBuilder<T> {
* @param headersToCopy The headers to copy.
* @return this MessageBuilder.
*/
@Override
public MessageBuilder<T> copyHeadersIfAbsent(Map<String, ?> headersToCopy) {
this.headerAccessor.copyHeadersIfAbsent(headersToCopy);
return this;
}
public MessageBuilder<T> setExpirationDate(Long expirationDate) {
return this.setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, expirationDate);
@SuppressWarnings("unchecked")
@Override
protected List<List<Object>> getSequenceDetails() {
return (List<List<Object>>) this.headerAccessor.getHeader(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS);
}
public MessageBuilder<T> setExpirationDate(Date expirationDate) {
if (expirationDate != null) {
return this.setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, expirationDate.getTime());
}
else {
return this.setHeader(IntegrationMessageHeaderAccessor.EXPIRATION_DATE, null);
}
@Override
protected Object getCorrelationId() {
return this.headerAccessor.getCorrelationId();
}
public MessageBuilder<T> setCorrelationId(Object correlationId) {
return this.setHeader(IntegrationMessageHeaderAccessor.CORRELATION_ID, correlationId);
@Override
protected Object getSequenceNumber() {
return this.headerAccessor.getSequenceNumber();
}
@Override
protected Object getSequenceSize() {
return this.headerAccessor.getSequenceSize();
}
/*
* The following overrides (delegating to super) are provided to ease the
* pain for existing applications that use the builder API and expect
* a MessageBuilder to be returned.
*/
@Override
public MessageBuilder<T> pushSequenceDetails(Object correlationId, int sequenceNumber, int sequenceSize) {
Object incomingCorrelationId = this.headerAccessor.getCorrelationId();
@SuppressWarnings("unchecked")
List<List<Object>> incomingSequenceDetails = (List<List<Object>>) this.headerAccessor.getHeader(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS);
if (incomingCorrelationId != null) {
if (incomingSequenceDetails == null) {
incomingSequenceDetails = new ArrayList<List<Object>>();
}
else {
incomingSequenceDetails = new ArrayList<List<Object>>(incomingSequenceDetails);
}
incomingSequenceDetails.add(Arrays.asList(incomingCorrelationId,
this.headerAccessor.getSequenceNumber(), this.headerAccessor.getSequenceSize()));
incomingSequenceDetails = Collections.unmodifiableList(incomingSequenceDetails);
}
if (incomingSequenceDetails != null) {
setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS, incomingSequenceDetails);
}
return setCorrelationId(correlationId).setSequenceNumber(sequenceNumber).setSequenceSize(sequenceSize);
}
public MessageBuilder<T> popSequenceDetails() {
String key = IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS;
@SuppressWarnings("unchecked")
List<List<Object>> incomingSequenceDetails = (List<List<Object>>) this.headerAccessor.getHeader(key);
if (incomingSequenceDetails == null) {
return this;
} else {
incomingSequenceDetails = new ArrayList<List<Object>>(incomingSequenceDetails);
}
List<Object> sequenceDetails = incomingSequenceDetails.remove(incomingSequenceDetails.size() - 1);
Assert.state(sequenceDetails.size() == 3, "Wrong sequence details (not created by MessageBuilder?): "
+ sequenceDetails);
setCorrelationId(sequenceDetails.get(0));
Integer sequenceNumber = (Integer) sequenceDetails.get(1);
Integer sequenceSize = (Integer) sequenceDetails.get(2);
if (sequenceNumber != null) {
setSequenceNumber(sequenceNumber);
}
if (sequenceSize != null) {
setSequenceSize(sequenceSize);
}
if (!incomingSequenceDetails.isEmpty()) {
this.headerAccessor.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS, incomingSequenceDetails);
}
else {
this.headerAccessor.removeHeader(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS);
}
super.pushSequenceDetails(correlationId, sequenceNumber, sequenceSize);
return this;
}
@Override
public MessageBuilder<T> popSequenceDetails() {
super.popSequenceDetails();
return this;
}
@Override
public MessageBuilder<T> setExpirationDate(Long expirationDate) {
super.setExpirationDate(expirationDate);
return this;
}
@Override
public MessageBuilder<T> setExpirationDate(Date expirationDate) {
super.setExpirationDate(expirationDate);
return this;
}
@Override
public MessageBuilder<T> setCorrelationId(Object correlationId) {
super.setCorrelationId(correlationId);
return this;
}
@Override
public MessageBuilder<T> setReplyChannel(MessageChannel replyChannel) {
return this.setHeader(MessageHeaders.REPLY_CHANNEL, replyChannel);
super.setReplyChannel(replyChannel);
return this;
}
@Override
public MessageBuilder<T> setReplyChannelName(String replyChannelName) {
return this.setHeader(MessageHeaders.REPLY_CHANNEL, replyChannelName);
super.setReplyChannelName(replyChannelName);
return this;
}
@Override
public MessageBuilder<T> setErrorChannel(MessageChannel errorChannel) {
return this.setHeader(MessageHeaders.ERROR_CHANNEL, errorChannel);
super.setErrorChannel(errorChannel);
return this;
}
@Override
public MessageBuilder<T> setErrorChannelName(String errorChannelName) {
return this.setHeader(MessageHeaders.ERROR_CHANNEL, errorChannelName);
super.setErrorChannelName(errorChannelName);
return this;
}
@Override
public MessageBuilder<T> setSequenceNumber(Integer sequenceNumber) {
return this.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, sequenceNumber);
super.setSequenceNumber(sequenceNumber);
return this;
}
@Override
public MessageBuilder<T> setSequenceSize(Integer sequenceSize) {
return this.setHeader(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE, sequenceSize);
super.setSequenceSize(sequenceSize);
return this;
}
@Override
public MessageBuilder<T> setPriority(Integer priority) {
return this.setHeader(IntegrationMessageHeaderAccessor.PRIORITY, priority);
super.setPriority(priority);
return this;
}
@Override
@SuppressWarnings("unchecked")
public Message<T> build() {
if (!this.modified && !this.headerAccessor.isModified() && this.originalMessage != null) {
@@ -265,4 +276,5 @@ public final class MessageBuilder<T> {
}
return new GenericMessage<T>(this.payload, this.headerAccessor.toMap());
}
}

View File

@@ -0,0 +1,31 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.support;
import org.springframework.messaging.Message;
/**
* @author Gary Russell
* @since 4.0
*
*/
public interface MessageBuilderFactory {
<T> AbstractIntegrationMessageBuilder<T> fromMessage(Message<T> message);
<T> AbstractIntegrationMessageBuilder<T> withPayload(T payload);
}

View File

@@ -0,0 +1,185 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.support;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.message.MutableMessage;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import org.springframework.util.PatternMatchUtils;
import org.springframework.util.StringUtils;
/**
* @author Gary Russell
* @since 4.0
*
*/
public class MutableMessageBuilder<T> extends AbstractIntegrationMessageBuilder<T> {
private MutableMessage<T> mutableMessage;
private final Map<String, Object> headers;
/**
* Private constructor to be invoked from the static factory methods only.
*/
private MutableMessageBuilder(Message<T> message) {
Assert.notNull(message, "message must not be null");
if (message instanceof MutableMessage) {
this.mutableMessage = (MutableMessage<T>) message;
}
else {
this.mutableMessage = new MutableMessage<T>(message.getPayload(), message.getHeaders());
}
this.headers = this.mutableMessage.getRawHeaders();
}
/**
* Create a builder for a new {@link Message} instance pre-populated with all of the headers copied from the
* provided message. The payload of the provided Message will also be used as the payload for the new message.
*
* @param message the Message from which the payload and all headers will be copied
* @param <T> The type of the payload.
* @return A MutableMessageBuilder.
*/
public static <T> MutableMessageBuilder<T> fromMessage(Message<T> message) {
Assert.notNull(message, "message must not be null");
MutableMessageBuilder<T> builder = new MutableMessageBuilder<T>(message);
return builder;
}
/**
* Create a builder for a new {@link Message} instance with the provided payload.
*
* @param payload the payload for the new message
* @param <T> The type of the payload.
* @return A MessageBuilder.
*/
public static <T> MutableMessageBuilder<T> withPayload(T payload) {
MutableMessageBuilder<T> builder = new MutableMessageBuilder<T>(new MutableMessage<T>(payload));
return builder;
}
@Override
public AbstractIntegrationMessageBuilder<T> setHeader(String headerName, Object headerValue) {
Assert.notNull(headerName);
if (headerValue == null) {
this.removeHeader(headerName);
}
else {
this.headers.put(headerName, headerValue);
}
return this;
}
@Override
public AbstractIntegrationMessageBuilder<T> setHeaderIfAbsent(String headerName, Object headerValue) {
if (!this.headers.containsKey(headerName)) {
this.headers.put(headerName, headerValue);
}
return this;
}
@Override
public AbstractIntegrationMessageBuilder<T> removeHeaders(String... headerPatterns) {
List<String> headersToRemove = new ArrayList<String>();
for (String pattern : headerPatterns) {
if (StringUtils.hasLength(pattern)){
if (pattern.contains("*")){
headersToRemove.addAll(getMatchingHeaderNames(pattern, this.headers));
}
else {
headersToRemove.add(pattern);
}
}
}
for (String headerToRemove : headersToRemove) {
removeHeader(headerToRemove);
}
return this;
}
private List<String> getMatchingHeaderNames(String pattern, Map<String, Object> headers) {
List<String> matchingHeaderNames = new ArrayList<String>();
if (headers != null) {
for (Map.Entry<String, Object> header: headers.entrySet()) {
if (PatternMatchUtils.simpleMatch(pattern, header.getKey())) {
matchingHeaderNames.add(header.getKey());
}
}
}
return matchingHeaderNames;
}
@Override
public AbstractIntegrationMessageBuilder<T> removeHeader(String headerName) {
if (StringUtils.hasLength(headerName)) {
this.headers.remove(headerName);
}
return this;
}
@Override
public AbstractIntegrationMessageBuilder<T> copyHeaders(Map<String, ?> headersToCopy) {
if (headersToCopy != null) {
this.headers.putAll(headersToCopy);
}
return this;
}
@Override
public AbstractIntegrationMessageBuilder<T> copyHeadersIfAbsent(Map<String, ?> headersToCopy) {
if (headersToCopy != null) {
for (Entry<String, ?> entry : headersToCopy.entrySet()) {
setHeaderIfAbsent(entry.getKey(), entry.getValue());
}
}
return this;
}
@SuppressWarnings("unchecked")
@Override
protected List<List<Object>> getSequenceDetails() {
return (List<List<Object>>) this.headers.get(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS);
}
@Override
protected Object getCorrelationId() {
return this.headers.get(IntegrationMessageHeaderAccessor.CORRELATION_ID);
}
@Override
protected Object getSequenceNumber() {
return this.headers.get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER);
}
@Override
protected Object getSequenceSize() {
return this.headers.get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE);
}
@Override
public Message<T> build() {
return this.mutableMessage;
}
}

View File

@@ -0,0 +1,37 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.support;
import org.springframework.messaging.Message;
/**
* @author Gary Russell
* @since 4.0
*
*/
public class MutableMessageBuilderFacfory implements MessageBuilderFactory {
@Override
public <T> MutableMessageBuilder<T> fromMessage(Message<T> message) {
return MutableMessageBuilder.fromMessage(message);
}
@Override
public <T> MutableMessageBuilder<T> withPayload(T payload) {
return MutableMessageBuilder.withPayload(payload);
}
}

View File

@@ -19,7 +19,12 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConverter;
@@ -31,12 +36,27 @@ import org.springframework.util.Assert;
* @since 3.0
*
*/
public class MapMessageConverter implements MessageConverter {
public class MapMessageConverter implements MessageConverter, BeanFactoryAware {
private volatile String[] headerNames;
private volatile boolean filterHeadersInToMessage;
private volatile BeanFactory beanFactory;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
@Override
public final void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(this.beanFactory);
}
protected MessageBuilderFactory getMessageBuilderFactory() {
return messageBuilderFactory;
}
/**
* Headers to be converted in {@link #fromMessage(Message, Class)}.
* {@link #toMessage(Object, MessageHeaders)} will populate all headers found in
@@ -67,7 +87,7 @@ public class MapMessageConverter implements MessageConverter {
Map<String, ?> map = (Map<String, ?>) object;
Object payload = map.get("payload");
Assert.notNull(payload, "'payload' entry cannot be null");
MessageBuilder<?> messageBuilder = MessageBuilder.withPayload(payload);
AbstractIntegrationMessageBuilder<?> messageBuilder = this.messageBuilderFactory.withPayload(payload);
@SuppressWarnings("unchecked")
Map<String, ?> headers = (Map<String, ?>) map.get("headers");
if (headers != null) {

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.
@@ -18,7 +18,8 @@ package org.springframework.integration.support.converter;
import org.springframework.integration.mapping.InboundMessageMapper;
import org.springframework.integration.mapping.OutboundMessageMapper;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.MessageConversionException;
@@ -26,6 +27,7 @@ import org.springframework.messaging.converter.MessageConverter;
/**
* @author Mark Fisher
* @author Gary Russell
* @since 2.0
*/
@SuppressWarnings({"unchecked", "rawtypes"})
@@ -35,6 +37,7 @@ public class SimpleMessageConverter implements MessageConverter {
private volatile OutboundMessageMapper outboundMessageMapper;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
public SimpleMessageConverter() {
this(null, null);
@@ -64,6 +67,11 @@ public class SimpleMessageConverter implements MessageConverter {
this.outboundMessageMapper = (outboundMessageMapper != null) ? outboundMessageMapper : new DefaultOutboundMessageMapper();
}
public final void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
this.messageBuilderFactory = messageBuilderFactory;
}
@Override
public Message<?> toMessage(Object object, MessageHeaders headers) {
try {
return this.inboundMessageMapper.toMessage(object);
@@ -73,6 +81,7 @@ public class SimpleMessageConverter implements MessageConverter {
}
}
@Override
public Object fromMessage(Message<?> message, Class<?> targetClass) {
try {
return this.outboundMessageMapper.fromMessage(message);
@@ -83,8 +92,9 @@ public class SimpleMessageConverter implements MessageConverter {
}
private static class DefaultInboundMessageMapper implements InboundMessageMapper<Object> {
private class DefaultInboundMessageMapper implements InboundMessageMapper<Object> {
@Override
public Message<?> toMessage(Object object) throws Exception {
if (object == null) {
return null;
@@ -92,13 +102,14 @@ public class SimpleMessageConverter implements MessageConverter {
if (object instanceof Message<?>) {
return (Message<?>) object;
}
return MessageBuilder.withPayload(object).build();
return messageBuilderFactory.withPayload(object).build();
}
}
private static class DefaultOutboundMessageMapper implements OutboundMessageMapper<Object> {
private class DefaultOutboundMessageMapper implements OutboundMessageMapper<Object> {
@Override
public Object fromMessage(Message<?> message) throws Exception {
return (message != null) ? message.getPayload() : null;
}

View File

@@ -18,8 +18,10 @@ package org.springframework.integration.support.json;
import java.lang.reflect.Type;
import org.springframework.integration.support.MessageBuilder;
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.
@@ -34,10 +36,21 @@ abstract class AbstractJacksonJsonMessageParser<P> implements JsonInboundMessage
private volatile JsonInboundMessageMapper messageMapper;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
protected AbstractJacksonJsonMessageParser(JsonObjectMapper<?, P> objectMapper) {
this.objectMapper = objectMapper;
}
public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
Assert.notNull(messageBuilderFactory, "'messageBuilderFactory' cannot be null");
this.messageBuilderFactory = messageBuilderFactory;
}
protected MessageBuilderFactory getMessageBuilderFactory() {
return messageBuilderFactory;
}
@Override
public Message<?> doInParser(JsonInboundMessageMapper messageMapper, String jsonMessage) throws Exception {
if (this.messageMapper == null) {
@@ -47,7 +60,7 @@ abstract class AbstractJacksonJsonMessageParser<P> implements JsonInboundMessage
if (messageMapper.isMapToPayload()) {
Object payload = this.readPayload(parser, jsonMessage);
return MessageBuilder.withPayload(payload).build();
return this.messageBuilderFactory.withPayload(payload).build();
}
else {
return this.parseWithHeaders(parser, jsonMessage);

View File

@@ -20,7 +20,6 @@ package org.springframework.integration.support.json;
import java.util.LinkedHashMap;
import java.util.Map;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
@@ -69,7 +68,7 @@ public class Jackson2JsonMessageParser extends AbstractJacksonJsonMessageParser<
}
}
Assert.notNull(headers, error);
return MessageBuilder.withPayload(payload).copyHeaders(headers).build();
return this.getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build();
}
private Map<String, Object> readHeaders(JsonParser parser, String jsonMessage) throws Exception {

View File

@@ -23,7 +23,6 @@ import org.codehaus.jackson.JsonFactory;
import org.codehaus.jackson.JsonParser;
import org.codehaus.jackson.JsonToken;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
@@ -70,7 +69,7 @@ public class JacksonJsonMessageParser extends AbstractJacksonJsonMessageParser<J
}
}
Assert.notNull(headers, error);
return MessageBuilder.withPayload(payload).copyHeaders(headers).build();
return this.getMessageBuilderFactory().withPayload(payload).copyHeaders(headers).build();
}
private Map<String, Object> readHeaders(JsonParser parser, String jsonMessage) throws Exception {

View File

@@ -16,11 +16,10 @@ import java.util.Map.Entry;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.util.Assert;
/**
@@ -114,7 +113,7 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int
"as part of '" + expressionType + "' transaction synchronization");
}
try {
spelResultMessage = MessageBuilder.withPayload(value)
spelResultMessage = this.getMessageBuilderFactory().withPayload(value)
.copyHeaders(message.getHeaders())
.build();
this.sendMessage(messageChannel, spelResultMessage);
@@ -138,7 +137,7 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int
// rollback will be initiated if any of the previous sync operations fail (e.g., beforeCommit)
// this means that this method will be called without explicit configuration thus no channel
if (messageChannel != null){
this.sendMessage(messageChannel, MessageBuilder.fromMessage(message).build());
this.sendMessage(messageChannel, this.getMessageBuilderFactory().fromMessage(message).build());
}
} catch (Exception e) {
logger.error("Failed to send " + message, e);

View File

@@ -19,29 +19,35 @@ package org.springframework.integration.transformer;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.core.convert.ConversionService;
import org.springframework.messaging.Message;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.handler.AbstractMessageProcessor;
import org.springframework.integration.handler.MessageProcessor;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
* Base class for Message Transformers that delegate to a {@link MessageProcessor}.
*
*
* @author Mark Fisher
*/
public abstract class AbstractMessageProcessingTransformer implements Transformer, BeanFactoryAware {
private final MessageProcessor<?> messageProcessor;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
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) {
if (this.messageProcessor instanceof BeanFactoryAware) {
((BeanFactoryAware) this.messageProcessor).setBeanFactory(beanFactory);
@@ -50,8 +56,10 @@ public abstract class AbstractMessageProcessingTransformer implements Transforme
if (conversionService != null && this.messageProcessor instanceof AbstractMessageProcessor) {
((AbstractMessageProcessor<?>) this.messageProcessor).setConversionService(conversionService);
}
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(beanFactory);
}
@Override
public final Message<?> transform(Message<?> message) {
Object result = this.messageProcessor.processMessage(message);
if (result == null) {
@@ -60,7 +68,7 @@ public abstract class AbstractMessageProcessingTransformer implements Transforme
if (result instanceof Message<?>) {
return (Message<?>) result;
}
return MessageBuilder.withPayload(result).copyHeaders(message.getHeaders()).build();
return this.messageBuilderFactory.withPayload(result).copyHeaders(message.getHeaders()).build();
}
}

View File

@@ -17,7 +17,6 @@
package org.springframework.integration.transformer;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
/**
@@ -36,7 +35,7 @@ public abstract class AbstractTransformer extends IntegrationObjectSupport imple
return null;
}
return (result instanceof Message) ? (Message<?>) result
: MessageBuilder.withPayload(result).copyHeaders(message.getHeaders()).build();
: this.getMessageBuilderFactory().withPayload(result).copyHeaders(message.getHeaders()).build();
}
catch (MessageTransformationException e) {
throw e;

View File

@@ -17,7 +17,7 @@
package org.springframework.integration.transformer;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
@@ -50,7 +50,7 @@ public class ClaimCheckInTransformer extends AbstractTransformer {
Object payload = message.getPayload();
Assert.notNull(payload, "payload must not be null");
Message<?> storedMessage = this.messageStore.addMessage(message);
MessageBuilder<?> responseBuilder = MessageBuilder.withPayload(storedMessage.getHeaders().getId());
AbstractIntegrationMessageBuilder<?> responseBuilder = this.getMessageBuilderFactory().withPayload(storedMessage.getHeaders().getId());
// headers on the 'current' message take precedence
responseBuilder.copyHeaders(message.getHeaders());
return responseBuilder.build();

View File

@@ -19,7 +19,7 @@ package org.springframework.integration.transformer;
import java.util.UUID;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
@@ -71,7 +71,7 @@ public class ClaimCheckOutTransformer extends AbstractTransformer {
}
Assert.notNull(retrievedMessage, "unable to locate Message for ID: " + id
+ " within MessageStore [" + this.messageStore + "]");
MessageBuilder<?> responseBuilder = MessageBuilder.fromMessage(retrievedMessage);
AbstractIntegrationMessageBuilder<?> responseBuilder = this.getMessageBuilderFactory().fromMessage(retrievedMessage);
// headers on the 'current' message take precedence
responseBuilder.copyHeaders(message.getHeaders());
return responseBuilder.build();

View File

@@ -30,7 +30,6 @@ import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.expression.IntegrationEvaluationContextAware;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.transformer.support.HeaderValueMessageProcessor;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -290,7 +289,7 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler implem
}
else {
final Object requestMessagePayload = this.requestPayloadExpression.getValue(this.sourceEvaluationContext, requestMessage);
actualRequestMessage = MessageBuilder.withPayload(requestMessagePayload)
actualRequestMessage = this.getMessageBuilderFactory().withPayload(requestMessagePayload)
.copyHeaders(requestMessage.getHeaders()).build();
}
final Message<?> replyMessage;
@@ -325,7 +324,7 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler implem
targetHeaders.put(header, value);
}
}
return MessageBuilder.withPayload(targetPayload).copyHeaders(targetHeaders).build();
return this.getMessageBuilderFactory().withPayload(targetPayload).copyHeaders(targetHeaders).build();
}
}

View File

@@ -24,8 +24,8 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.handler.MessageProcessor;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.transformer.support.HeaderValueMessageProcessor;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
@@ -40,7 +40,7 @@ import org.springframework.messaging.MessagingException;
* @author David Turanski
* @author Artem Bilan
*/
public class HeaderEnricher implements Transformer, BeanNameAware, InitializingBean {
public class HeaderEnricher extends IntegrationObjectSupport implements Transformer, BeanNameAware, InitializingBean {
private static final Log logger = LogFactory.getLog(HeaderEnricher.class);
@@ -52,8 +52,6 @@ public class HeaderEnricher implements Transformer, BeanNameAware, InitializingB
private volatile boolean shouldSkipNulls = true;
private Object beanName;
public HeaderEnricher() {
this(null);
}
@@ -89,6 +87,12 @@ public class HeaderEnricher implements Transformer, BeanNameAware, InitializingB
this.shouldSkipNulls = shouldSkipNulls;
}
@Override
public String getComponentType() {
return "transformer"; // backwards compatibility
}
@Override
public Message<?> transform(Message<?> message) {
try {
@@ -115,7 +119,7 @@ public class HeaderEnricher implements Transformer, BeanNameAware, InitializingB
}
}
}
return MessageBuilder.withPayload(message.getPayload()).copyHeaders(headerMap).build();
return this.getMessageBuilderFactory().withPayload(message.getPayload()).copyHeaders(headerMap).build();
}
catch (Exception e) {
throw new MessagingException(message, "failed to transform message headers", e);
@@ -145,27 +149,8 @@ public class HeaderEnricher implements Transformer, BeanNameAware, InitializingB
}
}
/*
* (non-Javadoc)
*
* @see
* org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang
* .String)
*/
@Override
public void setBeanName(String beanName) {
this.beanName = beanName;
}
/*
* (non-Javadoc)
*
* @see
* org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
*/
@Override
public void afterPropertiesSet() throws Exception {
public void onInit() throws Exception {
boolean shouldOverwrite = this.defaultOverwrite;
for (HeaderValueMessageProcessor<?> processor : this.headersToAdd.values()) {
Boolean processerOverwrite = processor.isOverwrite();
@@ -174,7 +159,7 @@ public class HeaderEnricher implements Transformer, BeanNameAware, InitializingB
}
}
if (!shouldOverwrite && !this.shouldSkipNulls) {
logger.warn(this.beanName
logger.warn(this.getComponentName()
+ " is configured to not overwrite existing headers. 'shouldSkipNulls = false' will have no effect");
}
}

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,21 +16,23 @@
package org.springframework.integration.transformer;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
/**
* Transformer that removes Message headers.
*
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Gary Russell
* @since 2.0
*/
public class HeaderFilter implements Transformer {
public class HeaderFilter extends IntegrationObjectSupport implements Transformer {
private final String[] headersToRemove;
private volatile boolean patternMatch = true;
@@ -43,8 +45,9 @@ public class HeaderFilter implements Transformer {
this.patternMatch = patternMatch;
}
@Override
public Message<?> transform(Message<?> message) {
MessageBuilder<?> builder = MessageBuilder.fromMessage(message);
AbstractIntegrationMessageBuilder<?> builder = this.getMessageBuilderFactory().fromMessage(message);
if (this.patternMatch){
builder.removeHeaders(headersToRemove);
}

View File

@@ -26,9 +26,13 @@ import org.springframework.expression.Expression;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.expression.ExpressionUtils;
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.integration.expression.ExpressionUtils;
import org.springframework.util.Assert;
/**
* @author Mark Fisher
@@ -51,6 +55,8 @@ public abstract class AbstractExpressionEvaluator implements BeanFactoryAware, I
private volatile BeanFactory beanFactory;
private volatile MessageBuilderFactory messageBuilderFactory;
/**
* Specify a BeanFactory in order to enable resolution via <code>@beanName</code> in the expression.
*/
@@ -70,9 +76,24 @@ 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();
}
return this.messageBuilderFactory;
}
@Override
public void afterPropertiesSet() throws Exception {
getEvaluationContext();
if (this.messageBuilderFactory == null) {
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(beanFactory);
}
}
protected StandardEvaluationContext getEvaluationContext() {

View File

@@ -23,13 +23,13 @@ import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessageHeaders;
/**

View File

@@ -23,6 +23,7 @@ import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
@@ -52,6 +53,23 @@ public class ChannelInterceptorTests {
private final QueueChannel channel = new QueueChannel();
@Test
public void test() {
final AtomicInteger bar = new AtomicInteger();
for (int i = 0; i < 1000000000; i++) {
List<String> foo = getFoo();
if (foo.size() > 0) {
for (String baz : foo) {
bar.incrementAndGet();
}
}
}
}
protected List<String> getFoo() {
List<String> foo = new ArrayList<String>();
return foo;
}
@Test
public void testPreSendInterceptorReturnsMessage() {

View File

@@ -11,6 +11,8 @@
<int:transformer input-channel="input" output-channel="output" expression="@foo + @barString"/>
<int:transformer input-channel="fromParentToChild" output-channel="output" expression="payload.toUpperCase()" />
<int:channel id="output">
<int:queue/>
</int:channel>
@@ -28,4 +30,7 @@
</int:spel-property-accessors>
<bean id="jsonPropertyAccessor" class="org.springframework.integration.json.JsonPropertyAccessor"/>
<bean id="messageBuilderFactory" class="org.springframework.integration.support.MutableMessageBuilderFacfory" />
</beans>

View File

@@ -15,6 +15,8 @@
<int:queue/>
</int:channel>
<int:channel id="fromParentToChild" />
<bean class="org.springframework.integration.expression.ParentContextTests$Foo" />
<bean class="org.springframework.integration.expression.ParentContextTests$Foo" />

View File

@@ -15,6 +15,7 @@
*/
package org.springframework.integration.expression;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
@@ -41,10 +42,11 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.IntegrationEvaluationContextFactoryBean;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.json.JsonPathUtils;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.integration.message.MutableMessage;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.GenericMessage;
/**
* @author Gary Russell
@@ -63,6 +65,7 @@ public class ParentContextTests {
* and parent contexts work. Verifies that PropertyAccessors are inherited in the child context
* and the parent's ones are last in the propertyAccessors list of EvaluationContext.
* Verifies that SpEL functions are inherited from parent context and overridden with the same 'id'.
* Verifies that child and parent contexts can have different message builders.
*/
@Test
@SuppressWarnings("unchecked")
@@ -135,9 +138,10 @@ public class ParentContextTests {
Message<?> out = child.getBean("output", QueueChannel.class).receive(0);
assertNotNull(out);
assertEquals("foobar", out.getPayload());
child.getBean("parentIn", MessageChannel.class).send(new GenericMessage<String>("bar"));
child.getBean("parentIn", MessageChannel.class).send(new MutableMessage<String>("bar"));
out = child.getBean("parentOut", QueueChannel.class).receive(0);
assertNotNull(out);
assertThat(out, instanceOf(GenericMessage.class));
assertEquals("foo", out.getPayload());
IntegrationEvaluationContextFactoryBean evaluationContextFactoryBean =
@@ -150,6 +154,15 @@ public class ParentContextTests {
catch (Exception e) {
assertThat(e, Matchers.instanceOf(IllegalArgumentException.class));
}
parent.getBean("fromParentToChild", MessageChannel.class).send(new GenericMessage<String>("foo"));
out = child.getBean("output", QueueChannel.class).receive(0);
assertNotNull(out);
assertThat(out, instanceOf(MutableMessage.class));
assertEquals("FOO", out.getPayload());
child.close();
parent.close();
}
public static class Foo implements IntegrationEvaluationContextAware {

View File

@@ -40,11 +40,11 @@ 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.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.MessageBuilder;
/**
* @author Oleg Zhurakousky

View File

@@ -93,4 +93,4 @@ public class ExpressionEvaluatingMessageHandlerTests {
}
}
}
}

View File

@@ -0,0 +1,142 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.message;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import org.hamcrest.Matchers;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.handler.BridgeHandler;
import org.springframework.integration.message.MessageBuilderAtConfigTests.MBConfig;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.MutableMessageBuilderFacfory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gary Russell
*/
@ContextConfiguration(classes=MBConfig.class)
@RunWith(SpringJUnit4ClassRunner.class)
public class MessageBuilderAtConfigTests {
@Autowired
private MessageChannel in;
@Autowired
private PollableChannel out;
@Autowired
private MessageBuilderFactory messageBuilderFactory;
@Test
public void mutate() {
assertTrue(messageBuilderFactory instanceof MutableMessageBuilderFacfory);
in.send(new GenericMessage<String>("foo"));
Message<?> m1 = out.receive(0);
Message<?> m2 = out.receive(0);
assertThat(m1, Matchers.instanceOf(MutableMessage.class));
assertTrue(m1 == m2);
}
@Configuration
@EnableIntegration
public static class MBConfig {
@Bean
public MessageChannel in() {
return new DirectChannel();
}
@Bean
public PollableChannel out() {
return new QueueChannel();
}
@Bean
public MessageBuilderFactory messageBuilderFactory() {
return new MutableMessageBuilderFacfory();
}
@Bean
public MessageChannel pubSub() {
return new PublishSubscribeChannel();
}
@Bean
public ConsumerEndpointFactoryBean bridge1() throws Exception {
ConsumerEndpointFactoryBean factory = new ConsumerEndpointFactoryBean();
factory.setHandler(handler1());
factory.setInputChannel(in());
return factory;
}
@Bean
public BridgeHandler handler1() {
BridgeHandler handler = new BridgeHandler();
handler.setOutputChannel(pubSub());
return handler;
}
@Bean
public ConsumerEndpointFactoryBean bridge2() throws Exception {
ConsumerEndpointFactoryBean factory = new ConsumerEndpointFactoryBean();
factory.setHandler(handler2());
factory.setInputChannel(pubSub());
return factory;
}
@Bean
public BridgeHandler handler2() {
BridgeHandler handler = new BridgeHandler();
handler.setOutputChannel(out());
return handler;
}
@Bean
public ConsumerEndpointFactoryBean bridge3() throws Exception {
ConsumerEndpointFactoryBean factory = new ConsumerEndpointFactoryBean();
factory.setHandler(handler3());
factory.setInputChannel(pubSub());
return factory;
}
@Bean
public BridgeHandler handler3() {
BridgeHandler handler = new BridgeHandler();
handler.setOutputChannel(out());
return handler;
}
}
}

View File

@@ -0,0 +1,24 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
<int:channel id="in" />
<int:bridge input-channel="in" output-channel="pub" />
<int:publish-subscribe-channel id="pub" />
<int:bridge input-channel="pub" output-channel="out" />
<int:bridge input-channel="pub" output-channel="out" />
<int:channel id="out">
<int:queue/>
</int:channel>
<bean id="messageBuilderFactory" class="org.springframework.integration.support.MutableMessageBuilderFacfory" />
</beans>

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.
@@ -19,6 +19,8 @@ package org.springframework.integration.message;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotSame;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.Date;
@@ -26,17 +28,41 @@ import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import org.hamcrest.Matchers;
import org.junit.Test;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.MutableMessageBuilder;
import org.springframework.integration.support.MutableMessageBuilderFacfory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Mark Fisher
* @author Gary Russell
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class MessageBuilderTests {
@Autowired
private MessageChannel in;
@Autowired
private PollableChannel out;
@Autowired
private MessageBuilderFactory messageBuilderFactory;
@Test(expected= IllegalArgumentException.class) // priority must be an Integer
public void testPriorityHeader(){
MessageBuilder.withPayload("ha").setHeader("priority", "10").build();
@@ -119,6 +145,49 @@ public class MessageBuilderTests {
assertNotSame(message1.getHeaders().getId(), message2.getHeaders().getId());
}
@Test
public void mutate() {
assertTrue(messageBuilderFactory instanceof MutableMessageBuilderFacfory);
in.send(new GenericMessage<String>("foo"));
Message<?> m1 = out.receive(0);
Message<?> m2 = out.receive(0);
assertThat(m1, Matchers.instanceOf(MutableMessage.class));
assertTrue(m1 == m2);
}
@Test
public void mutable() {
MutableMessageBuilder<String> builder = MutableMessageBuilder.withPayload("test");
Message<String> message1 = builder
.setHeader("foo", "bar").build();
Message<String> message2 = MutableMessageBuilder.fromMessage(message1).setHeader("another", 1).build();
assertEquals("bar", message2.getHeaders().get("foo"));
assertSame(message1.getHeaders().getId(), message2.getHeaders().getId());
assertTrue(message2 == message1);
}
@Test
public void mutableFromImmutable() {
Message<String> message1 = MessageBuilder.withPayload("test")
.setHeader("foo", "bar").build();
Message<String> message2 = MutableMessageBuilder.fromMessage(message1).setHeader("another", 1).build();
assertEquals("bar", message2.getHeaders().get("foo"));
assertSame(message1.getHeaders().getId(), message2.getHeaders().getId());
assertNotSame(message1, message2);
assertFalse(message2 == message1);
}
@Test
public void mutableFromImmutableMutate() {
Message<String> message1 = MessageBuilder.withPayload("test")
.setHeader("foo", "bar").build();
Message<String> message2 = new MutableMessageBuilderFacfory().fromMessage(message1).setHeader("another", 1).build();
assertEquals("bar", message2.getHeaders().get("foo"));
assertSame(message1.getHeaders().getId(), message2.getHeaders().getId());
assertNotSame(message1, message2);
assertFalse(message2 == message1);
}
@Test
public void testPriority() {
Message<Integer> importantMessage = MessageBuilder.withPayload(1)
@@ -210,6 +279,44 @@ public class MessageBuilderTests {
assertFalse(message3.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS));
}
@Test
public void testPushAndPopSequenceDetailsMutable() throws Exception {
Message<Integer> message1 = MutableMessageBuilder.withPayload(1).pushSequenceDetails("foo", 1, 2).build();
assertFalse(message1.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS));
Message<Integer> message2 = MutableMessageBuilder.fromMessage(message1).pushSequenceDetails("bar", 1, 1).build();
assertTrue(message2.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS));
Message<Integer> message3 = MutableMessageBuilder.fromMessage(message2).popSequenceDetails().build();
assertFalse(message3.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS));
}
@Test
public void testPushAndPopSequenceDetailsWhenNoCorrelationIdMutable() throws Exception {
Message<Integer> message1 = MutableMessageBuilder.withPayload(1).build();
assertFalse(message1.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS));
Message<Integer> message2 = MutableMessageBuilder.fromMessage(message1).pushSequenceDetails("bar", 1, 1).build();
assertFalse(message2.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS));
Message<Integer> message3 = MutableMessageBuilder.fromMessage(message2).popSequenceDetails().build();
assertFalse(message3.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS));
}
@Test
public void testPopSequenceDetailsWhenNotPoppedMutable() throws Exception {
Message<Integer> message1 = MutableMessageBuilder.withPayload(1).build();
assertFalse(message1.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS));
Message<Integer> message2 = MutableMessageBuilder.fromMessage(message1).popSequenceDetails().build();
assertFalse(message2.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS));
}
@Test
public void testPushAndPopSequenceDetailsWhenNoSequenceMutable() throws Exception {
Message<Integer> message1 = MutableMessageBuilder.withPayload(1).setCorrelationId("foo").build();
assertFalse(message1.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS));
Message<Integer> message2 = MutableMessageBuilder.fromMessage(message1).pushSequenceDetails("bar", 1, 1).build();
assertTrue(message2.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS));
Message<Integer> message3 = MutableMessageBuilder.fromMessage(message2).popSequenceDetails().build();
assertFalse(message3.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS));
}
@Test
public void testNotModifiedSameMessage() throws Exception {
Message<?> original = MessageBuilder.withPayload("foo").build();

View File

@@ -29,7 +29,6 @@ import org.springframework.context.event.SmartApplicationListener;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.core.Ordered;
import org.springframework.integration.endpoint.ExpressionMessageProducerSupport;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
@@ -114,7 +113,7 @@ public class ApplicationEventListeningMessageProducer extends ExpressionMessageP
}
else {
Object payload = this.evaluatePayloadExpression(event);
this.sendMessage(MessageBuilder.withPayload(payload).build());
this.sendMessage(this.getMessageBuilderFactory().withPayload(payload).build());
}
}
}

View File

@@ -30,7 +30,6 @@ import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.metadata.MetadataStore;
import org.springframework.integration.metadata.SimpleMetadataStore;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
@@ -124,7 +123,7 @@ public class FeedEntryMessageSource extends IntegrationObjectSupport implements
if (entry == null) {
return null;
}
return MessageBuilder.withPayload(entry).build();
return this.getMessageBuilderFactory().withPayload(entry).build();
}
@Override

View File

@@ -26,14 +26,14 @@ import java.util.concurrent.PriorityBlockingQueue;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.integration.aggregator.ResequencingMessageGroupProcessor;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
import org.springframework.integration.file.filters.FileListFilter;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
/**
@@ -259,7 +259,7 @@ public class FileReadingMessageSource extends IntegrationObjectSupport implement
}
if (file != null) {
message = MessageBuilder.withPayload(file).build();
message = this.getMessageBuilderFactory().withPayload(file).build();
if (logger.isInfoEnabled()) {
logger.info("Created message: [" + message + "]");
}

View File

@@ -37,7 +37,6 @@ import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.file.support.FileExistsMode;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.util.DefaultLockRegistry;
import org.springframework.integration.util.LockRegistry;
import org.springframework.integration.util.PassThruLockRegistry;
@@ -318,7 +317,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand
if (resultFile != null) {
if (originalFileFromHeader == null && payload instanceof File) {
return MessageBuilder.withPayload(resultFile)
return this.getMessageBuilderFactory().withPayload(resultFile)
.setHeader(FileHeaders.ORIGINAL_FILE, payload);
}
}

View File

@@ -44,7 +44,6 @@ import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
@@ -407,7 +406,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
return AbstractRemoteFileOutboundGateway.this.ls(session, fullDir);
}
});
return MessageBuilder.withPayload(payload)
return this.getMessageBuilderFactory().withPayload(payload)
.setHeader(FileHeaders.REMOTE_DIRECTORY, dir)
.build();
}
@@ -425,7 +424,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
}
});
return MessageBuilder.withPayload(payload)
return this.getMessageBuilderFactory().withPayload(payload)
.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
.build();
@@ -442,7 +441,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
return AbstractRemoteFileOutboundGateway.this.mGet(requestMessage, session, remoteDir, remoteFilename);
}
});
return MessageBuilder.withPayload(payload)
return this.getMessageBuilderFactory().withPayload(payload)
.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
.build();
@@ -453,7 +452,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
String remoteFilename = this.getRemoteFilename(remoteFilePath);
String remoteDir = this.getRemoteDirectory(remoteFilePath, remoteFilename);
boolean payload = this.remoteFileTemplate.remove(remoteFilePath);
return MessageBuilder.withPayload(payload)
return this.getMessageBuilderFactory().withPayload(payload)
.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
.build();
@@ -467,7 +466,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
Assert.hasLength(remoteFileNewPath, "New filename cannot be empty");
this.remoteFileTemplate.rename(remoteFilePath, remoteFileNewPath);
return MessageBuilder.withPayload(Boolean.TRUE)
return this.getMessageBuilderFactory().withPayload(Boolean.TRUE)
.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
.setHeader(FileHeaders.RENAME_TO, remoteFileNewPath)
@@ -512,7 +511,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
List<String> replies = new ArrayList<String>();
for (File filteredFile : filteredFiles) {
if (!filteredFile.isDirectory()) {
String path = this.doPut(MessageBuilder.withPayload(filteredFile)
String path = this.doPut(this.getMessageBuilderFactory().withPayload(filteredFile)
.copyHeaders(requestMessage.getHeaders())
.build(), subDirectory);
if (path == null) {

View File

@@ -24,7 +24,6 @@ import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.file.event.FileIntegrationEvent;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
@@ -100,7 +99,7 @@ public abstract class FileTailingMessageProducerSupport extends MessageProducerS
}
protected void send(String line) {
Message<?> message = MessageBuilder.withPayload(line)
Message<?> message = this.getMessageBuilderFactory().withPayload(line)
.setHeader(FileHeaders.FILENAME, this.file.getAbsolutePath())
.build();
super.sendMessage(message);

View File

@@ -22,7 +22,8 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.transformer.Transformer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
@@ -39,6 +40,7 @@ public abstract class AbstractFilePayloadTransformer<T> implements Transformer {
private volatile boolean deleteFiles;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
/**
* Specify whether to delete the File after transformation.
@@ -50,6 +52,11 @@ 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 final Message<?> transform(Message<?> message) {
try {
@@ -59,7 +66,7 @@ public abstract class AbstractFilePayloadTransformer<T> implements Transformer {
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 = MessageBuilder.withPayload(result)
Message<?> transformedMessage = this.messageBuilderFactory.withPayload(result)
.copyHeaders(message.getHeaders())
.setHeaderIfAbsent(FileHeaders.ORIGINAL_FILE, file)
.setHeaderIfAbsent(FileHeaders.FILENAME, file.getName())

View File

@@ -22,8 +22,8 @@ import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.endpoint.ExpressionMessageProducerSupport;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
import com.gemstone.gemfire.cache.CacheClosedException;
@@ -128,10 +128,8 @@ public class CacheListeningMessageProducer extends ExpressionMessageProducerSupp
}
private void publish(Object payload) {
sendMessage(MessageBuilder.withPayload(payload).build());
sendMessage(CacheListeningMessageProducer.this.getMessageBuilderFactory().withPayload(payload).build());
}
}
}

View File

@@ -22,12 +22,12 @@ import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.data.gemfire.listener.ContinuousQueryDefinition;
import org.springframework.data.gemfire.listener.ContinuousQueryListener;
import org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer;
import org.springframework.messaging.Message;
import org.springframework.integration.endpoint.ExpressionMessageProducerSupport;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import com.gemstone.gemfire.cache.query.CqEvent;
@@ -114,7 +114,7 @@ public class ContinuousQueryMessageProducer extends ExpressionMessageProducerSup
logger.debug(String.format("processing cq event key [%s] event [%s]", event.getQueryOperation()
.toString(), event.getKey()));
}
Message<?> cqEventMessage = MessageBuilder.withPayload(evaluatePayloadExpression(event)).build();
Message<?> cqEventMessage = this.getMessageBuilderFactory().withPayload(evaluatePayloadExpression(event)).build();
sendMessage(cqEventMessage);
}
}

View File

@@ -55,7 +55,7 @@ import org.springframework.integration.http.converter.MultipartAwareFormHttpMess
import org.springframework.integration.http.multipart.MultipartHttpInputMessage;
import org.springframework.integration.http.support.DefaultHttpHeaderMapper;
import org.springframework.integration.mapping.HeaderMapper;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.json.JacksonJsonUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
@@ -485,13 +485,13 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa
}
}
MessageBuilder<?> messageBuilder = null;
AbstractIntegrationMessageBuilder<?> messageBuilder = null;
if (payload instanceof Message<?>) {
messageBuilder = MessageBuilder.fromMessage((Message<?>) payload).copyHeadersIfAbsent(headers);
messageBuilder = this.getMessageBuilderFactory().fromMessage((Message<?>) payload).copyHeadersIfAbsent(headers);
}
else {
messageBuilder = MessageBuilder.withPayload(payload).copyHeaders(headers);
messageBuilder = this.getMessageBuilderFactory().withPayload(payload).copyHeaders(headers);
}
Message<?> message = messageBuilder
@@ -519,7 +519,7 @@ public abstract class HttpRequestHandlingEndpointSupport extends MessagingGatewa
if (logger.isDebugEnabled()) {
logger.debug("Endpoint is shutting down; returning status " + HttpStatus.SERVICE_UNAVAILABLE);
}
return MessageBuilder.withPayload("Endpoint is shutting down")
return this.getMessageBuilderFactory().withPayload("Endpoint is shutting down")
.setHeader(org.springframework.integration.http.HttpHeaders.STATUS_CODE, HttpStatus.SERVICE_UNAVAILABLE)
.build();
}

View File

@@ -50,7 +50,7 @@ import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.http.support.DefaultHttpHeaderMapper;
import org.springframework.integration.mapping.HeaderMapper;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
@@ -421,15 +421,15 @@ public class HttpRequestExecutingMessageHandler extends AbstractReplyProducingMe
if (this.transferCookies) {
this.doConvertSetCookie(headers);
}
MessageBuilder<?> replyBuilder = null;
AbstractIntegrationMessageBuilder<?> replyBuilder = null;
if (httpResponse.hasBody()) {
Object responseBody = httpResponse.getBody();
replyBuilder = (responseBody instanceof Message<?>) ?
MessageBuilder.fromMessage((Message<?>) responseBody) : MessageBuilder.withPayload(responseBody);
this.getMessageBuilderFactory().fromMessage((Message<?>) responseBody) : this.getMessageBuilderFactory().withPayload(responseBody);
}
else {
replyBuilder = MessageBuilder.withPayload(httpResponse);
replyBuilder = this.getMessageBuilderFactory().withPayload(httpResponse);
}
replyBuilder.setHeader(org.springframework.integration.http.HttpHeaders.STATUS_CODE, httpResponse.getStatusCode());
return replyBuilder.copyHeaders(headers).build();

View File

@@ -18,6 +18,8 @@ package org.springframework.integration.ip.config;
import java.util.concurrent.Executor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.context.ApplicationEventPublisher;
@@ -25,6 +27,7 @@ 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;
@@ -53,7 +56,7 @@ import org.springframework.util.Assert;
* @since 2.0.5
*/
public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<AbstractConnectionFactory> implements SmartLifecycle, BeanNameAware,
ApplicationEventPublisherAware {
BeanFactoryAware, ApplicationEventPublisherAware {
private volatile AbstractConnectionFactory connectionFactory;
@@ -85,6 +88,8 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
private volatile TcpMessageMapper mapper = new TcpMessageMapper();
private volatile boolean mapperSet;
private volatile boolean singleUse;
private volatile int backlog = 5;
@@ -113,6 +118,13 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
private volatile ApplicationEventPublisher applicationEventPublisher;
private volatile BeanFactory beanFactory;
@Override
public final void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
}
@Override
public Class<?> getObjectType() {
return this.connectionFactory != null ? this.connectionFactory.getClass()
@@ -121,6 +133,9 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
@Override
protected AbstractConnectionFactory createInstance() throws Exception {
if (!this.mapperSet) {
mapper.setMessageBuilderFactory(IntegrationContextUtils.getMessageBuilderFactory(this.beanFactory));
}
if (this.usingNio) {
if ("server".equals(this.type)) {
TcpNioServerConnectionFactory connectionFactory = new TcpNioServerConnectionFactory(this.port);
@@ -137,7 +152,8 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
connectionFactory.setTcpNioConnectionSupport(this.obtainNioConnectionSupport());
this.connectionFactory = connectionFactory;
}
} else {
}
else {
if ("server".equals(this.type)) {
TcpNetServerConnectionFactory connectionFactory = new TcpNetServerConnectionFactory(this.port);
this.setCommonAttributes(connectionFactory);
@@ -356,6 +372,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
public void setMapper(TcpMessageMapper mapper) {
Assert.notNull(mapper, "TcpMessageMapper may not be null");
this.mapper = mapper;
this.mapperSet = true;
}
/**

View File

@@ -19,12 +19,12 @@ import java.util.concurrent.Executor;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.util.SimplePool;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.util.SimplePool;
/**
* @author Gary Russell
@@ -146,8 +146,9 @@ public class CachingClientConnectionFactory extends AbstractClientConnectionFact
*/
@Override
public boolean onMessage(Message<?> message) {
MessageBuilder<?> messageBuilder = MessageBuilder.fromMessage(message)
.setHeader(IpHeaders.CONNECTION_ID, this.getConnectionId());
AbstractIntegrationMessageBuilder<?> messageBuilder = CachingClientConnectionFactory.this
.getMessageBuilderFactory().fromMessage(message)
.setHeader(IpHeaders.CONNECTION_ID, this.getConnectionId());
if (message.getHeaders().get(IpHeaders.ACTUAL_CONNECTION_ID) == null) {
messageBuilder.setHeader(IpHeaders.ACTUAL_CONNECTION_ID,
message.getHeaders().get(IpHeaders.CONNECTION_ID));

View File

@@ -23,10 +23,10 @@ import java.util.concurrent.atomic.AtomicLong;
import org.springframework.core.serializer.Deserializer;
import org.springframework.core.serializer.Serializer;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
/**
@@ -354,8 +354,9 @@ public class FailoverClientConnectionFactory extends AbstractClientConnectionFac
@Override
public boolean onMessage(Message<?> message) {
if (this.delegate.getConnectionId().equals(message.getHeaders().get(IpHeaders.CONNECTION_ID))) {
MessageBuilder<?> messageBuilder = MessageBuilder.fromMessage(message)
.setHeader(IpHeaders.CONNECTION_ID, this.getConnectionId());
AbstractIntegrationMessageBuilder<?> messageBuilder = FailoverClientConnectionFactory.this
.getMessageBuilderFactory().fromMessage(message)
.setHeader(IpHeaders.CONNECTION_ID, this.getConnectionId());
if (message.getHeaders().get(IpHeaders.ACTUAL_CONNECTION_ID) == null) {
messageBuilder.setHeader(IpHeaders.ACTUAL_CONNECTION_ID,
message.getHeaders().get(IpHeaders.CONNECTION_ID));

View File

@@ -15,7 +15,7 @@
*/
package org.springframework.integration.ip.tcp.connection;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.util.Assert;
@@ -39,7 +39,7 @@ public class MessageConvertingTcpMessageMapper extends TcpMessageMapper {
Object data = connection.getPayload();
if (data != null) {
Message<?> message = this.messageConverter.toMessage(data, null);
MessageBuilder<?> messageBuilder = MessageBuilder.fromMessage(message);
AbstractIntegrationMessageBuilder<?> messageBuilder = this.getMessageBuilderFactory().fromMessage(message);
this.addStandardHeaders(connection, messageBuilder);
this.addCustomHeaders(connection, messageBuilder);
return messageBuilder.build();

View File

@@ -22,7 +22,6 @@ import java.util.Set;
import org.springframework.context.ApplicationListener;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
@@ -44,7 +43,7 @@ public class TcpConnectionEventListeningMessageProducer extends MessageProducerS
* Set the list of event types (classes that extend TcpConnectionEvent) that
* this adapter should send to the message channel. By default, all event
* types will be sent.
*
*
* @param eventTypes The event types.
*/
public void setEventTypes(Class<? extends TcpConnectionEvent>[] eventTypes) {
@@ -72,7 +71,7 @@ public class TcpConnectionEventListeningMessageProducer extends MessageProducerS
}
protected Message<TcpConnectionEvent> messageFromEvent(TcpConnectionEvent event) {
return MessageBuilder.withPayload(event).build();
return this.getMessageBuilderFactory().withPayload(event).build();
}
}

View File

@@ -24,7 +24,9 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.integration.ip.IpHeaders;
import org.springframework.integration.mapping.InboundMessageMapper;
import org.springframework.integration.mapping.OutboundMessageMapper;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
@@ -54,12 +56,45 @@ public class TcpMessageMapper implements
private volatile boolean applySequence = false;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
/**
* @param charset the charset to set
*/
public void setCharset(String charset) {
this.charset = charset;
}
/**
* Sets whether outbound String payloads are to be converted
* to byte[]. Default is true.
* @param stringToBytes The stringToBytes to set.
*/
public void setStringToBytes(boolean stringToBytes) {
this.stringToBytes = stringToBytes;
}
/**
* @param applySequence The applySequence to set.
*/
public void setApplySequence(boolean applySequence) {
this.applySequence = applySequence;
}
public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
this.messageBuilderFactory = messageBuilderFactory;
}
protected MessageBuilderFactory getMessageBuilderFactory() {
return messageBuilderFactory;
}
@Override
public Message<?> toMessage(TcpConnection connection) throws Exception {
Message<Object> message = null;
Object payload = connection.getPayload();
if (payload != null) {
MessageBuilder<Object> messageBuilder = MessageBuilder.withPayload(payload);
AbstractIntegrationMessageBuilder<Object> messageBuilder = this.messageBuilderFactory.withPayload(payload);
this.addStandardHeaders(connection, messageBuilder);
this.addCustomHeaders(connection, messageBuilder);
message = messageBuilder.build();
@@ -72,7 +107,7 @@ public class TcpMessageMapper implements
return message;
}
protected final void addStandardHeaders(TcpConnection connection, MessageBuilder<?> messageBuilder) {
protected final void addStandardHeaders(TcpConnection connection, AbstractIntegrationMessageBuilder<?> messageBuilder) {
String connectionId = connection.getConnectionId();
messageBuilder
.setHeader(IpHeaders.HOSTNAME, connection.getHostName())
@@ -86,7 +121,7 @@ public class TcpMessageMapper implements
}
}
protected final void addCustomHeaders(TcpConnection connection, MessageBuilder<?> messageBuilder) {
protected final void addCustomHeaders(TcpConnection connection, AbstractIntegrationMessageBuilder<?> messageBuilder) {
Map<String, ?> customHeaders = this.supplyCustomHeaders(connection);
if (customHeaders != null) {
messageBuilder.copyHeadersIfAbsent(customHeaders);
@@ -136,29 +171,4 @@ public class TcpMessageMapper implements
return bytes;
}
/**
* @param charset the charset to set
*/
public void setCharset(String charset) {
this.charset = charset;
}
/**
* Sets whether outbound String payloads are to be converted
* to byte[]. Default is true.
* @param stringToBytes The stringToBytes to set.
*/
public void setStringToBytes(boolean stringToBytes) {
this.stringToBytes = stringToBytes;
}
/**
* @param applySequence The applySequence to set.
*/
public void setApplySequence(boolean applySequence) {
this.applySequence = applySequence;
}
}

View File

@@ -28,7 +28,8 @@ import org.springframework.integration.ip.util.RegexUtils;
import org.springframework.integration.mapping.InboundMessageMapper;
import org.springframework.integration.mapping.MessageMappingException;
import org.springframework.integration.mapping.OutboundMessageMapper;
import org.springframework.integration.support.MessageBuilder;
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.messaging.MessageHeaders;
@@ -68,12 +69,17 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
private boolean lookupHost = true;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
private static Pattern udpHeadersPattern =
Pattern.compile(RegexUtils.escapeRegexSpecials(IpHeaders.ACK_ADDRESS) +
"=" + "([^;]*);" +
RegexUtils.escapeRegexSpecials(MessageHeaders.ID) +
"=" + "([^;]*);");
public void setMessageBuilderFactory(MessageBuilderFactory messageBuilderFactory) {
this.messageBuilderFactory = messageBuilderFactory;
}
public void setCharset(String charset) {
this.charset = charset;
@@ -201,7 +207,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 = MessageBuilder.withPayload(payload)
message = this.messageBuilderFactory.withPayload(payload)
.setHeader(IpHeaders.ACK_ID, UUID.fromString(matcher.group(2)))
.setHeader(IpHeaders.ACK_ADDRESS, matcher.group(1))
.setHeader(IpHeaders.HOSTNAME, hostName)
@@ -218,7 +224,7 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
payload = new byte[length];
System.arraycopy(packet.getData(), offset, payload, 0, length);
if (payload.length > 0) {
message = MessageBuilder.withPayload(payload)
message = this.messageBuilderFactory.withPayload(payload)
.setHeader(IpHeaders.HOSTNAME, hostName)
.setHeader(IpHeaders.IP_ADDRESS, hostAddress)
.setHeader(IpHeaders.PORT, port)

View File

@@ -57,7 +57,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
*/
public UnicastReceivingChannelAdapter(int port) {
super(port);
mapper.setLengthCheck(false);
this.mapper.setLengthCheck(false);
}
/**
@@ -69,9 +69,14 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
*/
public UnicastReceivingChannelAdapter(int port, boolean lengthCheck) {
super(port);
mapper.setLengthCheck(lengthCheck);
this.mapper.setLengthCheck(lengthCheck);
}
@Override
protected void onInit() {
super.onInit();
this.mapper.setMessageBuilderFactory(this.getMessageBuilderFactory());
}
@Override
public void run() {

View File

@@ -44,7 +44,6 @@ import org.springframework.integration.store.AbstractMessageGroupStore;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.SimpleMessageGroup;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.util.UUIDConverter;
import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.core.JdbcTemplate;
@@ -334,7 +333,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
}
final long createdDate = System.currentTimeMillis();
Message<T> result = MessageBuilder.fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE)
Message<T> result = this.getMessageBuilderFactory().fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE)
.setHeader(CREATED_DATE_KEY, new Long(createdDate)).build();
Map innerMap = (Map) new DirectFieldAccessor(result.getHeaders()).getPropertyValue("headers");

View File

@@ -21,13 +21,12 @@ import java.util.List;
import javax.sql.DataSource;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.messaging.Message;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.namedparam.SqlParameterSource;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -161,7 +160,7 @@ public class JdbcOutboundGateway extends AbstractReplyProducingMessageHandler im
if (list.size() == 1) {
payload = list.get(0);
}
return MessageBuilder.withPayload(payload).copyHeaders(requestMessage.getHeaders()).build();
return this.getMessageBuilderFactory().withPayload(payload).copyHeaders(requestMessage.getHeaders()).build();
}
/**

View File

@@ -24,10 +24,8 @@ import java.util.List;
import javax.sql.DataSource;
import org.springframework.dao.DataAccessException;
import org.springframework.messaging.Message;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jdbc.core.ColumnMapRowMapper;
import org.springframework.jdbc.core.JdbcOperations;
import org.springframework.jdbc.core.ResultSetExtractor;
@@ -36,6 +34,7 @@ import org.springframework.jdbc.core.RowMapperResultSetExtractor;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcOperations;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.jdbc.core.namedparam.SqlParameterSource;
import org.springframework.messaging.Message;
/**
* A polling channel adapter that creates messages from the payload returned by
@@ -147,7 +146,7 @@ public class JdbcPollingChannelAdapter extends IntegrationObjectSupport implemen
if (payload == null) {
return null;
}
return MessageBuilder.withPayload(payload).build();
return this.getMessageBuilderFactory().withPayload(payload).build();
}
/**

View File

@@ -20,7 +20,6 @@ import java.sql.CallableStatement;
import java.util.Map;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
@@ -78,7 +77,7 @@ public class StoredProcOutboundGateway extends AbstractReplyProducingMessageHand
}
return MessageBuilder.withPayload(payload).copyHeaders(requestMessage.getHeaders()).build();
return this.getMessageBuilderFactory().withPayload(payload).copyHeaders(requestMessage.getHeaders()).build();
}

View File

@@ -20,7 +20,6 @@ import java.util.Map;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
@@ -65,7 +64,7 @@ public class StoredProcPollingChannelAdapter extends IntegrationObjectSupport im
if (payload == null) {
return null;
}
return MessageBuilder.withPayload(payload).build();
return this.getMessageBuilderFactory().withPayload(payload).build();
}
/**

View File

@@ -50,7 +50,6 @@ import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.SimpleMessageGroup;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.transaction.TransactionSynchronizationFactory;
import org.springframework.integration.util.UUIDConverter;
import org.springframework.jdbc.core.JdbcOperations;
@@ -400,7 +399,7 @@ public class JdbcChannelMessageStore extends AbstractMessageGroupStore implement
final String groupKey = getKey(groupId);
final long createdDate = System.currentTimeMillis();
final Message<?> result = MessageBuilder.fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE)
final Message<?> result = this.getMessageBuilderFactory().fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE)
.setHeader(CREATED_DATE_KEY, new Long(createdDate)).build();
final Map innerMap = (Map) new DirectFieldAccessor(result.getHeaders()).getPropertyValue("headers");

View File

@@ -32,9 +32,11 @@ import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.history.TrackableComponent;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.jms.listener.SessionAwareMessageListener;
import org.springframework.jms.support.JmsUtils;
import org.springframework.jms.support.converter.MessageConverter;
@@ -88,6 +90,8 @@ public class ChannelPublishingJmsMessageListener
private volatile BeanFactory beanFactory;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
/**
* Specify whether a JMS reply Message is expected.
*
@@ -317,8 +321,8 @@ public class ChannelPublishingJmsMessageListener
Map<String, Object> headers = headerMapper.toHeaders(jmsMessage);
Message<?> requestMessage = (result instanceof Message<?>) ?
MessageBuilder.fromMessage((Message<?>) result).copyHeaders(headers).build() :
MessageBuilder.withPayload(result).copyHeaders(headers).build();
this.messageBuilderFactory.fromMessage((Message<?>) result).copyHeaders(headers).build() :
this.messageBuilderFactory.withPayload(result).copyHeaders(headers).build();
if (!this.expectReply) {
this.gatewayDelegate.send(requestMessage);
}
@@ -357,6 +361,7 @@ public class ChannelPublishingJmsMessageListener
this.gatewayDelegate.setBeanFactory(this.beanFactory);
}
this.gatewayDelegate.afterPropertiesSet();
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(this.beanFactory);
}
protected void start(){

View File

@@ -22,7 +22,7 @@ import javax.jms.Destination;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.messaging.Message;
@@ -102,8 +102,9 @@ public class JmsDestinationPollingSource extends IntegrationObjectSupport implem
Map<String, Object> mappedHeaders = this.headerMapper.toHeaders(jmsMessage);
MessageConverter converter = this.jmsTemplate.getMessageConverter();
Object convertedObject = converter.fromMessage(jmsMessage);
MessageBuilder<Object> builder = (convertedObject instanceof Message)
? MessageBuilder.fromMessage((Message<Object>) convertedObject) : MessageBuilder.withPayload(convertedObject);
AbstractIntegrationMessageBuilder<Object> builder = (convertedObject instanceof Message) ?
this.getMessageBuilderFactory().fromMessage((Message<Object>) convertedObject) :
this.getMessageBuilderFactory().withPayload(convertedObject);
convertedMessage = builder.copyHeadersIfAbsent(mappedHeaders).build();
}
catch (Exception e) {

View File

@@ -48,7 +48,6 @@ import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.MessageTimeoutException;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jms.connection.ConnectionFactoryUtils;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.support.JmsUtils;
@@ -645,7 +644,7 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
if (!this.initialized) {
this.afterPropertiesSet();
}
final Message<?> requestMessage = MessageBuilder.fromMessage(message).build();
final Message<?> requestMessage = this.getMessageBuilderFactory().fromMessage(message).build();
try {
javax.jms.Message jmsReply;
if (this.replyContainer == null) {
@@ -678,10 +677,10 @@ public class JmsOutboundGateway extends AbstractReplyProducingMessageHandler imp
}
Message<?> replyMessage = null;
if (result instanceof Message){
replyMessage = MessageBuilder.fromMessage((Message<?>) result).copyHeaders(jmsReplyHeaders).build();
replyMessage = this.getMessageBuilderFactory().fromMessage((Message<?>) result).copyHeaders(jmsReplyHeaders).build();
}
else {
replyMessage = MessageBuilder.withPayload(result).copyHeaders(jmsReplyHeaders).build();
replyMessage = this.getMessageBuilderFactory().withPayload(result).copyHeaders(jmsReplyHeaders).build();
}
return replyMessage;
}

View File

@@ -16,10 +16,9 @@
package org.springframework.integration.jms;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jms.core.JmsTemplate;
/**
* @author Mark Fisher
@@ -59,7 +58,7 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh
replyMessage = (Message<?>) object;
}
else {
replyMessage = MessageBuilder.withPayload(object).build();
replyMessage = this.getMessageBuilderFactory().withPayload(object).build();
}
return this.getInterceptors().postReceive(replyMessage, this) ;
}

View File

@@ -30,7 +30,7 @@ import org.springframework.integration.dispatcher.BroadcastingDispatcher;
import org.springframework.integration.dispatcher.MessageDispatcher;
import org.springframework.integration.dispatcher.RoundRobinLoadBalancingStrategy;
import org.springframework.integration.dispatcher.UnicastingDispatcher;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.listener.AbstractMessageListenerContainer;
import org.springframework.messaging.Message;
@@ -92,7 +92,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
this.configureDispatcher(isPubSub);
MessageListener listener = new DispatchingMessageListener(
this.getJmsTemplate(), this.dispatcher,
this, isPubSub);
this, isPubSub,this.getMessageBuilderFactory());
this.container.setMessageListener(listener);
if (!this.container.isActive()) {
this.container.afterPropertiesSet();
@@ -131,13 +131,17 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
private final boolean isPubSub;
private final MessageBuilderFactory messageBuilderFactory;
private DispatchingMessageListener(JmsTemplate jmsTemplate,
MessageDispatcher dispatcher, SubscribableJmsChannel channel, boolean isPubSub) {
MessageDispatcher dispatcher, SubscribableJmsChannel channel, boolean isPubSub,
MessageBuilderFactory messageBuilderFactory) {
this.jmsTemplate = jmsTemplate;
this.dispatcher = dispatcher;
this.channel = channel;
this.isPubSub = isPubSub;
this.messageBuilderFactory = messageBuilderFactory;
}
@@ -148,7 +152,7 @@ public class SubscribableJmsChannel extends AbstractJmsChannel implements Subscr
Object converted = this.jmsTemplate.getMessageConverter().fromMessage(message);
if (converted != null) {
messageToSend = (converted instanceof Message<?>) ? (Message<?>) converted
: MessageBuilder.withPayload(converted).build();
: this.messageBuilderFactory.withPayload(converted).build();
this.dispatcher.dispatch(messageToSend);
}
else if (this.logger.isWarnEnabled()) {

View File

@@ -36,7 +36,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
@@ -115,7 +115,7 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport
if (logger.isInfoEnabled()) {
logger.info("received notification: " + notification + ", and handback: " + handback);
}
MessageBuilder<?> builder = MessageBuilder.withPayload(notification);
AbstractIntegrationMessageBuilder<?> builder = this.getMessageBuilderFactory().withPayload(notification);
if (handback != null) {
builder.setHeader(JmxHeaders.NOTIFICATION_HANDBACK, handback);
}

View File

@@ -15,11 +15,10 @@
*/
package org.springframework.integration.jpa.inbound;
import org.springframework.messaging.Message;
import org.springframework.integration.context.IntegrationObjectSupport;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.jpa.core.JpaExecutor;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
@@ -87,7 +86,7 @@ public class JpaPollingChannelAdapter extends IntegrationObjectSupport implement
return null;
}
return MessageBuilder.withPayload(payload).build();
return this.getMessageBuilderFactory().withPayload(payload).build();
}
@Override

View File

@@ -19,7 +19,6 @@ package org.springframework.integration.jpa.outbound;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.jpa.core.JpaExecutor;
import org.springframework.integration.jpa.support.OutboundGatewayType;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
@@ -82,7 +81,7 @@ public class JpaOutboundGateway extends AbstractReplyProducingMessageHandler {
return null;
}
return MessageBuilder.withPayload(result).copyHeaders(requestMessage.getHeaders()).build();
return this.getMessageBuilderFactory().withPayload(result).copyHeaders(requestMessage.getHeaders()).build();
}

View File

@@ -36,7 +36,6 @@ import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.mail.event.MailIntegrationEvent;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.transaction.IntegrationResourceHolder;
import org.springframework.integration.transaction.TransactionSynchronizationFactory;
import org.springframework.scheduling.TaskScheduler;
@@ -247,7 +246,7 @@ public class ImapIdleChannelAdapter extends MessageProducerSupport implements Be
@Override
public void run() {
org.springframework.messaging.Message<?> message =
MessageBuilder.withPayload(mailMessage).build();
ImapIdleChannelAdapter.this.getMessageBuilderFactory().withPayload(mailMessage).build();
if (TransactionSynchronizationManager.isActualTransactionActive()) {
if (transactionSynchronizationFactory != null){

View File

@@ -22,10 +22,15 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
/**
@@ -38,7 +43,8 @@ import org.springframework.util.Assert;
* @author Gary Russell
* @author Oleg Zhurakousky
*/
public class MailReceivingMessageSource implements MessageSource<javax.mail.Message> {
public class MailReceivingMessageSource implements MessageSource<javax.mail.Message>,
BeanFactoryAware {
private final Log logger = LogFactory.getLog(this.getClass());
@@ -46,12 +52,31 @@ public class MailReceivingMessageSource implements MessageSource<javax.mail.Mess
private final Queue<javax.mail.Message> mailQueue = new ConcurrentLinkedQueue<javax.mail.Message>();
private volatile BeanFactory beanFactory;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
public MailReceivingMessageSource(MailReceiver mailReceiver) {
Assert.notNull(mailReceiver, "mailReceiver must not be null");
this.mailReceiver = mailReceiver;
}
@Override
public final void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(this.beanFactory);
}
protected BeanFactory getBeanFactory() {
return beanFactory;
}
protected MessageBuilderFactory getMessageBuilderFactory() {
return messageBuilderFactory;
}
@Override
public Message<javax.mail.Message> receive() {
try {
javax.mail.Message mailMessage = this.mailQueue.poll();
@@ -66,7 +91,7 @@ public class MailReceivingMessageSource implements MessageSource<javax.mail.Mess
if (logger.isDebugEnabled()) {
logger.debug("received mail message [" + mailMessage + "]");
}
return MessageBuilder.withPayload(mailMessage).build();
return this.messageBuilderFactory.withPayload(mailMessage).build();
}
}
catch (Exception e) {

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.
@@ -25,25 +25,47 @@ import javax.mail.Message.RecipientType;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.mail.MailHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.transformer.MessageTransformationException;
import org.springframework.integration.transformer.Transformer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
/**
* Base class for Transformers that convert from a JavaMail Message to a
* Spring Integration Message.
*
*
* @author Mark Fisher
* @author Gary Russell
*/
public abstract class AbstractMailMessageTransformer<T> implements Transformer {
public abstract class AbstractMailMessageTransformer<T> implements Transformer,
BeanFactoryAware {
protected final Log logger = LogFactory.getLog(this.getClass());
private volatile BeanFactory beanFactory;
private volatile MessageBuilderFactory messageBuilderFactory = new DefaultMessageBuilderFactory();
@Override
public final void setBeanFactory(BeanFactory beanFactory) {
this.beanFactory = beanFactory;
this.messageBuilderFactory = IntegrationContextUtils.getMessageBuilderFactory(this.beanFactory);
}
protected MessageBuilderFactory getMessageBuilderFactory() {
return messageBuilderFactory;
}
@Override
public Message<?> transform(Message<?> message) {
Object payload = message.getPayload();
if (!(payload instanceof javax.mail.Message)) {
@@ -51,7 +73,7 @@ public abstract class AbstractMailMessageTransformer<T> implements Transformer {
+ " requires a javax.mail.Message payload");
}
javax.mail.Message mailMessage = (javax.mail.Message) payload;
MessageBuilder<T> builder = null;
AbstractIntegrationMessageBuilder<T> builder = null;
try {
builder = this.doTransform(mailMessage);
}
@@ -65,7 +87,7 @@ public abstract class AbstractMailMessageTransformer<T> implements Transformer {
return builder.build();
}
protected abstract MessageBuilder<T> doTransform(javax.mail.Message mailMessage) throws Exception;
protected abstract AbstractIntegrationMessageBuilder<T> doTransform(javax.mail.Message mailMessage) throws Exception;
private Map<String, Object> extractHeaderMapFromMailMessage(javax.mail.Message mailMessage) {

View File

@@ -21,7 +21,7 @@ import java.nio.charset.Charset;
import javax.mail.Multipart;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.util.Assert;
/**
@@ -31,6 +31,7 @@ import org.springframework.util.Assert;
* an output stream of bytes using the provided charset (or UTF-8 by default).
*
* @author Mark Fisher
* @author Gary Russell
*/
public class MailToStringTransformer extends AbstractMailMessageTransformer<String> {
@@ -50,15 +51,15 @@ public class MailToStringTransformer extends AbstractMailMessageTransformer<Stri
}
@Override
protected MessageBuilder<String> doTransform(javax.mail.Message mailMessage) throws Exception {
protected AbstractIntegrationMessageBuilder<String> doTransform(javax.mail.Message mailMessage) throws Exception {
Object content = mailMessage.getContent();
if (content instanceof String) {
return MessageBuilder.withPayload((String) content);
return this.getMessageBuilderFactory().withPayload((String) content);
}
if (content instanceof Multipart) {
ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
((Multipart) content).writeTo(outputStream);
return MessageBuilder.withPayload(
return this.getMessageBuilderFactory().withPayload(
new String(outputStream.toByteArray(), this.charset));
}
throw new IllegalArgumentException("failed to transform contentType ["

Some files were not shown because too many files have changed in this diff Show More