From 836c8e25568caa4e5ab52688b3cdf191eaacf13a Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 15 Nov 2013 15:21:44 +0200 Subject: [PATCH] INT-2269 Add ReplyChannelRegistry Allows reply channel resolution after a message has been serialized somewhere in a flow. Previously, the reply channel was lost. With this change, the reply channel can be registered and the header becomes a string (channel name) that can be serialized. JIRA: https://jira.springsource.org/browse/INT-2269 INT-2269 Rename Registry to HeaderChannelRegistry - Extract interface - DefaultHeaderChannelRgistry INT-2269 Polishing; PR Comments - Add errorChannel registration as well. INT-2269 HeaderChannelRegistry - Fix The internal BridgeHandler in MessagingGatewaySupport did not have a channel resolver. When a reply was explicitly routed to the gateway's reply channel, the String representation of the reply channel could not be resolved to a channel. Set the BeanFactory on the bridge handler. Add tests. INT-2269 HeaderChannelRegistry - Fix JMS/Enricher The JMS inbound gateway and ContentEnricher instantiate a MessagingGatewaySupport internally it does not get a reference to the BeanFactory. This means that the internal BridgeHandler cannot resolve the String representation of the reply channel to a channel. Make ChannelPublishingJmsMessageListener BeanFactory aware, and propagate the bean factory to the MGS. Set the MGS bean factory in the ContentEnricher (which is already BFA). INT-2269: Polishing --- .../DefaultHeaderChannelRegistry.java | 249 +++++++++++++++ .../registry/HeaderChannelRegistry.java | 63 ++++ .../AbstractIntegrationNamespaceHandler.java | 39 ++- ...actPollingInboundChannelAdapterParser.java | 1 - .../xml/HeaderEnricherParserSupport.java | 284 ++++++++++-------- .../context/IntegrationContextUtils.java | 3 + .../gateway/MessagingGatewaySupport.java | 22 +- .../channel/BeanFactoryChannelResolver.java | 38 ++- .../transformer/ContentEnricher.java | 7 + .../config/xml/spring-integration-3.0.xsd | 9 + .../HeaderChannelRegistryTests-context.xml | 69 +++++ .../registry/HeaderChannelRegistryTests.java | 167 ++++++++++ .../config/xml/ControlBusTests.java | 31 +- .../xml/EnricherParserTests-context.xml | 10 +- .../config/xml/EnricherParserTests.java | 9 +- .../ChannelPublishingJmsMessageListener.java | 71 +++-- ...waySerializedReplyChannelTests-context.xml | 37 +++ .../GatewaySerializedReplyChannelTests.java | 55 ++++ src/reference/docbook/content-enrichment.xml | 51 +++- src/reference/docbook/message-store.xml | 20 +- src/reference/docbook/whats-new.xml | 10 + 21 files changed, 1060 insertions(+), 185 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/registry/DefaultHeaderChannelRegistry.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/registry/HeaderChannelRegistry.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/channel/registry/HeaderChannelRegistryTests-context.xml create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/channel/registry/HeaderChannelRegistryTests.java create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/GatewaySerializedReplyChannelTests-context.xml create mode 100644 spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/GatewaySerializedReplyChannelTests.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/DefaultHeaderChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/DefaultHeaderChannelRegistry.java new file mode 100644 index 0000000000..3aa7050153 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/DefaultHeaderChannelRegistry.java @@ -0,0 +1,249 @@ +/* + * Copyright 2013 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.channel.registry; + +import java.util.Date; +import java.util.Iterator; +import java.util.Map; +import java.util.Map.Entry; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.atomic.AtomicLong; + +import org.springframework.context.SmartLifecycle; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.context.IntegrationObjectSupport; +import org.springframework.integration.support.channel.BeanFactoryChannelResolver; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.util.Assert; + +/** + * Converts a channel to a name, retaining a reference to the channel keyed by the name. + * Allows a downstream {@link BeanFactoryChannelResolver} to find the channel by name + * in the event that the flow serialized the message at some point. + * Channels are expired after a configurable delay (60 seconds by default). + * The actual average expiry time will be 1.5x the delay. + * + * @author Gary Russell + * @since 3.0 + * + */ +public class DefaultHeaderChannelRegistry extends IntegrationObjectSupport + implements HeaderChannelRegistry, SmartLifecycle, Runnable { + + private static final int DEFAULT_REAPER_DELAY = 60000; + + private final Map channels = new ConcurrentHashMap(); + + private static final AtomicLong id = new AtomicLong(); + + private final String uuid = UUID.randomUUID().toString() + ":"; + + private volatile long reaperDelay; + + private volatile ScheduledFuture reaperScheduledFuture; + + private volatile boolean running; + + private volatile int phase; + + private volatile boolean autoStartup = true; + + /** + * Constructs a registry with the default delay for channel expiry. + */ + public DefaultHeaderChannelRegistry() { + this(DEFAULT_REAPER_DELAY); + } + + /** + * Constructs a registry with the provided delay (milliseconds) for + * channel expiry. + * + * @param reaperDelay the delay in milliseconds. + */ + public DefaultHeaderChannelRegistry(long reaperDelay) { + this.setReaperDelay(reaperDelay); + } + + /** + * Set the reaper delay. + * + * @param reaperDelay the delay in milliseconds. + */ + public final void setReaperDelay(long reaperDelay) { + Assert.isTrue(reaperDelay > 0, "'reaperDelay' must be > 0"); + this.reaperDelay = reaperDelay; + } + + public final long getReaperDelay() { + return reaperDelay; + } + + @Override + public void setTaskScheduler(TaskScheduler taskScheduler) { + super.setTaskScheduler(taskScheduler); + } + + @Override + public int getPhase() { + return this.phase; + } + + public final void setPhase(int phase) { + this.phase = phase; + } + + @Override + public boolean isAutoStartup() { + return this.autoStartup; + } + + public final void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + @Override + public final int size() { + return this.channels.size(); + } + + @Override + protected void onInit() throws Exception { + super.onInit(); + Assert.notNull(this.getTaskScheduler(), "a task scheduler is required"); + } + + @Override + public synchronized void start() { + if (!this.running) { + Assert.notNull(this.getTaskScheduler(), "a task scheduler is required"); + this.reaperScheduledFuture = this.getTaskScheduler().schedule(this, + new Date(System.currentTimeMillis() + this.reaperDelay)); + this.running = true; + } + } + + @Override + public synchronized void stop() { + this.running = false; + if (this.reaperScheduledFuture != null) { + this.reaperScheduledFuture.cancel(true); + } + } + + @Override + public void stop(Runnable callback) { + this.stop(); + callback.run(); + } + + @Override + public boolean isRunning() { + return this.running; + } + + @Override + public Object channelToChannelName(Object channel) { + if (channel != null && channel instanceof MessageChannel) { + String name = this.uuid + DefaultHeaderChannelRegistry.id.incrementAndGet(); + channels.put(name, new MessageChannelWrapper((MessageChannel) channel)); + if (logger.isDebugEnabled()) { + logger.debug("Registered " + channel + " as " + name); + } + return name; + } + else { + return channel; + } + } + + @Override + public MessageChannel channelNameToChannel(String name) { + if (name != null) { + MessageChannelWrapper messageChannelWrapper = this.channels.get(name); + if (logger.isDebugEnabled() && messageChannelWrapper != null) { + logger.debug("Retrieved " + messageChannelWrapper.getChannel() + " with " + name); + } + return messageChannelWrapper == null ? null : messageChannelWrapper.getChannel(); + } + return null; + } + + /** + * Cancel the scheduled reap task and run immediately; then reschedule. + */ + @Override + public void runReaper() { + synchronized(this) { + this.reaperScheduledFuture.cancel(false); + this.reaperScheduledFuture = null; + } + this.run(); + } + + @Override + public void run() { + this.reaperScheduledFuture = null; + if (logger.isTraceEnabled()) { + logger.trace("Reaper started; channels size=" + this.channels.size()); + } + Iterator> iterator = this.channels.entrySet().iterator(); + long threshold = System.currentTimeMillis() - this.reaperDelay; + while (iterator.hasNext()) { + Entry entry = iterator.next(); + if (entry.getValue().getCreated() < threshold) { + if (logger.isDebugEnabled()) { + logger.debug("Expiring " + entry.getKey() + " (" + entry.getValue().getChannel() + ")"); + } + iterator.remove(); + } + } + synchronized (this) { + if (this.reaperScheduledFuture == null) { + this.reaperScheduledFuture = this.getTaskScheduler().schedule(this, + new Date(System.currentTimeMillis() + this.reaperDelay)); + } + } + if (logger.isTraceEnabled()) { + logger.trace("Reaper completed; channels size=" + this.channels.size()); + } + } + + + private class MessageChannelWrapper { + + private final MessageChannel channel; + + private final long created; + + private MessageChannelWrapper(MessageChannel channel) { + this.channel = channel; + this.created = System.currentTimeMillis(); + } + + public final long getCreated() { + return created; + } + + public final MessageChannel getChannel() { + return channel; + } + + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/HeaderChannelRegistry.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/HeaderChannelRegistry.java new file mode 100644 index 0000000000..66fd2eb118 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/registry/HeaderChannelRegistry.java @@ -0,0 +1,63 @@ +/* + * Copyright 2013 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.channel.registry; + +import org.springframework.integration.MessageChannel; +import org.springframework.integration.support.channel.BeanFactoryChannelResolver; +import org.springframework.jmx.export.annotation.ManagedAttribute; +import org.springframework.jmx.export.annotation.ManagedOperation; + +/** + * Implementations convert a channel to a name, retaining a reference to the channel keyed by the name. + * Allows a downstream {@link BeanFactoryChannelResolver} to find the channel by name in + * the event that the flow serialized the message at some point. + * + * @author Gary Russell + * @since 3.0 + * + */ +public interface HeaderChannelRegistry { + + /** + * Converts the channel to a name (String). If the channel is not a + * {@link MessageChannel}, it is returned unchanged. + * + * @param channel The channel. + * @return The channel name, or the channel if it is not a MessageChannel. + */ + public abstract Object channelToChannelName(Object channel); + + /** + * Converts the channel name back to a {@link MessageChannel} (if it is + * registered). + * @param name The name of the channel. + * @return The channel, or null if there is no channel registered with the name. + */ + public abstract MessageChannel channelNameToChannel(String name); + + /** + * @return the current size of the registry + */ + @ManagedAttribute + public abstract int size(); + + /** + * Cancel the scheduled reap task and run immediately; then reschedule. + */ + @ManagedOperation(description = "Cancel the scheduled reap task and run immediately; then reschedule.") + public abstract void runReaper(); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractIntegrationNamespaceHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractIntegrationNamespaceHandler.java index 104d17e7e4..dd65c4337f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractIntegrationNamespaceHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractIntegrationNamespaceHandler.java @@ -16,8 +16,6 @@ package org.springframework.integration.config.xml; -import static org.springframework.integration.context.IntegrationContextUtils.INTEGRATION_EVALUATION_CONTEXT_BEAN_NAME; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.w3c.dom.Element; @@ -35,6 +33,7 @@ import org.springframework.beans.factory.xml.BeanDefinitionParser; import org.springframework.beans.factory.xml.NamespaceHandler; import org.springframework.beans.factory.xml.NamespaceHandlerSupport; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.channel.registry.DefaultHeaderChannelRegistry; import org.springframework.integration.config.IntegrationEvaluationContextFactoryBean; import org.springframework.integration.config.xml.ChannelInitializer.AutoCreateCandidatesCollector; import org.springframework.integration.context.IntegrationContextUtils; @@ -68,15 +67,18 @@ public abstract class AbstractIntegrationNamespaceHandler implements NamespaceHa private final NamespaceHandlerDelegate delegate = new NamespaceHandlerDelegate(); + @Override public final BeanDefinition parse(Element element, ParserContext parserContext) { this.verifySchemaVersion(element, parserContext); this.registerImplicitChannelCreator(parserContext); this.registerIntegrationEvaluationContext(parserContext); + this.registerHeaderChannelRegistry(parserContext); this.registerBuiltInBeans(parserContext); this.registerDefaultConfiguringBeanFactoryPostProcessorIfNecessary(parserContext); return this.delegate.parse(element, parserContext); } + @Override public final BeanDefinitionHolder decorate(Node source, BeanDefinitionHolder definition, ParserContext parserContext) { return this.delegate.decorate(source, definition, parserContext); } @@ -127,10 +129,11 @@ public abstract class AbstractIntegrationNamespaceHandler implements NamespaceHa // unlike DefaultConfiguringBeanFactoryPostProcessor, we need one of these per registry // therefore we need to call containsBeanDefinition(..) which does not consider the parent registry alreadyRegistered = ((ListableBeanFactory) parserContext.getRegistry()).containsBeanDefinition( - INTEGRATION_EVALUATION_CONTEXT_BEAN_NAME); + IntegrationContextUtils.INTEGRATION_EVALUATION_CONTEXT_BEAN_NAME); } else { - alreadyRegistered = parserContext.getRegistry().isBeanNameInUse(INTEGRATION_EVALUATION_CONTEXT_BEAN_NAME); + alreadyRegistered = parserContext.getRegistry().isBeanNameInUse( + IntegrationContextUtils.INTEGRATION_EVALUATION_CONTEXT_BEAN_NAME); } if (!alreadyRegistered) { BeanDefinitionBuilder integrationEvaluationContextBuilder = BeanDefinitionBuilder @@ -195,6 +198,33 @@ public abstract class AbstractIntegrationNamespaceHandler implements NamespaceHa } } + /** + * Register a DefaultHeaderChannelRegistry in the given BeanDefinitionRegistry, if necessary. + */ + private void registerHeaderChannelRegistry(ParserContext parserContext) { + boolean alreadyRegistered = false; + if (parserContext.getRegistry() instanceof ListableBeanFactory) { + alreadyRegistered = ((ListableBeanFactory) parserContext.getRegistry()) + .containsBean(IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME); + } + else { + alreadyRegistered = parserContext.getRegistry().isBeanNameInUse( + IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME); + } + if (!alreadyRegistered) { + if (logger.isInfoEnabled()) { + logger.info("No bean named '" + IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME + + "' has been explicitly defined. Therefore, a default DefaultHeaderChannelRegistry will be created."); + } + BeanDefinitionBuilder schedulerBuilder = BeanDefinitionBuilder.genericBeanDefinition(DefaultHeaderChannelRegistry.class); + BeanDefinitionHolder replyChannelRegistryComponent = new BeanDefinitionHolder( + schedulerBuilder.getBeanDefinition(), + IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME); + BeanDefinitionReaderUtils.registerBeanDefinition(replyChannelRegistryComponent, parserContext.getRegistry()); + } + } + + protected final void registerBeanDefinitionDecorator(String elementName, BeanDefinitionDecorator decorator) { this.delegate.doRegisterBeanDefinitionDecorator(elementName, decorator); } @@ -225,6 +255,7 @@ public abstract class AbstractIntegrationNamespaceHandler implements NamespaceHa private class NamespaceHandlerDelegate extends NamespaceHandlerSupport { + @Override public void init() { AbstractIntegrationNamespaceHandler.this.init(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java index 6da6aec529..2e7d981a87 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java @@ -37,7 +37,6 @@ import org.springframework.util.xml.DomUtils; public abstract class AbstractPollingInboundChannelAdapterParser extends AbstractChannelAdapterParser { @Override - @SuppressWarnings("unchecked") protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) { BeanMetadataElement source = this.parseSource(element, parserContext); if (source == 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 5dba32f685..fba8d51b2e 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 @@ -29,11 +29,12 @@ import org.springframework.beans.factory.config.TypedStringValue; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.ManagedMap; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.expression.DynamicExpression; +import org.springframework.integration.transformer.HeaderEnricher; import org.springframework.util.ClassUtils; import org.springframework.util.StringUtils; import org.springframework.util.xml.DomUtils; -import org.springframework.integration.expression.DynamicExpression; -import org.springframework.integration.transformer.HeaderEnricher; /** * Base support class for 'header-enricher' parsers. @@ -41,6 +42,7 @@ import org.springframework.integration.transformer.HeaderEnricher; * @author Mark Fisher * @author Oleg Zhurakousky * @author Artem Bilan + * @author Gary Russell * @since 2.0 */ public abstract class HeaderEnricherParserSupport extends AbstractTransformerParser { @@ -49,6 +51,16 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar private final Map> elementToTypeMap = new HashMap>(); + private final static Map cannedHeaderElementExpressions = new HashMap(); + + static { + cannedHeaderElementExpressions.put("header-channels-to-string", new String[][] { + {"replyChannel", "@" + IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME + + ".channelToChannelName(headers.replyChannel)" }, + {"errorChannel", "@" + IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME + + ".channelToChannelName(headers.errorChannel)" }, + }); + } @Override protected final String getTransformerClassName() { @@ -86,6 +98,8 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar Element headerElement = (Element) node; String elementName = node.getLocalName(); Class headerType = null; + String expression = null; + String overwrite = headerElement.getAttribute("overwrite"); if ("header".equals(elementName)) { headerName = headerElement.getAttribute(NAME_ATTRIBUTE); } @@ -114,135 +128,157 @@ public abstract class HeaderEnricherParserSupport extends AbstractTransformerPar } } } - if (headerName != null) { - String value = headerElement.getAttribute("value"); - String ref = headerElement.getAttribute(REF_ATTRIBUTE); - String method = headerElement.getAttribute(METHOD_ATTRIBUTE); - String expression = headerElement.getAttribute(EXPRESSION_ATTRIBUTE); - - Element beanElement = null; - Element scriptElement = null; - Element expressionElement = null; - - List subElements = DomUtils.getChildElements(headerElement); - if (!subElements.isEmpty()) { - Element subElement = subElements.get(0); - String subElementLocalName = subElement.getLocalName(); - if ("bean".equals(subElementLocalName)) { - beanElement = subElement; - } - else if ("script".equals(subElementLocalName)) { - scriptElement = subElement; - } - else if ("expression".equals(subElementLocalName)) { - expressionElement = subElement; - } - if (beanElement == null && scriptElement == null && expressionElement == null) { - parserContext.getReaderContext().error("Only 'bean', 'script' or 'expression' can be defined as a sub-element", element); + if (headerName == null) { + if (cannedHeaderElementExpressions.containsKey(elementName)) { + for (int j = 0; j < cannedHeaderElementExpressions.get(elementName).length; j++) { + headerName = cannedHeaderElementExpressions.get(elementName)[j][0]; + expression = cannedHeaderElementExpressions.get(elementName)[j][1]; + overwrite = "true"; + this.addHeader(element, headers, parserContext, headerName, headerElement, headerType, + expression, overwrite); } } - if (StringUtils.hasText(expression) && expressionElement != null) { - parserContext.getReaderContext().error("The 'expression' attribute and sub-element are mutually exclusive", element); - } - - boolean isValue = StringUtils.hasText(value); - boolean isRef = StringUtils.hasText(ref); - boolean hasMethod = StringUtils.hasText(method); - boolean isExpression = StringUtils.hasText(expression) || expressionElement != null; - boolean isScript = scriptElement != null; - - BeanDefinition innerComponentDefinition = null; - - if (beanElement != null) { - innerComponentDefinition = parserContext.getDelegate().parseBeanDefinitionElement(beanElement).getBeanDefinition(); - } - else if (isScript) { - innerComponentDefinition = parserContext.getDelegate().parseCustomElement(scriptElement); - } - - boolean isCustomBean = innerComponentDefinition != null; - - if (hasMethod && isScript) { - parserContext.getReaderContext().error("The 'method' attribute cannot be used when a 'script' sub-element is defined", element); - } - - if (!(isValue ^ (isRef ^ (isExpression ^ isCustomBean)))) { - parserContext.getReaderContext().error( - "Exactly one of the 'ref', 'value', 'expression' or inner bean is required.", element); - } - BeanDefinitionBuilder valueProcessorBuilder = null; - if (isValue) { - if (hasMethod) { - 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( - IntegrationNamespaceUtils.BASE_PACKAGE + ".transformer.support.StaticHeaderValueMessageProcessor"); - valueProcessorBuilder.addConstructorArgValue(headerValue); - } - else if (isExpression) { - if (hasMethod) { - parserContext.getReaderContext().error( - "The 'method' attribute cannot be used with the 'expression' attribute.", element); - } - valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".transformer.support.ExpressionEvaluatingHeaderValueMessageProcessor"); - if (expressionElement != null) { - BeanDefinitionBuilder dynamicExpressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(DynamicExpression.class); - dynamicExpressionBuilder.addConstructorArgValue(expressionElement.getAttribute("key")); - dynamicExpressionBuilder.addConstructorArgReference(expressionElement.getAttribute("source")); - valueProcessorBuilder.addConstructorArgValue(dynamicExpressionBuilder.getBeanDefinition()); - } - else { - valueProcessorBuilder.addConstructorArgValue(expression); - } - valueProcessorBuilder.addConstructorArgValue(headerType); - } - else if (isCustomBean) { - if (StringUtils.hasText(headerElement.getAttribute("type"))) { - parserContext.getReaderContext().error( - "The 'type' attribute cannot be used with an inner bean.", element); - } - if (hasMethod || isScript) { - valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".transformer.support.MessageProcessingHeaderValueMessageProcessor"); - valueProcessorBuilder.addConstructorArgValue(innerComponentDefinition); - if (hasMethod) { - valueProcessorBuilder.addConstructorArgValue(method); - } - } - else { - valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".transformer.support.StaticHeaderValueMessageProcessor"); - valueProcessorBuilder.addConstructorArgValue(innerComponentDefinition); - } - } - else { - if (StringUtils.hasText(headerElement.getAttribute("type"))) { - parserContext.getReaderContext().error( - "The 'type' attribute cannot be used with the 'ref' attribute.", element); - } - if (hasMethod) { - valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".transformer.support.MessageProcessingHeaderValueMessageProcessor"); - valueProcessorBuilder.addConstructorArgReference(ref); - valueProcessorBuilder.addConstructorArgValue(method); - } - else { - valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition( - IntegrationNamespaceUtils.BASE_PACKAGE + ".transformer.support.StaticHeaderValueMessageProcessor"); - valueProcessorBuilder.addConstructorArgReference(ref); - } - } - IntegrationNamespaceUtils.setValueIfAttributeDefined(valueProcessorBuilder, headerElement, "overwrite"); - headers.put(headerName, valueProcessorBuilder.getBeanDefinition()); + } + else { + this.addHeader(element, headers, parserContext, headerName, headerElement, headerType, expression, + overwrite); } } } } + private void addHeader(Element element, ManagedMap headers, ParserContext parserContext, + String headerName, Element headerElement, Class headerType, String expression, String overwrite) { + + String value = headerElement.getAttribute("value"); + String ref = headerElement.getAttribute(REF_ATTRIBUTE); + String method = headerElement.getAttribute(METHOD_ATTRIBUTE); + if (expression == null) { + expression = headerElement.getAttribute(EXPRESSION_ATTRIBUTE); + } + + Element beanElement = null; + Element scriptElement = null; + Element expressionElement = null; + + List subElements = DomUtils.getChildElements(headerElement); + if (!subElements.isEmpty()) { + Element subElement = subElements.get(0); + String subElementLocalName = subElement.getLocalName(); + if ("bean".equals(subElementLocalName)) { + beanElement = subElement; + } + else if ("script".equals(subElementLocalName)) { + scriptElement = subElement; + } + else if ("expression".equals(subElementLocalName)) { + expressionElement = subElement; + } + if (beanElement == null && scriptElement == null && expressionElement == null) { + parserContext.getReaderContext().error("Only 'bean', 'script' or 'expression' can be defined as a sub-element", element); + } + } + if (StringUtils.hasText(expression) && expressionElement != null) { + parserContext.getReaderContext().error("The 'expression' attribute and sub-element are mutually exclusive", element); + } + + boolean isValue = StringUtils.hasText(value); + boolean isRef = StringUtils.hasText(ref); + boolean hasMethod = StringUtils.hasText(method); + boolean isExpression = StringUtils.hasText(expression) || expressionElement != null; + boolean isScript = scriptElement != null; + + BeanDefinition innerComponentDefinition = null; + + if (beanElement != null) { + innerComponentDefinition = parserContext.getDelegate().parseBeanDefinitionElement(beanElement).getBeanDefinition(); + } + else if (isScript) { + innerComponentDefinition = parserContext.getDelegate().parseCustomElement(scriptElement); + } + + boolean isCustomBean = innerComponentDefinition != null; + + if (hasMethod && isScript) { + parserContext.getReaderContext().error("The 'method' attribute cannot be used when a 'script' sub-element is defined", element); + } + + if (!(isValue ^ (isRef ^ (isExpression ^ isCustomBean)))) { + parserContext.getReaderContext().error( + "Exactly one of the 'ref', 'value', 'expression' or inner bean is required.", element); + } + BeanDefinitionBuilder valueProcessorBuilder = null; + if (isValue) { + if (hasMethod) { + 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( + IntegrationNamespaceUtils.BASE_PACKAGE + ".transformer.support.StaticHeaderValueMessageProcessor"); + valueProcessorBuilder.addConstructorArgValue(headerValue); + } + else if (isExpression) { + if (hasMethod) { + parserContext.getReaderContext().error( + "The 'method' attribute cannot be used with the 'expression' attribute.", element); + } + valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition( + IntegrationNamespaceUtils.BASE_PACKAGE + ".transformer.support.ExpressionEvaluatingHeaderValueMessageProcessor"); + if (expressionElement != null) { + BeanDefinitionBuilder dynamicExpressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(DynamicExpression.class); + dynamicExpressionBuilder.addConstructorArgValue(expressionElement.getAttribute("key")); + dynamicExpressionBuilder.addConstructorArgReference(expressionElement.getAttribute("source")); + valueProcessorBuilder.addConstructorArgValue(dynamicExpressionBuilder.getBeanDefinition()); + } + else { + valueProcessorBuilder.addConstructorArgValue(expression); + } + valueProcessorBuilder.addConstructorArgValue(headerType); + } + else if (isCustomBean) { + if (StringUtils.hasText(headerElement.getAttribute("type"))) { + parserContext.getReaderContext().error( + "The 'type' attribute cannot be used with an inner bean.", element); + } + if (hasMethod || isScript) { + valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition( + IntegrationNamespaceUtils.BASE_PACKAGE + ".transformer.support.MessageProcessingHeaderValueMessageProcessor"); + valueProcessorBuilder.addConstructorArgValue(innerComponentDefinition); + if (hasMethod) { + valueProcessorBuilder.addConstructorArgValue(method); + } + } + else { + valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition( + IntegrationNamespaceUtils.BASE_PACKAGE + ".transformer.support.StaticHeaderValueMessageProcessor"); + valueProcessorBuilder.addConstructorArgValue(innerComponentDefinition); + } + } + else { + if (StringUtils.hasText(headerElement.getAttribute("type"))) { + parserContext.getReaderContext().error( + "The 'type' attribute cannot be used with the 'ref' attribute.", element); + } + if (hasMethod) { + valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition( + IntegrationNamespaceUtils.BASE_PACKAGE + ".transformer.support.MessageProcessingHeaderValueMessageProcessor"); + valueProcessorBuilder.addConstructorArgReference(ref); + valueProcessorBuilder.addConstructorArgValue(method); + } + else { + valueProcessorBuilder = BeanDefinitionBuilder.genericBeanDefinition( + IntegrationNamespaceUtils.BASE_PACKAGE + ".transformer.support.StaticHeaderValueMessageProcessor"); + valueProcessorBuilder.addConstructorArgReference(ref); + } + } + if (StringUtils.hasText(overwrite)) { + valueProcessorBuilder.addPropertyValue("overwrite", overwrite); + } + headers.put(headerName, valueProcessorBuilder.getBeanDefinition()); + } + /** * Subclasses may override this method to provide any additional processing. */ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java index a3d6f70a0a..f432a0ea5e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationContextUtils.java @@ -45,6 +45,9 @@ public abstract class IntegrationContextUtils { public static final String INTEGRATION_EVALUATION_CONTEXT_BEAN_NAME = "integrationEvaluationContext"; + public static final String INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME = "integrationHeaderChannelRegistry"; + + /** * Return the {@link MetadataStore} bean whose name is "metadataStore". * @param beanFactory BeanFactory for lookup, must not be null. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index 809770fa1d..a3d840822a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2013 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,7 +19,6 @@ package org.springframework.integration.gateway; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; import org.springframework.integration.MessagingException; -import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.core.PollableChannel; import org.springframework.integration.core.SubscribableChannel; @@ -41,8 +40,9 @@ import org.springframework.util.Assert; * {@link MessageChannel}s for sending, receiving, or request-reply operations. * Exposes setters for configuring request and reply {@link MessageChannel}s as * well as the timeout values for sending and receiving Messages. - * + * * @author Mark Fisher + * @author Gary Russell */ public abstract class MessagingGatewaySupport extends AbstractEndpoint implements TrackableComponent { @@ -84,7 +84,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement /** * Set the request channel. - * + * * @param requestChannel the channel to which request messages will be sent */ public void setRequestChannel(MessageChannel requestChannel) { @@ -94,7 +94,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement /** * Set the reply channel. If no reply channel is provided, this gateway will * always use an anonymous, temporary channel for handling replies. - * + * * @param replyChannel the channel from which reply messages will be received */ public void setReplyChannel(MessageChannel replyChannel) { @@ -113,7 +113,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement /** * Set the timeout value for sending request messages. If not * explicitly configured, the default is one second. - * + * * @param requestTimeout the timeout value in milliseconds */ public void setRequestTimeout(long requestTimeout) { @@ -123,7 +123,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement /** * Set the timeout value for receiving reply messages. If not * explicitly configured, the default is one second. - * + * * @param replyTimeout the timeout value in milliseconds */ public void setReplyTimeout(long replyTimeout) { @@ -153,6 +153,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement * Specify whether this gateway should be tracked in the Message History * of Messages that originate from its send or sendAndReceive operations. */ + @Override public void setShouldTrack(boolean shouldTrack) { this.historyWritingPostProcessor.setShouldTrack(shouldTrack); } @@ -283,7 +284,11 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement return; } AbstractEndpoint correlator = null; - MessageHandler handler = new BridgeHandler(); + BridgeHandler handler = new BridgeHandler(); + if (this.getBeanFactory() != null) { + handler.setBeanFactory(this.getBeanFactory()); + } + handler.afterPropertiesSet(); if (this.replyChannel instanceof SubscribableChannel) { correlator = new EventDrivenConsumer( (SubscribableChannel) this.replyChannel, handler); @@ -320,6 +325,7 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint implement private static class DefaultRequestMapper implements InboundMessageMapper { + @Override public Message toMessage(Object object) throws Exception { if (object instanceof Message) { return (Message) object; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/channel/BeanFactoryChannelResolver.java b/spring-integration-core/src/main/java/org/springframework/integration/support/channel/BeanFactoryChannelResolver.java index 8f9abe8e95..2a17a0e139 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/channel/BeanFactoryChannelResolver.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/channel/BeanFactoryChannelResolver.java @@ -16,25 +16,33 @@ package org.springframework.integration.support.channel; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.integration.MessageChannel; +import org.springframework.integration.channel.registry.HeaderChannelRegistry; +import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.util.Assert; /** * {@link ChannelResolver} implementation based on a Spring {@link BeanFactory}. - * + * *

