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
This commit is contained in:
Gary Russell
2013-11-15 15:21:44 +02:00
committed by Artem Bilan
parent 1d0c28852f
commit 836c8e2556
21 changed files with 1060 additions and 185 deletions

View File

@@ -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<String, MessageChannelWrapper> channels = new ConcurrentHashMap<String, DefaultHeaderChannelRegistry.MessageChannelWrapper>();
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<Entry<String, MessageChannelWrapper>> iterator = this.channels.entrySet().iterator();
long threshold = System.currentTimeMillis() - this.reaperDelay;
while (iterator.hasNext()) {
Entry<String, MessageChannelWrapper> 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;
}
}
}

View File

@@ -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();
}

View File

@@ -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();
}

View File

@@ -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) {

View File

@@ -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<String, Class<?>> elementToTypeMap = new HashMap<String, Class<?>>();
private final static Map<String, String[][]> cannedHeaderElementExpressions = new HashMap<String, String[][]>();
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<Element> 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<String, Object> 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<Element> 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.
*/

View File

@@ -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.

View File

@@ -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<Object> {
@Override
public Message<?> toMessage(Object object) throws Exception {
if (object instanceof Message<?>) {
return (Message<?>) object;

View File

@@ -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}.
*
*
* <p>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);
}

View File

@@ -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();

View File

@@ -1853,6 +1853,15 @@
</xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:element name="header-channels-to-string">
<xsd:annotation>
<xsd:documentation>
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.
</xsd:documentation>
</xsd:annotation>
</xsd:element>
<xsd:element name="error-channel" type="referenceOrValueHeaderType">
<xsd:annotation>
<xsd:documentation>

View File

@@ -0,0 +1,69 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
<int:channel id="input" />
<int:header-enricher input-channel="input" output-channel="next">
<int:header-channels-to-string />
</int:header-enricher>
<int:transformer input-channel="next">
<bean class="org.springframework.integration.channel.registry.HeaderChannelRegistryTests$Foo" />
</int:transformer>
<int:channel id="inputPolled">
<int:queue />
</int:channel>
<int:header-enricher input-channel="inputPolled" output-channel="nextPolled">
<int:header-channels-to-string />
<int:poller fixed-delay="100" />
</int:header-enricher>
<int:transformer input-channel="nextPolled">
<bean class="org.springframework.integration.channel.registry.HeaderChannelRegistryTests$Foo" />
</int:transformer>
<int:channel id="alreadyAString">
<int:queue />
</int:channel>
<int:channel id="alreadyAnotherString">
<int:queue />
</int:channel>
<int:gateway id="gatewayNoReplyChannel"
service-interface="org.springframework.integration.channel.registry.HeaderChannelRegistryTests$Gateway"
default-request-channel="inputGateNoReplyChannel" />
<int:channel id="inputGateNoReplyChannel" />
<int:header-enricher input-channel="inputGateNoReplyChannel" output-channel="nextGateNoReplyChannel">
<int:header-channels-to-string />
</int:header-enricher>
<int:transformer input-channel="nextGateNoReplyChannel">
<bean class="org.springframework.integration.channel.registry.HeaderChannelRegistryTests$Foo" />
</int:transformer>
<int:gateway id="gatewayExplicitReplyChannel"
service-interface="org.springframework.integration.channel.registry.HeaderChannelRegistryTests$Gateway"
default-request-channel="inputGateExplicitReplyChannel" default-reply-channel="reply" />
<int:channel id="inputGateExplicitReplyChannel" />
<int:header-enricher input-channel="inputGateExplicitReplyChannel" output-channel="nextGateExplicitReplyChannel">
<int:header-channels-to-string />
</int:header-enricher>
<int:transformer input-channel="nextGateExplicitReplyChannel" output-channel="reply">
<bean class="org.springframework.integration.channel.registry.HeaderChannelRegistryTests$Foo" />
</int:transformer>
<int:channel id="reply" />
</beans>

View File

@@ -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<String>("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<String>("bar"));
assertNotNull(reply);
assertTrue(reply instanceof ErrorMessage);
}
@Test
public void testAlreadyAString() {
Message<String> 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<String> 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);
}
}

View File

@@ -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<String> receive() {
return new GenericMessage<String>(new Date().toString());
}
}
}

View File

@@ -13,11 +13,17 @@
<queue />
</channel>
<header-enricher input-channel="requests1" output-channel="requests">
<header-channels-to-string />
</header-enricher>
<channel id="requests"/>
<channel id="replies"/>
<enricher id="enricher" input-channel="input"
request-channel="requests" request-timeout="1234"
reply-timeout="9876"
request-channel="requests1" request-timeout="1234"
reply-timeout="9876" reply-channel="replies"
order="99" should-clone-payload="true" output-channel="output">
<property name="name" expression="payload.sourceName"/>
<property name="age" value="42"/>

View File

