diff --git a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java index eeda4f9f51..41430437b7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationMessageHeaderAccessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2013 the original author or authors. + * Copyright 2013-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. @@ -13,9 +13,11 @@ * See the License for the specific language governing permissions and * limitations under the License. */ + package org.springframework.integration; import java.util.Date; +import java.util.Map; import org.springframework.messaging.Message; import org.springframework.messaging.support.MessageHeaderAccessor; @@ -26,6 +28,7 @@ import org.springframework.util.Assert; * Adds standard SI Headers. * * @author Andy Wilkinson + * @author Artem Bilan * @since 4.0 * */ @@ -45,6 +48,8 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor { public static final String POSTPROCESS_RESULT = "postProcessResult"; + public static final String ROUTING_SLIP = "routingSlip"; + public IntegrationMessageHeaderAccessor(Message message) { super(message); } @@ -93,14 +98,16 @@ public class IntegrationMessageHeaderAccessor extends MessageHeaderAccessor { + "' header value must be a Date or Long."); } else if (IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER.equals(headerName) - || IntegrationMessageHeaderAccessor.SEQUENCE_SIZE.equals(headerName)) { + || IntegrationMessageHeaderAccessor.SEQUENCE_SIZE.equals(headerName) + || IntegrationMessageHeaderAccessor.PRIORITY.equals(headerName)) { Assert.isTrue(Integer.class.isAssignableFrom(headerValue.getClass()), "The '" + headerName + "' header value must be an Integer."); } - else if (IntegrationMessageHeaderAccessor.PRIORITY.equals(headerName)) { - Assert.isTrue(Integer.class.isAssignableFrom(headerValue.getClass()), "The '" + headerName - + "' header value must be an Integer."); + else if (IntegrationMessageHeaderAccessor.ROUTING_SLIP.equals(headerName)) { + Assert.isTrue(Map.class.isAssignableFrom(headerValue.getClass()), "The '" + headerName + + "' header value must be an List."); } } } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index cb59ad23f8..b10f882834 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -53,7 +53,6 @@ import org.springframework.integration.util.UUIDConverter; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; -import org.springframework.messaging.MessagingException; import org.springframework.messaging.core.DestinationResolutionException; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; @@ -655,7 +654,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP this.verifyResultCollectionConsistsOfMessages((Collection) result); partialSequence = (Collection>) result; } - this.sendReplies(result, message); + this.sendOutputs(result, message); return partialSequence; } @@ -665,59 +664,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP "The expected collection of Messages contains non-Message element: " + commonElementType); } - @SuppressWarnings("rawtypes") - protected void sendReplies(Object processorResult, Message message) { - Object replyChannelHeader = null; - if (message != null) { - replyChannelHeader = message.getHeaders().getReplyChannel(); - } - - Object replyChannel = getOutputChannel(); - if (replyChannel == null) { - replyChannel = replyChannelHeader; - } - Assert.notNull(replyChannel, "no outputChannel or replyChannel header available"); - if (processorResult instanceof Iterable && shouldSendMultipleReplies((Iterable) processorResult)) { - for (Object next : (Iterable) processorResult) { - this.sendReplyMessage(next, replyChannel); - } - } - else { - this.sendReplyMessage(processorResult, replyChannel); - } - } - - protected void sendReplyMessage(Object reply, Object replyChannel) { - if (replyChannel instanceof MessageChannel) { - if (reply instanceof Message) { - this.messagingTemplate.send((MessageChannel) replyChannel, (Message) reply); - } - else { - this.messagingTemplate.convertAndSend((MessageChannel) replyChannel, reply); - } - } - else if (replyChannel instanceof String) { - if (reply instanceof Message) { - this.messagingTemplate.send((String) replyChannel, (Message) reply); - } - else { - this.messagingTemplate.convertAndSend((String) replyChannel, reply); - } - } - else { - throw new MessagingException("replyChannel must be a MessageChannel or String"); - } - } - - protected boolean shouldSendMultipleReplies(Iterable iter) { - for (Object next : iter) { - if (next instanceof Message) { - return true; - } - } - return false; - } - protected Long obtainGroupTimeout(MessageGroup group) { return this.groupTimeoutExpression != null ? this.groupTimeoutExpression.getValue(this.evaluationContext, group, Long.class) : null; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/HeaderEnricherParserSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/HeaderEnricherParserSupport.java index 4b744495df..3ff0f0ba26 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/HeaderEnricherParserSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/HeaderEnricherParserSupport.java @@ -16,6 +16,7 @@ package org.springframework.integration.config.xml; +import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -27,13 +28,16 @@ import org.w3c.dom.NodeList; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.TypedStringValue; import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.ManagedList; import org.springframework.beans.factory.support.ManagedMap; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.expression.DynamicExpression; import org.springframework.integration.transformer.HeaderEnricher; import org.springframework.integration.transformer.support.ExpressionEvaluatingHeaderValueMessageProcessor; import org.springframework.integration.transformer.support.MessageProcessingHeaderValueMessageProcessor; +import org.springframework.integration.transformer.support.RoutingSlipHeaderValueMessageProcessor; import org.springframework.integration.transformer.support.StaticHeaderValueMessageProcessor; import org.springframework.util.ClassUtils; import org.springframework.util.StringUtils; @@ -228,10 +232,20 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar parserContext.getReaderContext().error( "The 'method' attribute cannot be used with the 'value' attribute.", element); } - Object headerValue = (headerType != null) ? - new TypedStringValue(value, headerType) : value; - valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition(StaticHeaderValueMessageProcessor.class) - .addConstructorArgValue(headerValue); + if (IntegrationMessageHeaderAccessor.ROUTING_SLIP.equals(headerName)) { + List routingSlipPath = new ManagedList(); + routingSlipPath.addAll(Arrays.asList(StringUtils.tokenizeToStringArray(value, ";"))); + valueProcessorBuilder = + BeanDefinitionBuilder.genericBeanDefinition(RoutingSlipHeaderValueMessageProcessor.class) + .addConstructorArgValue(routingSlipPath); + } + else { + Object headerValue = (headerType != null) ? + new TypedStringValue(value, headerType) : value; + valueProcessorBuilder = + BeanDefinitionBuilder.genericBeanDefinition(StaticHeaderValueMessageProcessor.class) + .addConstructorArgValue(headerValue); + } } else if (isExpression) { if (hasMethod) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/StandardHeaderEnricherParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/StandardHeaderEnricherParser.java index 8197a9149c..0c557bc12b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/StandardHeaderEnricherParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/StandardHeaderEnricherParser.java @@ -16,6 +16,8 @@ package org.springframework.integration.config.xml; +import java.util.Map; + import org.w3c.dom.Element; import org.springframework.beans.factory.support.BeanDefinitionBuilder; @@ -43,10 +45,12 @@ public class StandardHeaderEnricherParser extends HeaderEnricherParserSupport { this.addElementToHeaderMapping("correlation-id", IntegrationMessageHeaderAccessor.CORRELATION_ID); this.addElementToHeaderMapping("expiration-date", IntegrationMessageHeaderAccessor.EXPIRATION_DATE, Long.class); this.addElementToHeaderMapping("priority", IntegrationMessageHeaderAccessor.PRIORITY, Integer.class); + this.addElementToHeaderMapping("routing-slip", IntegrationMessageHeaderAccessor.ROUTING_SLIP, Map.class); } @Override - protected void postProcessHeaderEnricher(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) { + protected void postProcessHeaderEnricher(BeanDefinitionBuilder builder, Element element, + ParserContext parserContext) { String ref = element.getAttribute("ref"); String method = element.getAttribute("method"); if (StringUtils.hasText(ref) || StringUtils.hasText(method)) { @@ -56,7 +60,8 @@ public class StandardHeaderEnricherParser extends HeaderEnricherParserSupport { parserContext.extractSource(element)); return; } - BeanDefinitionBuilder processorBuilder = BeanDefinitionBuilder.genericBeanDefinition(MethodInvokingMessageProcessor.class); + BeanDefinitionBuilder processorBuilder = + BeanDefinitionBuilder.genericBeanDefinition(MethodInvokingMessageProcessor.class); processorBuilder.addConstructorArgReference(ref); processorBuilder.addConstructorArgValue(method); builder.addPropertyValue("messageProcessor", processorBuilder.getBeanDefinition()); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java index d23065eeac..f92b3f3153 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageProducingHandler.java @@ -16,12 +16,25 @@ package org.springframework.integration.handler; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; + import org.springframework.beans.BeansException; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessagingTemplate; +import org.springframework.integration.routingslip.RoutingSlipRouteStrategy; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; +import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.MessagingException; import org.springframework.messaging.core.DestinationResolutionException; +import org.springframework.messaging.core.DestinationResolver; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * The base {@link AbstractMessageHandler} implementation for the {@link MessageProducer}. @@ -57,6 +70,15 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan this.outputChannelName = outputChannelName; } + /** + * Set the DestinationResolver<MessageChannel> to be used when there is no default output channel. + * @param channelResolver The channel resolver. + */ + public void setChannelResolver(DestinationResolver channelResolver) { + Assert.notNull(channelResolver, "'channelResolver' must not be null"); + this.messagingTemplate.setDestinationResolver(channelResolver); + } + @Override protected void onInit() throws Exception { super.onInit(); @@ -72,7 +94,8 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan synchronized (this) { if (this.outputChannelName != null) { try { - Assert.state(getBeanFactory() != null, "A bean factory is required to resolve the outputChannel at runtime."); + Assert.state(getBeanFactory() != null, + "A bean factory is required to resolve the outputChannel at runtime."); this.outputChannel = getBeanFactory().getBean(this.outputChannelName, MessageChannel.class); this.outputChannelName = null; } @@ -86,4 +109,167 @@ public abstract class AbstractMessageProducingHandler extends AbstractMessageHan return outputChannel; } + protected void sendOutputs(Object result, Message requestMessage) { + if (result instanceof Iterable && shouldSplitOutput((Iterable) result)) { + for (Object o : (Iterable) result) { + this.produceOutput(o, requestMessage); + } + } + else if (result != null) { + this.produceOutput(result, requestMessage); + } + } + + protected boolean shouldSplitOutput(Iterable reply) { + for (Object next : reply) { + if (next instanceof Message || next instanceof AbstractIntegrationMessageBuilder) { + return true; + } + } + return false; + } + + protected void produceOutput(Object reply, Message requestMessage) { + MessageHeaders requestHeaders = requestMessage.getHeaders(); + + Object replyChannel = null; + if (getOutputChannel() == null) { + Map routingSlipHeader = requestHeaders.get(IntegrationMessageHeaderAccessor.ROUTING_SLIP, Map.class); + if (routingSlipHeader != null) { + Assert.isTrue(routingSlipHeader.size() == 1, "The RoutingSlip header value must be a SingletonMap"); + Object key = routingSlipHeader.keySet().iterator().next(); + Object value = routingSlipHeader.values().iterator().next(); + Assert.isInstanceOf(List.class, key, "The RoutingSlip key must be List"); + Assert.isInstanceOf(Integer.class, value, "The RoutingSlip value must be Integer"); + List routingSlip = (List) key; + AtomicInteger routingSlipIndex = new AtomicInteger((Integer) value); + replyChannel = getOutputChannelFromRoutingSlip(reply, requestMessage, routingSlip, routingSlipIndex); + if (replyChannel != null) { + //TODO Migrate to the SF MessageBuilder + AbstractIntegrationMessageBuilder builder = null; + if (reply instanceof Message) { + builder = this.getMessageBuilderFactory().fromMessage((Message) reply); + } + else if (reply instanceof AbstractIntegrationMessageBuilder) { + builder = (AbstractIntegrationMessageBuilder) reply; + } + else { + builder = this.getMessageBuilderFactory().withPayload(reply); + } + builder.setHeader(IntegrationMessageHeaderAccessor.ROUTING_SLIP, + Collections.singletonMap(routingSlip, routingSlipIndex.get())); + reply = builder; + } + } + + if (replyChannel == null) { + replyChannel = requestHeaders.getReplyChannel(); + } + } + + Message replyMessage = createOutputMessage(reply, requestHeaders); + sendOutput(replyMessage, replyChannel); + } + + private Object getOutputChannelFromRoutingSlip(Object reply, Message requestMessage, List routingSlip, + AtomicInteger routingSlipIndex) { + if (routingSlipIndex.get() >= routingSlip.size()) { + return null; + } + + Object path = routingSlip.get(routingSlipIndex.get()); + Object routingSlipPathValue = null; + + if (path instanceof String) { + routingSlipPathValue = getBeanFactory().getBean((String) path); + } + else if (path instanceof RoutingSlipRouteStrategy) { + routingSlipPathValue = path; + } + else { + throw new IllegalArgumentException("The RoutingSlip 'path' can be of " + + "String or RoutingSlipRouteStrategy type, but gotten: " + path); + } + + if (routingSlipPathValue instanceof MessageChannel) { + routingSlipIndex.incrementAndGet(); + return routingSlipPathValue; + } + else { + String nextPath = ((RoutingSlipRouteStrategy) routingSlipPathValue).getNextPath(requestMessage, reply); + if (StringUtils.hasText(nextPath)) { + return nextPath; + } + else { + routingSlipIndex.incrementAndGet(); + return getOutputChannelFromRoutingSlip(reply, requestMessage, routingSlip, routingSlipIndex); + } + } + } + + private Message createOutputMessage(Object output, MessageHeaders requestHeaders) { + AbstractIntegrationMessageBuilder builder = null; + if (output instanceof Message) { + if (!this.shouldCopyRequestHeaders()) { + return (Message) output; + } + builder = this.getMessageBuilderFactory().fromMessage((Message) output); + } + else if (output instanceof AbstractIntegrationMessageBuilder) { + builder = (AbstractIntegrationMessageBuilder) output; + } + else { + builder = this.getMessageBuilderFactory().withPayload(output); + } + if (this.shouldCopyRequestHeaders()) { + builder.copyHeadersIfAbsent(requestHeaders); + } + return builder.build(); + } + + /** + * Send an output Message. The 'replyChannel' will be considered only if this handler's + * 'outputChannel' is null. In that case, the 'replyChannel' value must not also be + * null, and it must be an instance of either String or {@link MessageChannel}. + * @param output the output object to send + * @param replyChannel the 'replyChannel' value from the original request + */ + private void sendOutput(Object output, Object replyChannel) { + MessageChannel outputChannel = getOutputChannel(); + if (outputChannel != null) { + replyChannel = outputChannel; + } + if (replyChannel == null) { + throw new DestinationResolutionException("no output-channel or replyChannel header available"); + } + + if (replyChannel instanceof MessageChannel) { + if (output instanceof Message) { + this.messagingTemplate.send((MessageChannel) replyChannel, (Message) output); + } + else { + this.messagingTemplate.convertAndSend((MessageChannel) replyChannel, output); + } + } + else if (replyChannel instanceof String) { + if (output instanceof Message) { + this.messagingTemplate.send((String) replyChannel, (Message) output); + } + else { + this.messagingTemplate.convertAndSend((String) replyChannel, output); + } + } + else { + throw new MessagingException("replyChannel must be a MessageChannel or String"); + } + } + + /** + * Subclasses may override this. True by default. + * @return true if the request headers should be copied. + */ + protected boolean shouldCopyRequestHeaders() { + return true; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java index 03f3da8dff..6c8850fbe2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java @@ -22,13 +22,7 @@ import org.aopalliance.aop.Advice; import org.springframework.aop.framework.ProxyFactory; import org.springframework.beans.factory.BeanClassLoaderAware; -import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessageDeliveryException; -import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.core.DestinationResolutionException; -import org.springframework.messaging.core.DestinationResolver; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.CollectionUtils; @@ -63,15 +57,6 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa this.requiresReply = requiresReply; } - /** - * Set the DestinationResolver<MessageChannel> to be used when there is no default output channel. - * @param channelResolver The channel resolver. - */ - public void setChannelResolver(DestinationResolver channelResolver) { - Assert.notNull(channelResolver, "'channelResolver' must not be null"); - this.messagingTemplate.setDestinationResolver(channelResolver); - } - public void setAdviceChain(List adviceChain) { Assert.notNull(adviceChain, "adviceChain cannot be null"); @@ -98,7 +83,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa } this.advisedRequestHandler = (RequestHandler) proxyFactory.getProxy(this.beanClassLoader); } - this.doInit(); + doInit(); } protected void doInit() { @@ -111,18 +96,17 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa protected final void handleMessageInternal(Message message) { Object result; if (this.advisedRequestHandler == null) { - result = this.handleRequestMessage(message); + result = handleRequestMessage(message); } else { result = doInvokeAdvisedRequestHandler(message); } if (result != null) { - MessageHeaders requestHeaders = message.getHeaders(); - this.handleResult(result, requestHeaders); + sendOutputs(result, message); } else if (this.requiresReply) { throw new ReplyRequiredException(message, "No reply produced by handler '" + - this.getComponentName() + "', and its 'requiresReply' property is set to true."); + getComponentName() + "', and its 'requiresReply' property is set to true."); } else if (logger.isDebugEnabled()) { logger.debug("handler '" + this + "' produced no reply for request Message: " + message); @@ -133,102 +117,6 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa return this.advisedRequestHandler.handleRequestMessage(message); } - private void handleResult(Object result, MessageHeaders requestHeaders) { - if (result instanceof Iterable && this.shouldSplitReply((Iterable) result)) { - for (Object o : (Iterable) result) { - this.produceReply(o, requestHeaders); - } - } - else if (result != null) { - this.produceReply(result, requestHeaders); - } - } - - protected void produceReply(Object reply, MessageHeaders requestHeaders) { - Message replyMessage = this.createReplyMessage(reply, requestHeaders); - this.sendReplyMessage(replyMessage, requestHeaders.getReplyChannel()); - } - - private Message createReplyMessage(Object reply, MessageHeaders requestHeaders) { - AbstractIntegrationMessageBuilder builder = null; - if (reply instanceof Message) { - if (!this.shouldCopyRequestHeaders()) { - return (Message) reply; - } - builder = this.getMessageBuilderFactory().fromMessage((Message) reply); - } - else if (reply instanceof AbstractIntegrationMessageBuilder) { - builder = (AbstractIntegrationMessageBuilder) reply; - } - else { - builder = this.getMessageBuilderFactory().withPayload(reply); - } - if (this.shouldCopyRequestHeaders()) { - builder.copyHeadersIfAbsent(requestHeaders); - } - return builder.build(); - } - - /** - * Send a reply Message. The 'replyChannelHeaderValue' will be considered only if this handler's - * 'outputChannel' is null. In that case, the header value must not also be - * null, and it must be an instance of either String or {@link MessageChannel}. - * @param replyMessage the reply Message to send - * @param replyChannelHeaderValue the 'replyChannel' header value from the original request - */ - private void sendReplyMessage(Message replyMessage, Object replyChannelHeaderValue) { - if (logger.isDebugEnabled()) { - logger.debug("handler '" + this + "' sending reply Message: " + replyMessage); - } - - MessageChannel outputChannel = getOutputChannel(); - if (outputChannel != null) { - this.sendMessage(replyMessage, outputChannel); - } - else if (replyChannelHeaderValue != null) { - this.sendMessage(replyMessage, replyChannelHeaderValue); - } - else { - throw new DestinationResolutionException("no output-channel or replyChannel header available"); - } - } - - /** - * Send the message to the given channel. The channel must be a String or - * {@link MessageChannel} instance, never null. - * @param message The message. - * @param channel The channel to which to send the message. - */ - private void sendMessage(final Message message, final Object channel) { - if (channel instanceof MessageChannel) { - this.messagingTemplate.send((MessageChannel) channel, message); - } - else if (channel instanceof String) { - this.messagingTemplate.send((String) channel, message); - } - else { - throw new MessageDeliveryException(message, - "a non-null reply channel value of type MessageChannel or String is required"); - } - } - - private boolean shouldSplitReply(Iterable reply) { - for (Object next : reply) { - if (next instanceof Message || next instanceof AbstractIntegrationMessageBuilder) { - return true; - } - } - return false; - } - - /** - * Subclasses may override this. True by default. - * @return true if the request headers should be copied. - */ - protected boolean shouldCopyRequestHeaders() { - return true; - } - /** * Subclasses must implement this method to handle the request Message. The return * value may be a Message, a MessageBuilder, or any plain Object. The base class diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java index 254d8874b3..0c6042f132 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java @@ -21,16 +21,12 @@ import java.util.List; import java.util.concurrent.locks.ReentrantLock; import org.springframework.beans.DirectFieldAccessor; -import org.springframework.beans.factory.BeanFactory; import org.springframework.context.Lifecycle; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.filter.MessageFilter; -import org.springframework.integration.support.channel.BeanFactoryChannelResolver; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.MessageHandlingException; -import org.springframework.messaging.core.DestinationResolver; import org.springframework.util.Assert; /** @@ -67,21 +63,10 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Artem Bilan */ -public class MessageHandlerChain extends AbstractMessageHandler implements MessageProducer, Lifecycle { +public class MessageHandlerChain extends AbstractMessageProducingHandler implements MessageProducer, Lifecycle { private volatile List handlers; - private volatile MessageChannel outputChannel; - - /** - * If the sendTimeout is configured explicitly on this chain instance, it will - * take precedence over the actual settings on the final handler in the chain. - * By default, it is null, so the actual handler configuration is used. - */ - private volatile Long sendTimeout = null; - - private volatile DestinationResolver channelResolver; - private volatile boolean initialized; private final Object initializationMonitor = new Object(); @@ -94,15 +79,6 @@ public class MessageHandlerChain extends AbstractMessageHandler implements Messa this.handlers = handlers; } - @Override - public void setOutputChannel(MessageChannel outputChannel) { - this.outputChannel = outputChannel; - } - - public void setSendTimeout(long sendTimeout) { - this.sendTimeout = sendTimeout; - } - @Override public String getComponentType() { return "chain"; @@ -110,14 +86,11 @@ public class MessageHandlerChain extends AbstractMessageHandler implements Messa @Override protected void onInit() throws Exception { + super.onInit(); synchronized (this.initializationMonitor) { if (!this.initialized) { Assert.notEmpty(this.handlers, "handler list must not be empty"); this.configureChain(); - BeanFactory beanFactory = this.getBeanFactory(); - if (this.channelResolver == null && beanFactory != null) { - this.channelResolver = new BeanFactoryChannelResolver(beanFactory); - } this.initialized = true; } } @@ -137,7 +110,7 @@ public class MessageHandlerChain extends AbstractMessageHandler implements Messa for (int i = 0; i < this.handlers.size(); i++) { MessageHandler handler = handlers.get(i); if (i < handlers.size() - 1) { // not the last handler - Assert.isTrue(handler instanceof MessageProducer, "All handlers except for " + + Assert.isInstanceOf(MessageProducer.class, handler, "All handlers except for " + "the last one in the chain must implement the MessageProducer interface."); final MessageHandler nextHandler = handlers.get(i + 1); final MessageChannel nextChannel = new MessageChannel() { @@ -165,13 +138,18 @@ public class MessageHandlerChain extends AbstractMessageHandler implements Messa ((MessageProducer) handler).setOutputChannel(replyChannel); } else { - Assert.isNull(this.outputChannel, + Assert.isNull(getOutputChannel(), "An output channel was provided, but the final handler in " + "the chain does not implement the MessageProducer interface."); } } } + @Override + protected boolean shouldCopyRequestHeaders() { + return false; + } + /** * SmartLifecycle implementation (delegates to the {@link #handlers}) */ @@ -252,35 +230,15 @@ public class MessageHandlerChain extends AbstractMessageHandler implements Messa @Override public boolean send(Message message) { - return this.send(message, -1); + produceOutput(message, message); + return true; } @Override public boolean send(Message message, long timeout) { - timeout = (MessageHandlerChain.this.sendTimeout != null) - ? MessageHandlerChain.this.sendTimeout : timeout; - if (MessageHandlerChain.this.outputChannel != null) { - return MessageHandlerChain.this.outputChannel.send(message, timeout); - } - Object replyChannelHeader = message.getHeaders().getReplyChannel(); - if (replyChannelHeader == null) { - throw new MessageHandlingException(message, "no replyChannel header available"); - } - MessageChannel replyChannel = null; - if (replyChannelHeader instanceof MessageChannel) { - replyChannel = (MessageChannel) replyChannelHeader; - } - else if (replyChannelHeader instanceof String) { - Assert.notNull(channelResolver, "ChannelResolver is required"); - replyChannel = channelResolver.resolveDestination((String) replyChannelHeader); - } - else { - throw new MessageHandlingException(message, - "invalid replyChannel type [" + replyChannelHeader.getClass() + "]"); - } - return (timeout >= 0) ? replyChannel.send(message, timeout) - : replyChannel.send(message); + return send(message); } + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/routingslip/ExpressionEvaluatingRoutingSlipRouteStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/routingslip/ExpressionEvaluatingRoutingSlipRouteStrategy.java new file mode 100644 index 0000000000..c743e40d53 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/routingslip/ExpressionEvaluatingRoutingSlipRouteStrategy.java @@ -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.routingslip; + +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.expression.ExpressionParser; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.expression.IntegrationEvaluationContextAware; +import org.springframework.messaging.Message; + +/** + * The {@link Expression} based {@link RoutingSlipRouteStrategy} implementation. + * The {@code requestMessage} and {@code reply} object are wrapped + * to the {@link RequestAndReply} which is used as a {@link EvaluationContext} {@code rootObject}. + * This is necessary to avoid a creation of a new {@link EvaluationContext} on each invocation + * when additional parameter can be populated as expression variable, but {@link EvaluationContext} + * isn't thread-safe. + *

+ * The {@link ExpressionEvaluatingRoutingSlipRouteStrategy} can be used directly as a regular bean + * in the {@code ApplicationContext} and its {@code beanName} can be used from {@code routingSlip} + * header configuration. + *

+ * Usage of {@link ExpressionEvaluatingRoutingSlipRouteStrategy} as a regular bean definition is + * a recommended way in case of distributed environment, when message with {@code routingSlip} + * header can be sent across the network. One of this case is a {@code QueueChannel} with + * persistent {@code MessageStore}, when {@link ExpressionEvaluatingRoutingSlipRouteStrategy} + * instance as a header value will be non-serializable. + *

+ * This class is used internally from {@code RoutingSlipHeaderValueMessageProcessor} + * to populate {@code routingSlip} header value item, when the {@code value} + * from configuration contains expression definitions: + * 

+ * {@code
+ * 
+ *     
+ * 
+ * }
+ *
+ * + * @author Artem Bilan + * @since 4.1 + */ +public class ExpressionEvaluatingRoutingSlipRouteStrategy + implements RoutingSlipRouteStrategy, IntegrationEvaluationContextAware { + + private static final ExpressionParser PARSER = new SpelExpressionParser(); + + private final Expression expression; + + private EvaluationContext evaluationContext; + + public ExpressionEvaluatingRoutingSlipRouteStrategy(String expression) { + this(PARSER.parseExpression(expression)); + } + + public ExpressionEvaluatingRoutingSlipRouteStrategy(Expression expression) { + this.expression = expression; + } + + @Override + public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) { + this.evaluationContext = evaluationContext; + } + + @Override + public String getNextPath(Message requestMessage, Object reply) { + return this.expression.getValue(this.evaluationContext, new RequestAndReply(requestMessage, reply), + String.class); + } + + public static class RequestAndReply { + + private final Message request; + + private final Object reply; + + + RequestAndReply(Message request, Object reply) { + this.request = request; + this.reply = reply; + } + + public Message getRequest() { + return request; + } + + public Object getReply() { + return reply; + } + + } + + @Override + public String toString() { + return "ExpressionEvaluatingRoutingSlipRouteStrategy for: [" + this.expression.getExpressionString() + "]"; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/routingslip/RoutingSlipRouteStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/routingslip/RoutingSlipRouteStrategy.java new file mode 100644 index 0000000000..7995550be0 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/routingslip/RoutingSlipRouteStrategy.java @@ -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.routingslip; + +import org.springframework.messaging.Message; + +/** + * The {@code RoutingSlip} strategy to determine the next {@code replyChannel}. + * + * @author Artem Bilan + * @since 4.1 + */ +public interface RoutingSlipRouteStrategy { + + String getNextPath(Message requestMessage, Object reply); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/routingslip/package-info.java b/spring-integration-core/src/main/java/org/springframework/integration/routingslip/package-info.java new file mode 100644 index 0000000000..13ec34ed51 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/routingslip/package-info.java @@ -0,0 +1,4 @@ +/** + * Provides classes supporting the RoutingSlip pattern. + */ +package org.springframework.integration.routingslip; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java index 3a0e4b4e28..4294431442 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java @@ -22,14 +22,14 @@ import java.util.Collections; import java.util.Iterator; import java.util.concurrent.atomic.AtomicInteger; -import reactor.function.Function; - import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.util.FunctionIterator; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; +import reactor.function.Function; + /** * Base class for Message-splitting handlers. * @@ -101,12 +101,11 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess }); } - @SuppressWarnings( { "unchecked", "rawtypes" }) - private AbstractIntegrationMessageBuilder createBuilder(Object item, MessageHeaders headers, Object correlationId, - int sequenceNumber, int sequenceSize) { - AbstractIntegrationMessageBuilder builder; + private AbstractIntegrationMessageBuilder createBuilder(Object item, MessageHeaders headers, + Object correlationId, int sequenceNumber, int sequenceSize) { + AbstractIntegrationMessageBuilder builder; if (item instanceof Message) { - builder = this.getMessageBuilderFactory().fromMessage((Message) item); + builder = this.getMessageBuilderFactory().fromMessage((Message) item); } else { builder = this.getMessageBuilderFactory().withPayload(item); @@ -119,10 +118,15 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess } @Override - protected void produceReply(Object result, MessageHeaders requestHeaders) { + protected boolean shouldCopyRequestHeaders() { + return false; + } + + @Override + protected void produceOutput(Object result, Message requestMessage) { Iterator iterator = (Iterator) result; while (iterator.hasNext()) { - super.produceReply(iterator.next(), requestHeaders); + super.produceOutput(iterator.next(), requestMessage); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/support/RoutingSlipHeaderValueMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/support/RoutingSlipHeaderValueMessageProcessor.java new file mode 100644 index 0000000000..88322a65e2 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/support/RoutingSlipHeaderValueMessageProcessor.java @@ -0,0 +1,102 @@ +/* + * 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.transformer.support; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Map; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.expression.EvaluationContext; +import org.springframework.integration.expression.IntegrationEvaluationContextAware; +import org.springframework.integration.routingslip.ExpressionEvaluatingRoutingSlipRouteStrategy; +import org.springframework.integration.routingslip.RoutingSlipRouteStrategy; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; + +/** + * The {@code RoutingSlip} {@link HeaderValueMessageProcessor} specific implementation. + * Accepts the {@code routingSlipPath} array, checks each of them against + * {@link BeanFactory} on the first {@link #processMessage} invocation. + * Converts those items, which aren't beans in the application context, to the + * {@link ExpressionEvaluatingRoutingSlipRouteStrategy} and return a {@code singletonMap} + * with the {@code path} as {@code key} and {@code 0} as initial {@code routingSlipIndex}. + * + * @author Artem Bilan + * @since 4.1 + */ +public class RoutingSlipHeaderValueMessageProcessor + extends AbstractHeaderValueMessageProcessor, Integer>> + implements BeanFactoryAware, IntegrationEvaluationContextAware { + + private final List routingSlipPath; + + private EvaluationContext evaluationContext; + + private volatile Map, Integer> routingSlip; + + private BeanFactory beanFactory; + + public RoutingSlipHeaderValueMessageProcessor(String... routingSlipPath) { + Assert.notNull(routingSlipPath); + Assert.noNullElements(routingSlipPath); + this.routingSlipPath = Arrays.asList(routingSlipPath); + } + + @Override + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + this.beanFactory = beanFactory; + } + + @Override + public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) { + this.evaluationContext = evaluationContext; + } + + @Override + public Map, Integer> processMessage(Message message) { + if (this.routingSlip == null) { + synchronized (this) { + if (this.routingSlip == null) { + List routingSlipValues = new ArrayList(this.routingSlipPath.size()); + for (String path : this.routingSlipPath) { + if (this.beanFactory.containsBean(path)) { + Object bean = this.beanFactory.getBean(path); + Assert.state(bean instanceof MessageChannel || bean instanceof RoutingSlipRouteStrategy, + "The RoutingSlip can contain only bean names of MessageChannel or " + + "RoutingSlipRouteStrategy: " + bean); + routingSlipValues.add(path); + } + else { + ExpressionEvaluatingRoutingSlipRouteStrategy strategy = new + ExpressionEvaluatingRoutingSlipRouteStrategy(path); + strategy.setIntegrationEvaluationContext(this.evaluationContext); + routingSlipValues.add(strategy); + } + } + this.routingSlip = Collections.singletonMap(Collections.unmodifiableList(routingSlipValues), 0); + } + } + } + return this.routingSlip; + } +} diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd index dd888a5cac..a591a88fc4 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd @@ -1995,13 +1995,31 @@ - + Shortcut to specify a value for the 'priority' header. + + + + The 'RoutingSlip' header. semicolon-separated value of message channel names, + or bean names of 'org.springframework.integration.routingslip.RoutingSlipRouteStrategy' + or SpEL expression like 'request.headers[nextPathHeader]'. + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests.java index 09e1781bf7..5d1ad8804b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests.java @@ -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. @@ -278,7 +278,8 @@ public class ChainParserTests { @Test // INT-1165 public void chainWithSendTimeout() { - long sendTimeout = TestUtils.getPropertyValue(this.chainWithSendTimeout, "sendTimeout", Long.class); + long sendTimeout = TestUtils.getPropertyValue(this.chainWithSendTimeout, "messagingTemplate.sendTimeout", + Long.class); assertEquals(9876, sendTimeout); } @@ -343,7 +344,7 @@ public class ChainParserTests { assertEquals(256, chainEndpoint.getPhase()); MessageHandlerChain handlerChain = ctx.getBean("chain.handler", MessageHandlerChain.class); - assertEquals(3000L, TestUtils.getPropertyValue(handlerChain, "sendTimeout")); + assertEquals(3000L, TestUtils.getPropertyValue(handlerChain, "messagingTemplate.sendTimeout")); assertEquals(false, TestUtils.getPropertyValue(handlerChain, "running")); //INT-3108 MessageHandler serviceActivator = ctx.getBean("chain$child.sa-within-chain.handler", MessageHandler.class); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests-context.xml index 172dd91ed7..6ff811df6b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests-context.xml @@ -76,6 +76,16 @@ + + + + + + + + + +
diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests.java index 1bdfdcc0f0..0cfa142ade 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/HeaderEnricherTests.java @@ -16,13 +16,16 @@ package org.springframework.integration.config.xml; +import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import java.util.Collections; import java.util.Date; +import java.util.List; import java.util.Map; import org.junit.Test; @@ -34,6 +37,7 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.core.MessagingTemplate; +import org.springframework.integration.routingslip.ExpressionEvaluatingRoutingSlipRouteStrategy; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.transformer.MessageTransformationException; import org.springframework.messaging.Message; @@ -119,7 +123,7 @@ public class HeaderEnricherTests { assertNotNull(result); Object correlationId = new IntegrationMessageHeaderAccessor(result).getCorrelationId(); assertEquals(Long.class, correlationId.getClass()); - assertEquals(new Long(123), correlationId); + assertEquals(123L, correlationId); } @Test @@ -128,7 +132,7 @@ public class HeaderEnricherTests { MessageChannel channel = context.getBean("correlationIdRefInput", MessageChannel.class); Message result = template.sendAndReceive(channel, new GenericMessage("test")); assertNotNull(result); - assertEquals(new Integer(123), new IntegrationMessageHeaderAccessor(result).getCorrelationId()); + assertEquals(123, new IntegrationMessageHeaderAccessor(result).getCorrelationId()); } @Test @@ -162,7 +166,8 @@ public class HeaderEnricherTests { public void priorityExpression() { MessagingTemplate template = new MessagingTemplate(); MessageChannel channel = context.getBean("priorityExpressionInput", MessageChannel.class); - Message result = template.sendAndReceive(channel, new GenericMessage>(Collections.singletonMap("priority", "-10"))); + Message result = template.sendAndReceive(channel, + new GenericMessage>(Collections.singletonMap("priority", "-10"))); assertNotNull(result); assertEquals(new Integer(-10), new IntegrationMessageHeaderAccessor(result).getPriority()); } @@ -205,7 +210,7 @@ public class HeaderEnricherTests { Message result = template.sendAndReceive(channel, new GenericMessage("test")); assertNotNull(result); assertEquals(Long.class, result.getHeaders().get("number").getClass()); - assertEquals(new Long(12345), result.getHeaders().get("number")); + assertEquals(12345L, result.getHeaders().get("number")); } @Test @@ -252,9 +257,29 @@ public class HeaderEnricherTests { @Test(expected = BeanDefinitionParsingException.class) public void testFailConfigUnexpectedSubElement() { - new ClassPathXmlApplicationContext("HeaderEnricherWithUnexpectedSubElementForHeader-fail-context.xml", this.getClass()); + new ClassPathXmlApplicationContext("HeaderEnricherWithUnexpectedSubElementForHeader-fail-context.xml", + this.getClass()); } + @Test + public void testRoutingSlip() { + MessagingTemplate template = new MessagingTemplate(); + MessageChannel channel = context.getBean("routingSlipInput", MessageChannel.class); + Message result = template.sendAndReceive(channel, new GenericMessage("test")); + assertNotNull(result); + Object routingSlip = new IntegrationMessageHeaderAccessor(result) + .getHeader(IntegrationMessageHeaderAccessor.ROUTING_SLIP); + assertNotNull(routingSlip); + assertThat(routingSlip, instanceOf(Map.class)); + @SuppressWarnings("unchecked") + List routingSlipPath = (List) ((Map) routingSlip).keySet().iterator().next(); + + assertEquals("fooChannel", routingSlipPath.get(0)); + assertThat(routingSlipPath.get(1), instanceOf(ExpressionEvaluatingRoutingSlipRouteStrategy.class)); + assertEquals("bazRoutingSlip", routingSlipPath.get(2)); + } + + public static class TestBean { private final String name; @@ -274,9 +299,8 @@ public class HeaderEnricherTests { TestBean testBean = (TestBean) o; - if (name != null ? !name.equals(testBean.name) : testBean.name != null) return false; + return !(name != null ? !name.equals(testBean.name) : testBean.name != null); - return true; } @Override diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java index 2a19dd7496..4545935034 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java @@ -86,7 +86,7 @@ public class MessageHandlerChainTests { chain.setHandlers(handlers); chain.setOutputChannel(outputChannel); chain.handleMessage(message); - Mockito.verify(outputChannel).send(Mockito.eq(message), Mockito.eq(-1L)); + Mockito.verify(outputChannel).send(Mockito.eq(message)); } @Test(expected = IllegalArgumentException.class) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/routingslip/RoutingSlipTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/routingslip/RoutingSlipTests-context.xml new file mode 100644 index 0000000000..f6f82c87b3 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/routingslip/RoutingSlipTests-context.xml @@ -0,0 +1,64 @@ + + + + + channel1 + request.headers[myRoutingSlipChannel] + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/routingslip/RoutingSlipTests.java b/spring-integration-core/src/test/java/org/springframework/integration/routingslip/RoutingSlipTests.java new file mode 100644 index 0000000000..e33b2da099 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/routingslip/RoutingSlipTests.java @@ -0,0 +1,107 @@ +/* + * 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.routingslip; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.history.MessageHistory; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.PollableChannel; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Artem Bilan + * @since 4.1 + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class RoutingSlipTests { + + @Autowired + private MessageChannel input; + + @Test + @SuppressWarnings("unchecked") + public void testRoutingSlip() { + PollableChannel replyChannel = new QueueChannel(); + Message> request = MessageBuilder.withPayload(Arrays.asList("test1", "test2")) + .setReplyChannel(replyChannel) + .setHeader("myRoutingSlipChannel", "channel5").build(); + this.input.send(request); + Message reply = replyChannel.receive(10000); + assertNotNull(reply); + List> messages = (List>) reply.getPayload(); + for (Message message : messages) { + Map, Integer> routingSlip = message.getHeaders() + .get(IntegrationMessageHeaderAccessor.ROUTING_SLIP, Map.class); + assertEquals(routingSlip.keySet().iterator().next().size(), routingSlip.values().iterator().next().intValue()); + MessageHistory messageHistory = MessageHistory.read(message); + List channelNames = Arrays.asList("input", "split", "process", "channel1", "channel2", + "channel3", "channel4", "channel5", "aggregate"); + int i = 0; + for (Properties properties : messageHistory) { + assertTrue(channelNames.contains(properties.getProperty("name"))); + } + } + } + + public static class TestRoutingSlipRoutePojo { + + final String[] channels = {"channel2", "channel3"}; + + private int i = 0; + + public String get(Message requestMessage, Object reply) { + try { + return this.channels[i++]; + } + catch (Exception e) { + return null; + } + } + + } + + public static class TestRoutingSlipRouteStrategy implements RoutingSlipRouteStrategy { + + private AtomicBoolean invoked = new AtomicBoolean(); + + @Override + public String getNextPath(Message requestMessage, Object reply) { + return !invoked.getAndSet(true) ? "channel4" : null; + } + + } + +} diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml index 2f60e04b60..947f5c8a28 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml @@ -8,7 +8,7 @@ xmlns:tx="http://www.springframework.org/schema/tx" xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd - http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd + http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd"> @@ -36,9 +36,9 @@ - @@ -69,4 +69,8 @@ + + + + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java index 2f82ed5831..25f2978d9a 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java @@ -13,12 +13,15 @@ package org.springframework.integration.jdbc; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.instanceOf; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import java.io.NotSerializableException; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; @@ -31,8 +34,11 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.core.serializer.support.SerializationFailedException; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.store.MessageGroup; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.annotation.Repeat; @@ -71,6 +77,9 @@ public class JdbcMessageStoreChannelIntegrationTests { @Autowired private PlatformTransactionManager transactionManager; + @Autowired + private MessageChannel routingSlip; + @Before public void clear() { Service.reset(1); @@ -220,6 +229,21 @@ public class JdbcMessageStoreChannelIntegrationTests { } + @Test + public void testWithRoutingSlip() { + try { + this.routingSlip.send(new GenericMessage("foo")); + fail("MessageDeliveryException expected"); + } + catch (Exception e) { + assertThat(e, instanceOf(MessageDeliveryException.class)); + assertThat(e.getCause(), instanceOf(SerializationFailedException.class)); + assertThat(e.getCause().getCause(), instanceOf(NotSerializableException.class)); + assertThat(e.getMessage(), + containsString("org.springframework.integration.routingslip.ExpressionEvaluatingRoutingSlipRouteStrategy")); + } + } + public static class Service { private static boolean fail = false; diff --git a/src/reference/docbook/content-enrichment.xml b/src/reference/docbook/content-enrichment.xml index 7d00b7c43a..fc223fdf03 100644 --- a/src/reference/docbook/content-enrichment.xml +++ b/src/reference/docbook/content-enrichment.xml @@ -63,18 +63,22 @@ + ]]> In the above configuration you can clearly see that for well-known headers such as errorChannel, correlationId, - priority, replyChanneletc., instead of - using generic <header> sub-elements where + priority, replyChannel, routing-slip etc., + instead of using generic <header> sub-elements where you would have to provide both header 'name' and 'value', you can use convenient sub-elements to set those values directly. - + + Starting with version 4.1 the Header Enricher + provides routing-slip sub-element. See for more information. + POJO Support diff --git a/src/reference/docbook/router.xml b/src/reference/docbook/router.xml index 8ca426fa4c..d4de630660 100644 --- a/src/reference/docbook/router.xml +++ b/src/reference/docbook/router.xml @@ -1022,5 +1022,11 @@ public List route(@Header("orderStatus") OrderStatus status)]]> +
+ Routing Slip + + TBD + +
diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index d0baec2766..0741657b9e 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -34,6 +34,13 @@ See for more information. +
+ Routing Slip Pattern + + The Routing Slip EIP pattern implementation is now provided. + See for more information. + +
BoonJsonObjectMapper