diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/FilterFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/FilterFactoryBean.java index 6a563ea53b..b7e3e0b535 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/FilterFactoryBean.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/FilterFactoryBean.java @@ -36,6 +36,8 @@ public class FilterFactoryBean extends AbstractMessageHandlerFactoryBean { private volatile Boolean throwExceptionOnRejection; + private volatile Long sendTimeout; + public void setDiscardChannel(MessageChannel discardChannel) { this.discardChannel = discardChannel; @@ -45,6 +47,10 @@ public class FilterFactoryBean extends AbstractMessageHandlerFactoryBean { this.throwExceptionOnRejection = throwExceptionOnRejection; } + public void setSendTimeout(Long sendTimeout) { + this.sendTimeout = sendTimeout; + } + @Override MessageHandler createMethodInvokingHandler(Object targetObject, String targetMethodName) { if (targetObject instanceof MessageSelector) { @@ -71,6 +77,9 @@ public class FilterFactoryBean extends AbstractMessageHandlerFactoryBean { if (this.discardChannel != null) { filter.setDiscardChannel(discardChannel); } + if (this.sendTimeout != null) { + filter.setSendTimeout(this.sendTimeout.longValue()); + } return filter; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/RouterFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/RouterFactoryBean.java index 9c715f042c..67a79ad26d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/RouterFactoryBean.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/RouterFactoryBean.java @@ -69,27 +69,12 @@ public class RouterFactoryBean extends AbstractMessageHandlerFactoryBean { MessageHandler createMethodInvokingHandler(Object targetObject, String targetMethodName) { Assert.notNull(targetObject, "target object must not be null"); AbstractMessageRouter router = this.createRouter(targetObject, targetMethodName); - if (this.defaultOutputChannel != null) { - router.setDefaultOutputChannel(this.defaultOutputChannel); - } - if (this.timeout != null) { - router.setTimeout(timeout.longValue()); - } - if (this.ignoreChannelNameResolutionFailures != null) { - Assert.isTrue(router instanceof AbstractChannelNameResolvingMessageRouter, - "The 'ignoreChannelNameResolutionFailures' property can only be set on routers that extend " - + AbstractChannelNameResolvingMessageRouter.class.getName()); - ((AbstractChannelNameResolvingMessageRouter) router).setIgnoreChannelNameResolutionFailures(ignoreChannelNameResolutionFailures); - } - if (this.resolutionRequired != null) { - router.setResolutionRequired(this.resolutionRequired); - } - return router; + return this.configureRouter(router); } @Override MessageHandler createExpressionEvaluatingHandler(String expression) { - return new ExpressionEvaluatingRouter(expression); + return this.configureRouter(new ExpressionEvaluatingRouter(expression)); } private AbstractMessageRouter createRouter(Object targetObject, String targetMethodName) { @@ -108,4 +93,23 @@ public class RouterFactoryBean extends AbstractMessageHandlerFactoryBean { return router; } + private AbstractMessageRouter configureRouter(AbstractMessageRouter router) { + if (this.defaultOutputChannel != null) { + router.setDefaultOutputChannel(this.defaultOutputChannel); + } + if (this.timeout != null) { + router.setTimeout(timeout.longValue()); + } + if (this.ignoreChannelNameResolutionFailures != null) { + Assert.isTrue(router instanceof AbstractChannelNameResolvingMessageRouter, + "The 'ignoreChannelNameResolutionFailures' property can only be set on routers that extend " + + AbstractChannelNameResolvingMessageRouter.class.getName()); + ((AbstractChannelNameResolvingMessageRouter) router).setIgnoreChannelNameResolutionFailures(ignoreChannelNameResolutionFailures); + } + if (this.resolutionRequired != null) { + router.setResolutionRequired(this.resolutionRequired); + } + return router; + } + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java index 4452270ca8..1a3b6547ee 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/SplitterFactoryBean.java @@ -30,24 +30,41 @@ import org.springframework.util.StringUtils; */ public class SplitterFactoryBean extends AbstractMessageHandlerFactoryBean { + private volatile Long sendTimeout; + + public void setSendTimeout(Long sendTimeout) { + this.sendTimeout = sendTimeout; + } + @Override MessageHandler createMethodInvokingHandler(Object targetObject, String targetMethodName) { + AbstractMessageSplitter splitter = null; if (targetObject instanceof AbstractMessageSplitter) { - return (AbstractMessageSplitter) targetObject; + splitter = (AbstractMessageSplitter) targetObject; } - return (StringUtils.hasText(targetMethodName)) - ? new MethodInvokingSplitter(targetObject, targetMethodName) - : new MethodInvokingSplitter(targetObject); + else { + splitter = (StringUtils.hasText(targetMethodName)) + ? new MethodInvokingSplitter(targetObject, targetMethodName) + : new MethodInvokingSplitter(targetObject); + } + return this.configureSplitter(splitter); } @Override MessageHandler createExpressionEvaluatingHandler(String expression) { - return new ExpressionEvaluatingSplitter(expression); + return this.configureSplitter(new ExpressionEvaluatingSplitter(expression)); } @Override MessageHandler createDefaultHandler() { - return new DefaultMessageSplitter(); + return this.configureSplitter(new DefaultMessageSplitter()); + } + + private AbstractMessageSplitter configureSplitter(AbstractMessageSplitter splitter) { + if (this.sendTimeout != null) { + splitter.setSendTimeout(sendTimeout); + } + return splitter; } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/TransformerFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/TransformerFactoryBean.java index 6d1a2e1a90..2f7d8f737d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/TransformerFactoryBean.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/TransformerFactoryBean.java @@ -31,6 +31,12 @@ import org.springframework.util.StringUtils; */ public class TransformerFactoryBean extends AbstractMessageHandlerFactoryBean { + private volatile Long sendTimeout; + + public void setSendTimeout(Long sendTimeout) { + this.sendTimeout = sendTimeout; + } + @Override MessageHandler createMethodInvokingHandler(Object targetObject, String targetMethodName) { Assert.notNull(targetObject, "targetObject must not be null"); @@ -44,13 +50,21 @@ public class TransformerFactoryBean extends AbstractMessageHandlerFactoryBean { else { transformer = new MethodInvokingTransformer(targetObject); } - return new MessageTransformingHandler(transformer); + return this.createHandler(transformer); } @Override MessageHandler createExpressionEvaluatingHandler(String expression) { Transformer transformer = new ExpressionEvaluatingTransformer(expression); - return new MessageTransformingHandler(transformer); + return this.createHandler(transformer); + } + + private MessageTransformingHandler createHandler(Transformer transformer) { + MessageTransformingHandler handler = new MessageTransformingHandler(transformer); + if (this.sendTimeout != null) { + handler.setSendTimeout(this.sendTimeout.longValue()); + } + return handler; } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractDelegatingConsumerEndpointParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractDelegatingConsumerEndpointParser.java index 7d1f2ff166..452f182957 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractDelegatingConsumerEndpointParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/AbstractDelegatingConsumerEndpointParser.java @@ -75,6 +75,7 @@ abstract class AbstractDelegatingConsumerEndpointParser extends AbstractConsumer "a 'ref' or inner-bean definition is provided.", element); } } + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout"); this.postProcess(builder, element, parserContext); return builder; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ServiceActivatorParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ServiceActivatorParser.java index 33bfb1ed63..3b9a27e49a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ServiceActivatorParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ServiceActivatorParser.java @@ -52,6 +52,7 @@ public class ServiceActivatorParser extends AbstractConsumerEndpointParser { String method = element.getAttribute(METHOD_ATTRIBUTE); builder.getRawBeanDefinition().getConstructorArgumentValues().addGenericArgumentValue(method, "java.lang.String"); } + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout"); return builder; } diff --git a/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd b/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd index 727e593a83..7123046c55 100644 --- a/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd +++ b/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd @@ -490,6 +490,14 @@ + + + + Specify the maximum amount of time in milliseconds to wait when sending a reply + Message to the output channel. By default the send will block for one second. + + + @@ -1268,7 +1276,14 @@ - + + + + Specify the maximum amount of time in milliseconds to wait when sending Messages + to the target MessageChannels. By default the send will block indefinitely. + + + @@ -1322,7 +1337,6 @@ - - diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/SendTimeoutConfigurationTests-context.xml b/org.springframework.integration/src/test/java/org/springframework/integration/handler/SendTimeoutConfigurationTests-context.xml new file mode 100644 index 0000000000..7da8cea0d5 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/SendTimeoutConfigurationTests-context.xml @@ -0,0 +1,24 @@ + + + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/SendTimeoutConfigurationTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/handler/SendTimeoutConfigurationTests.java new file mode 100644 index 0000000000..a93a7e69d2 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/SendTimeoutConfigurationTests.java @@ -0,0 +1,74 @@ +/* + * Copyright 2002-2009 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.handler; + +import static org.junit.Assert.assertEquals; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ApplicationContext; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +/** + * @author Mark Fisher + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class SendTimeoutConfigurationTests { + + @Autowired + private ApplicationContext context; + + + @Test + public void serviceActivator() { + assertEquals(123, this.getTimeout("serviceActivator")); + } + + @Test + public void filter() { + assertEquals(123, this.getTimeout("filter")); + } + + @Test + public void transformer() { + assertEquals(123, this.getTimeout("transformer")); + } + + @Test + public void splitter() { + assertEquals(123, this.getTimeout("splitter")); + } + + @Test + public void router() { + assertEquals(123, this.getTimeout("router")); + } + + + private long getTimeout(String endpointName) { + DirectFieldAccessor endpointAccessor = new DirectFieldAccessor(context.getBean(endpointName)); + DirectFieldAccessor handlerAccessor = new DirectFieldAccessor(endpointAccessor.getPropertyValue("handler")); + DirectFieldAccessor templateAccessor = new DirectFieldAccessor(handlerAccessor.getPropertyValue("channelTemplate")); + return ((Long) templateAccessor.getPropertyValue("sendTimeout")).longValue(); + } + +}