Will lookup Spring managed beans identified by bean name, * expecting them to be of type {@link MessageChannel}. - * + * * @author Mark Fisher * @see org.springframework.beans.factory.BeanFactory */ public class BeanFactoryChannelResolver implements ChannelResolver, BeanFactoryAware { + private final static Log logger = LogFactory.getLog(BeanFactoryChannelResolver.class); + private volatile BeanFactory beanFactory; + private volatile HeaderChannelRegistry replyChannelRegistry; /** * Create a new instance of the {@link BeanFactoryChannelResolver} class. @@ -53,25 +61,45 @@ public class BeanFactoryChannelResolver implements ChannelResolver, BeanFactoryA * replaced by the {@link BeanFactory} that creates it (c.f. the * {@link BeanFactoryAware} contract). So only use this constructor if you * are instantiating this object explicitly rather than defining a bean. - * + * * @param beanFactory the bean factory to be used to lookup {@link MessageChannel}s. */ public BeanFactoryChannelResolver(BeanFactory beanFactory) { Assert.notNull(beanFactory, "BeanFactory must not be null"); - this.beanFactory = beanFactory; + this.lookupHeaderChannelRegistry(beanFactory); } - + @Override public void setBeanFactory(BeanFactory beanFactory) { this.beanFactory = beanFactory; + this.lookupHeaderChannelRegistry(beanFactory); } + private void lookupHeaderChannelRegistry(BeanFactory beanFactory) { + this.beanFactory = beanFactory; + try { + this.replyChannelRegistry = beanFactory.getBean( + IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME, + HeaderChannelRegistry.class); + } + catch (Exception e) { + logger.warn("No HeaderChannelRegistry found", e); + } + } + + @Override public MessageChannel resolveChannelName(String name) { Assert.state(this.beanFactory != null, "BeanFactory is required"); try { return this.beanFactory.getBean(name, MessageChannel.class); } catch (BeansException e) { + if (this.replyChannelRegistry != null) { + MessageChannel channel = this.replyChannelRegistry.channelNameToChannel(name); + if (channel != null) { + return channel; + } + } throw new ChannelResolutionException( "failed to look up MessageChannel bean with name '" + name + "'", e); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java index b6d18c5173..21c60f32e4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java @@ -214,6 +214,10 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler implem this.gateway.setReplyChannel(replyChannel); } + if (this.getBeanFactory() != null) { + this.gateway.setBeanFactory(this.getBeanFactory()); + } + this.gateway.afterPropertiesSet(); } @@ -294,6 +298,7 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler implem * Lifecycle implementation. If no requestChannel is defined, this method * has no effect as in that case no Gateway is initialized. */ + @Override public void start() { if (this.gateway != null) { this.gateway.start(); @@ -304,6 +309,7 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler implem * Lifecycle implementation. If no requestChannel is defined, this method * has no effect as in that case no Gateway is initialized. */ + @Override public void stop() { if (this.gateway != null) { this.gateway.stop(); @@ -314,6 +320,7 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler implem * Lifecycle implementation. If no requestChannel is defined, this method * will return always return true as no Gateway is initialized. */ + @Override public boolean isRunning() { if (this.gateway != null) { return this.gateway.isRunning(); diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd index a7b47c6edb..d715557a4f 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-3.0.xsd @@ -1853,6 +1853,15 @@ + + + + Converts the 'replyChannel' and 'errorChannel' headers to a String after registering it in the HeaderChannelRegistry. + Use this when a message is serialized for any reason. No changes are made + if the header does not exist, or if the header does not currently reference a MessageChannel. + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/registry/HeaderChannelRegistryTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/channel/registry/HeaderChannelRegistryTests-context.xml new file mode 100644 index 0000000000..5e073d435e --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/registry/HeaderChannelRegistryTests-context.xml @@ -0,0 +1,69 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/registry/HeaderChannelRegistryTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/registry/HeaderChannelRegistryTests.java new file mode 100644 index 0000000000..363f154299 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/registry/HeaderChannelRegistryTests.java @@ -0,0 +1,167 @@ +/* + * Copyright 2013 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.channel.registry; + +import static org.hamcrest.Matchers.instanceOf; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import org.hamcrest.Matchers; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.MessagePublishingErrorHandler; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.MessagingTemplate; +import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.message.ErrorMessage; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.scheduling.TaskScheduler; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 3.0 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class HeaderChannelRegistryTests { + + @Autowired + MessageChannel input; + + @Autowired + MessageChannel inputPolled; + + @Autowired + QueueChannel alreadyAString; + + @Autowired + TaskScheduler taskScheduler; + + @Autowired + Gateway gatewayNoReplyChannel; + + @Autowired + Gateway gatewayExplicitReplyChannel; + + @Test + public void testReplace() { + MessagingTemplate template = new MessagingTemplate(); + template.setDefaultChannel(this.input); + Message reply = template.sendAndReceive(new GenericMessage("foo")); + assertNotNull(reply); + assertEquals("echo:foo", reply.getPayload()); + } + + @Test + public void testReplaceGatewayWithNoReplyChannel() { + String reply = this.gatewayNoReplyChannel.exchange("foo"); + assertNotNull(reply); + assertEquals("echo:foo", reply); + } + + @Test + public void testReplaceGatewayWithExplicitReplyChannel() { + String reply = this.gatewayExplicitReplyChannel.exchange("foo"); + assertNotNull(reply); + assertEquals("echo:foo", reply); + } + + /** + * MessagingTemplate sets the errorChannel to the replyChannel so it gets any async + * exceptions via the default {@link MessagePublishingErrorHandler}. + */ + @Test + public void testReplaceError() { + MessagingTemplate template = new MessagingTemplate(); + template.setDefaultChannel(this.inputPolled); + Message reply = template.sendAndReceive(new GenericMessage("bar")); + assertNotNull(reply); + assertTrue(reply instanceof ErrorMessage); + } + + @Test + public void testAlreadyAString() { + Message requestMessage = MessageBuilder.withPayload("foo") + .setReplyChannelName("alreadyAString") + .setErrorChannelName("alreadyAnotherString") + .build(); + this.input.send(requestMessage); + Message reply = alreadyAString.receive(0); + assertNotNull(reply); + assertEquals("echo:foo", reply.getPayload()); + } + + @Test + public void testNull() { + Message requestMessage = MessageBuilder.withPayload("foo") + .build(); + try { + this.input.send(requestMessage); + fail("expected exception"); + } + catch (Exception e) { + assertThat(e.getMessage(), Matchers.containsString("no output-channel or replyChannel")); + } + } + + @Test + public void testExpire() throws Exception { + DefaultHeaderChannelRegistry registry = new DefaultHeaderChannelRegistry(50); + registry.setTaskScheduler(this.taskScheduler); + registry.start(); + Thread.sleep(200); + String id = (String) registry.channelToChannelName(new DirectChannel()); + Thread.sleep(300); + assertNull(registry.channelNameToChannel(id)); + registry.stop(); + } + + public static class Foo extends AbstractReplyProducingMessageHandler { + + @Override + protected Object handleRequestMessage(Message requestMessage) { + assertThat(requestMessage.getHeaders().getReplyChannel(), + Matchers.anyOf(instanceOf(String.class), Matchers.nullValue())); + assertThat(requestMessage.getHeaders().getErrorChannel(), + Matchers.anyOf(instanceOf(String.class), Matchers.nullValue())); + if (requestMessage.getPayload().equals("bar")) { + throw new RuntimeException("intentional"); + } + return "echo:" + requestMessage.getPayload(); + } + + } + + public interface Gateway { + + String exchange(String foo); + + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ControlBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ControlBusTests.java index a33a614b87..a5efb64f48 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ControlBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/ControlBusTests.java @@ -16,8 +16,8 @@ package org.springframework.integration.config.xml; -import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import java.util.Date; @@ -30,6 +30,9 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.channel.registry.DefaultHeaderChannelRegistry; +import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.core.PollableChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.support.MessageBuilder; @@ -52,6 +55,9 @@ public class ControlBusTests { @Autowired private PollableChannel output; + @Autowired + private DefaultHeaderChannelRegistry registry; + @Test public void testDefaultEvaluationContext() { Message message = MessageBuilder.withPayload("@service.convert('aardvark')+headers.foo").setHeader("foo", "bar").build(); @@ -71,6 +77,27 @@ public class ControlBusTests { assertNotNull(outputChannel.receive(1000)); } + @Test + public void testControlHeaderChannelReaper() throws InterruptedException { + MessagingTemplate messagingTemplate = new MessagingTemplate(); + messagingTemplate.convertAndSend(input, "@integrationHeaderChannelRegistry.size()"); + Message result = this.output.receive(0); + assertNotNull(result); + assertEquals(0, result.getPayload()); + this.registry.setReaperDelay(10); + this.registry.channelToChannelName(new DirectChannel()); + messagingTemplate.convertAndSend(input, "@integrationHeaderChannelRegistry.size()"); + result = this.output.receive(0); + assertNotNull(result); + assertEquals(1, result.getPayload()); + Thread.sleep(100); + messagingTemplate.convertAndSend(input, "@integrationHeaderChannelRegistry.runReaper()"); + messagingTemplate.convertAndSend(input, "@integrationHeaderChannelRegistry.size()"); + result = this.output.receive(0); + assertNotNull(result); + assertEquals(0, result.getPayload()); + this.registry.setReaperDelay(60000); + } public static class Service { @@ -78,12 +105,14 @@ public class ControlBusTests { public String convert(String input) { return "cat"; } + } public static class AdapterService { public Message receive() { return new GenericMessage(new Date().toString()); } + } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests-context.xml index b0c558e552..f62be84be9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests-context.xml @@ -13,11 +13,17 @@ + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests.java index 18cb5ed44b..45dd1979a1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/xml/EnricherParserTests.java @@ -17,6 +17,7 @@ package org.springframework.integration.config.xml; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; @@ -79,6 +80,7 @@ public class EnricherParserTests { assertEquals(context.getBean("output"), accessor.getPropertyValue("outputChannel")); assertEquals(true, accessor.getPropertyValue("shouldClonePayload")); assertNull(accessor.getPropertyValue("requestPayloadExpression")); + assertNotNull(TestUtils.getPropertyValue(enricher, "gateway.beanFactory")); Map propertyExpressions = (Map) accessor.getPropertyValue("propertyExpressions"); for (Map.Entry e : propertyExpressions.entrySet()) { @@ -125,12 +127,15 @@ public class EnricherParserTests { @Test public void integrationTest() { SubscribableChannel requests = context.getBean("requests", SubscribableChannel.class); - requests.subscribe(new AbstractReplyProducingMessageHandler() { + class Foo extends AbstractReplyProducingMessageHandler { @Override protected Object handleRequestMessage(Message requestMessage) { return new Source("foo"); } - }); + }; + Foo foo = new Foo(); + foo.setOutputChannel(context.getBean("replies", MessageChannel.class)); + requests.subscribe(foo); Target original = new Target(); Message request = MessageBuilder.withPayload(original) .setHeader("sourceName", "test") diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java index d88b55f555..d042b8c39a 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/ChannelPublishingJmsMessageListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2013 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. @@ -28,6 +28,9 @@ import javax.jms.Session; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; @@ -47,18 +50,18 @@ import org.springframework.util.Assert; * Message and sends that Message to a channel. If the 'expectReply' value is * true, it will also wait for a Spring Integration reply Message * and convert that into a JMS reply. - * + * * @author Mark Fisher * @author Juergen Hoeller * @author Oleg Zhurakousky */ -public class ChannelPublishingJmsMessageListener - implements SessionAwareMessageListener, InitializingBean, TrackableComponent { - +public class ChannelPublishingJmsMessageListener + implements SessionAwareMessageListener, InitializingBean, TrackableComponent, BeanFactoryAware { + protected final Log logger = LogFactory.getLog(getClass()); - + private volatile boolean expectReply; - + private volatile MessageConverter messageConverter = new SimpleMessageConverter(); private volatile boolean extractRequestPayload = true; @@ -80,9 +83,11 @@ public class ChannelPublishingJmsMessageListener private volatile DestinationResolver destinationResolver = new DynamicDestinationResolver(); private volatile JmsHeaderMapper headerMapper = new DefaultJmsHeaderMapper(); - + private final GatewayDelegate gatewayDelegate = new GatewayDelegate(); + private volatile BeanFactory beanFactory; + /** * Specify whether a JMS reply Message is expected. */ @@ -93,39 +98,42 @@ public class ChannelPublishingJmsMessageListener public void setComponentName(String componentName){ this.gatewayDelegate.setComponentName(componentName); } - + public void setRequestChannel(MessageChannel requestChannel){ this.gatewayDelegate.setRequestChannel(requestChannel); } - + public void setReplyChannel(MessageChannel replyChannel){ this.gatewayDelegate.setReplyChannel(replyChannel); } - + public void setErrorChannel(MessageChannel errorChannel){ this.gatewayDelegate.setErrorChannel(errorChannel); } - + public void setRequestTimeout(long requestTimeout){ this.gatewayDelegate.setRequestTimeout(requestTimeout); } - + public void setReplyTimeout(long replyTimeout){ this.gatewayDelegate.setReplyTimeout(replyTimeout); } - + + @Override public void setShouldTrack(boolean shouldTrack) { this.gatewayDelegate.setShouldTrack(shouldTrack); } + @Override public String getComponentName() { return this.gatewayDelegate.getComponentName(); } + @Override public String getComponentType() { return this.gatewayDelegate.getComponentType(); } - + /** * Set the default reply destination to send reply messages to. This will * be applied in case of a request message that does not carry a @@ -189,7 +197,7 @@ public class ChannelPublishingJmsMessageListener * JMSMessageID from the request will be copied into the JMSCorrelationID of the reply * unless there is already a value in the JMSCorrelationID property of the newly created * reply Message in which case nothing will be copied. If the JMSCorrelationID of the - * request Message should be copied into the JMSCorrelationID of the reply Message + * request Message should be copied into the JMSCorrelationID of the reply Message * instead, then this value should be set to "JMSCorrelationID". * Any other value will be treated as a JMS String Property to be copied as-is * from the request Message into the reply Message with the same property name. @@ -200,7 +208,7 @@ public class ChannelPublishingJmsMessageListener /** * Specify whether explicit QoS should be enabled for replies - * (for timeToLive, priority, and deliveryMode settings). + * (for timeToLive, priority, and deliveryMode settings). */ public void setExplicitQosEnabledForReplies(boolean explicitQosEnabledForReplies) { this.explicitQosEnabledForReplies = explicitQosEnabledForReplies; @@ -224,7 +232,7 @@ public class ChannelPublishingJmsMessageListener * converting between JMS Messages and Spring Integration Messages. * If none is provided, a {@link SimpleMessageConverter} will * be used. - * + * * @param messageConverter */ public void setMessageConverter(MessageConverter messageConverter) { @@ -260,6 +268,12 @@ public class ChannelPublishingJmsMessageListener this.extractReplyPayload = extractReplyPayload; } + @Override + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + this.beanFactory = beanFactory; + } + + @Override public void onMessage(javax.jms.Message jmsMessage, Session session) throws JMSException { Object result = jmsMessage; if (this.extractRequestPayload) { @@ -268,10 +282,10 @@ public class ChannelPublishingJmsMessageListener logger.debug("converted JMS Message [" + jmsMessage + "] to integration Message payload [" + result + "]"); } } - + Map headers = headerMapper.toHeaders(jmsMessage); Message requestMessage = (result instanceof Message) ? - MessageBuilder.fromMessage((Message) result).copyHeaders(headers).build() : + MessageBuilder.fromMessage((Message) result).copyHeaders(headers).build() : MessageBuilder.withPayload(result).copyHeaders(headers).build(); if (!this.expectReply) { this.gatewayDelegate.send(requestMessage); @@ -305,18 +319,22 @@ public class ChannelPublishingJmsMessageListener } } + @Override public void afterPropertiesSet() { + if (this.beanFactory != null) { + this.gatewayDelegate.setBeanFactory(this.beanFactory); + } this.gatewayDelegate.afterPropertiesSet(); } - + protected void start(){ this.gatewayDelegate.start(); } - + protected void stop(){ this.gatewayDelegate.stop(); } - + private void copyCorrelationIdFromRequestToReply(javax.jms.Message requestMessage, javax.jms.Message replyMessage) throws JMSException { if (this.correlationKey != null) { if (this.correlationKey.equals("JMSCorrelationID")) { @@ -416,17 +434,20 @@ public class ChannelPublishingJmsMessageListener this.isTopic = isTopic; } } - + private class GatewayDelegate extends MessagingGatewaySupport { + @Override protected void send(Object request) { super.send(request); } - + + @Override protected Message sendAndReceiveMessage(Object request) { return super.sendAndReceiveMessage(request); } + @Override public String getComponentType() { if (expectReply) { return "jms:inbound-gateway"; diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/GatewaySerializedReplyChannelTests-context.xml b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/GatewaySerializedReplyChannelTests-context.xml new file mode 100644 index 0000000000..daeb4b956e --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/GatewaySerializedReplyChannelTests-context.xml @@ -0,0 +1,37 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/GatewaySerializedReplyChannelTests.java b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/GatewaySerializedReplyChannelTests.java new file mode 100644 index 0000000000..540c4abde3 --- /dev/null +++ b/spring-integration-jms/src/test/java/org/springframework/integration/jms/request_reply/GatewaySerializedReplyChannelTests.java @@ -0,0 +1,55 @@ +/* + * Copyright 2013 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.jms.request_reply; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.message.GenericMessage; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Gary Russell + * @since 3.0 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class GatewaySerializedReplyChannelTests { + + @Autowired + MessageChannel input; + + @Autowired + PollableChannel output; + + @Test + public void test() { + input.send(new GenericMessage("foo")); + Message reply = output.receive(0); + assertNotNull(reply); + assertEquals("echo:foo", reply.getPayload()); + } + +} diff --git a/src/reference/docbook/content-enrichment.xml b/src/reference/docbook/content-enrichment.xml index 6e8cf27f3d..657a5600c2 100644 --- a/src/reference/docbook/content-enrichment.xml +++ b/src/reference/docbook/content-enrichment.xml @@ -73,10 +73,10 @@ 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. - + - POJO Support + POJO Support @@ -121,7 +121,7 @@ ]]> - SpEL Support + SpEL Support In Spring Integration 2.0 we have introduced the convenience of the @@ -147,6 +147,51 @@ are bound to the SpEL Evaluation Context, giving you full access to the incoming Message. + + Header Channel Registry + + + Starting with Spring Integration 3.0, a new sub-element + <int:header-channels-to-string/> is available; it has no attributes. + This converts existing replyChannel and errorChannel + headers (when they are a + MessageChannel) to a String and stores the channel(s) in + a registry for later resolution when it is time to send a reply, or handle an error. + This is useful + for cases where the headers might be lost; for example when + serializing a message into a message store or when transporting the message + over JMS. If the header does not already exist, or it + is not a MessageChannel, no changes are made. + + + + Use of this functionality requires the presence of a HeaderChannelRegistry + bean. By default, the framework creates a DefaultHeaderChannelRegistry + with the default expiry (60 seconds). Channels + are removed from the registry after this time. To change this, simply define a bean + with id integrationHeaderChannelRegistry and configure the required delay using + a constructor argument (milliseconds). + + + + The HeaderChannelRegistry has a size() method to + determine the current size of the registry. The runReaper() method + cancels the current scheduled task and runs the reaper immediately; the task is + then scheduled to run again based on the current delay. These methods can be invoked + directly by getting a reference to the registry, or you can send a message with, for example, + the following content to a control bus: + + + + + + This sub-element is a convenience only, and is the equivalent of specifying: + + +]]> + For more examples for configuring header enrichers, see diff --git a/src/reference/docbook/message-store.xml b/src/reference/docbook/message-store.xml index 1a1a38297e..87976f6349 100644 --- a/src/reference/docbook/message-store.xml +++ b/src/reference/docbook/message-store.xml @@ -75,21 +75,21 @@ For example, if one of the headers contains an instance of some Spring Bean, upon deserialization you may end up with a different instance of that bean, which directly affects some of the implicit headers created by the framework (e.g., REPLY_CHANNEL or ERROR_CHANNEL). - Currently they are not serializable, but even if they were the deserialized channel would not represent the expected instance. - As a workaround we suggest to remove bean-ref headers via a <header-filter/> - before sending a message to an endpoint backed by a persistent MessageStore. - Also, we recommend using channel names instead of channel instances when setting those types of headers, - thus allowing it to be resolved in real time by the ChannelResolver. + Currently they are not serializable, but even if they were, the deserialized channel would not represent the expected instance. - Also avoid configuration of a message-flow like this: + Beginning with Spring Integration version 3.0, this issue can be resolved with a header enricher, + configured to replace these headers with a name after registering the channel with the HeaderChannelRegistry. + + + Also when configuring a message-flow like this: gateway -> queue-channel (backed by a persistent Message Store) -> service-activator - That gateway creates a Temporary Reply Channel in the background, and it will be lost by the time the - service-activator's poller reads from the queue, because it has been deserialized by another thread on the sending side. + That gateway creates a Temporary Reply Channel, and it will be lost by the time the + service-activator's poller reads from the queue. Again, you can use the header enricher to replace the headers with a + String representation. - Nevertheless we are constantly thinking about potential improvements to the framework, such as a way to provide some - robust default serialization strategy for messages in these cases. + For more information, refer to the . diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index c10f243959..08372f48d3 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -166,6 +166,16 @@ For more information see . +

+ Header Channel Registry + + It is now possible to instruct the framework to store reply and error channels + in a registry for later resolution. This is useful for cases where + the replyChannel or errorChannel might be lost; for example + when serializing + a message. See for more information. + +