@@ -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<Expression, Expression> propertyExpressions = (Map<Expression, Expression>) accessor.getPropertyValue("propertyExpressions");
for (Map.Entry<Expression, Expression> 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")

View File

@@ -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
* <code>true</code>, 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<javax.jms.Message>, InitializingBean, TrackableComponent {
public class ChannelPublishingJmsMessageListener
implements SessionAwareMessageListener<javax.jms.Message>, 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<String, Object> 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";

View File

@@ -0,0 +1,37 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jms="http://www.springframework.org/schema/integration/jms"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/jms http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd">
<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
<property name="brokerURL" value="vm://localhost"/>
</bean>
<int:channel id="input" />
<int:channel id="output">
<int:queue />
</int:channel>
<int-jms:outbound-gateway request-channel="input" reply-channel="output"
connection-factory="connectionFactory"
request-destination-name="serialized.reply.channel" />
<int-jms:inbound-gateway request-channel="foo" reply-channel="baz"
request-destination-name="serialized.reply.channel"
connection-factory="connectionFactory"/>
<int:header-enricher input-channel="foo" output-channel="bar">
<int:header-channels-to-string />
</int:header-enricher>
<int:transformer input-channel="bar" output-channel="baz"
expression="'echo:' + payload" />
<int:channel id="baz" />
</beans>

View File

@@ -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<String>("foo"));
Message<?> reply = output.receive(0);
assertNotNull(reply);
assertEquals("echo:foo", reply.getPayload());
}
}

View File

@@ -73,10 +73,10 @@
using generic <emphasis>&lt;header&gt;</emphasis> sub-elements where
you would have to provide both header 'name' and 'value', you can use
convenient sub-elements to set those values directly.
</para>
</para>
<para>
<emphasis>POJO Support</emphasis>
<emphasis role="bold">POJO Support</emphasis>
</para>
<para>
@@ -121,7 +121,7 @@
</int:header-enricher>]]></programlisting>
<para>
<emphasis>SpEL Support</emphasis>
<emphasis role="bold">SpEL Support</emphasis>
</para>
<para>
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.
</para>
<para><emphasis role="bold">Header Channel Registry</emphasis></para>
<para>
Starting with <emphasis>Spring Integration 3.0</emphasis>, a new sub-element
<code>&lt;int:header-channels-to-string/&gt;</code> is available; it has no attributes.
This converts existing <code>replyChannel</code> and <code>errorChannel</code>
headers (when they are a
<classname>MessageChannel</classname>) 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 <classname>MessageChannel</classname>, no changes are made.
</para>
<para>
Use of this functionality requires the presence of a <classname>HeaderChannelRegistry</classname>
bean. By default, the framework creates a <classname>DefaultHeaderChannelRegistry</classname>
with the default expiry (60 seconds). Channels
are removed from the registry after this time. To change this, simply define a bean
with id <code>integrationHeaderChannelRegistry</code> and configure the required delay using
a constructor argument (milliseconds).
</para>
<para>
The <classname>HeaderChannelRegistry</classname> has a <code>size()</code> method to
determine the current size of the registry. The <code>runReaper()</code> 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:
</para>
<programlisting><![CDATA["@integrationHeaderChannelRegistry.runReaper()"]]></programlisting>
<para>
This sub-element is a convenience only, and is the equivalent of specifying:
</para>
<programlisting language="xml"><![CDATA[<int:reply-channel
expression="@integrationHeaderChannelRegistry.channelToChannelName(headers.replyChannel)"/>
<int:error-channel
expression="@integrationHeaderChannelRegistry.channelToChannelName(headers.errorChannel)"/>]]></programlisting>
<tip>
For more examples for configuring header enrichers, see
<ulink url="https://github.com/SpringSource/spring-integration/wiki/Header-Enricher-Advanced-Configuration">

View File

@@ -75,21 +75,21 @@
For example, if one of the headers contains an instance of some <emphasis>Spring Bean</emphasis>, 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 <literal>&lt;header-filter/&gt;</literal>
before sending a message to an endpoint backed by a persistent <classname>MessageStore</classname>.
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 <classname>ChannelResolver</classname>.
Currently they are not serializable, but even if they were, the deserialized channel would not represent the expected instance.
</para>
<para>
Also avoid configuration of a message-flow like this:
Beginning with <emphasis>Spring Integration version 3.0</emphasis>, this issue can be resolved with a header enricher,
configured to replace these headers with a name after registering the channel with the <classname>HeaderChannelRegistry</classname>.
</para>
<para>
Also when configuring a message-flow like this:
<emphasis>gateway -&gt; queue-channel (backed by a persistent Message Store) -&gt; service-activator</emphasis>
That gateway creates a <emphasis>Temporary Reply Channel</emphasis> 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 <emphasis>Temporary Reply Channel</emphasis>, 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.
</para>
<para>
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 <xref linkend="header-enricher"/>.
</para>
</important>
</para>

View File

@@ -166,6 +166,16 @@
For more information see <xref linkend="redis" />.
</para>
</section>
<section id="3.0-hcr">
<title>Header Channel Registry</title>
<para>
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 <code>replyChannel</code> or <code>errorChannel</code> might be lost; for example
when serializing
a message. See <xref linkend="header-enricher"/> for more information.
</para>
</section>
</section>
<section id="3.0-general">