INT-1772 fixed support for send-timeout for inbound-channel-adapters

spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java

@@ -47,6 +47,8 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 	private volatile PollerMetadata pollerMetadata;
 
 	private volatile boolean autoStartup = true;
+	
+	private volatile long sendTimeout = -1;
 
 	private volatile String beanName;
 
@@ -63,6 +65,10 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 	public void setSource(MessageSource<?> source) {
 		this.source = source;
 	}
+	
+	public void setSendTimeout(long sendTimeout) {
+		this.sendTimeout = sendTimeout;
+	}
 
 	public void setOutputChannel(MessageChannel outputChannel) {
 		this.outputChannel = outputChannel;
@@ -130,6 +136,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
 				this.pollerMetadata.setMaxMessagesPerPoll(1);
 			}
 			spca.setMaxMessagesPerPoll(this.pollerMetadata.getMaxMessagesPerPoll());
+			spca.setSendTimeout(this.sendTimeout);
 			spca.setTaskExecutor(this.pollerMetadata.getTaskExecutor());
 			spca.setAdviceChain(this.pollerMetadata.getAdviceChain());
 			spca.setTrigger(this.pollerMetadata.getTrigger());

spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java

@@ -41,6 +41,7 @@ public abstract class AbstractPollingInboundChannelAdapterParser extends Abstrac
 				IntegrationNamespaceUtils.BASE_PACKAGE + ".config.SourcePollingChannelAdapterFactoryBean");
 		adapterBuilder.addPropertyValue("source", source);
 		adapterBuilder.addPropertyReference("outputChannel", channelName);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(adapterBuilder, element, "send-timeout");
 		Element pollerElement = DomUtils.getChildElementByTagName(element, "poller");
 		if (pollerElement != null) {
 			IntegrationNamespaceUtils.configurePollerMetadata(pollerElement, adapterBuilder, parserContext);

spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd

@@ -809,6 +809,15 @@ endpoint itself is a Polling Consumer for a channel with a queue.
 
 	<xsd:attributeGroup name="channelAdapterAttributes">
 		<xsd:attribute name="id" type="xsd:ID" />
+		<xsd:attribute name="send-timeout" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+					Maximum amount of time in milliseconds to wait when sending a message to the channel if such channel 
+					is a Queue Channel since Queue Channel is the only channel that has the concept capacity. If maximum capasity is reached
+					the messages will be rejected by the Queue Channel.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
 		<xsd:attribute name="channel" type="xsd:string">
 			<xsd:annotation>
 				<xsd:appinfo>

spring-integration-core/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests-context.xml

@@ -5,7 +5,7 @@
 	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/spring-integration-2.0.xsd">
 
 	<channel id="queueChannel">
 		<queue capacity="1"/>
@@ -19,9 +19,19 @@
 
 	<outbound-channel-adapter id="methodInvokingConsumer" ref="testBean" method="store"/>
 
-	<inbound-channel-adapter id="methodInvokingSource" ref="testBean" method="getMessage" channel="queueChannel" auto-startup="false">
+	<inbound-channel-adapter id="methodInvokingSource" ref="testBean" method="getMessage" 
+							 channel="queueChannel" auto-startup="false">
 		<poller max-messages-per-poll="1" fixed-delay="10000"/>
 	</inbound-channel-adapter>
+	
+	<channel id="withTimeoutChannel">
+		<queue/>
+	</channel>
+	<inbound-channel-adapter id="methodInvokingSourceWithTimeout" ref="testBean" method="getMessage" 
+							 channel="withTimeoutChannel" auto-startup="false"
+							 send-timeout="999">
+		<poller max-messages-per-poll="1" fixed-rate="800"/>
+	</inbound-channel-adapter>
 
 	<inbound-channel-adapter id="methodInvokingSourceWithHeaders" ref="testBean" method="getMessage" channel="queueChannelForHeadersTest" auto-startup="false">
 		<poller max-messages-per-poll="1" fixed-delay="10000"/>

spring-integration-core/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java

@@ -36,6 +36,7 @@ import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
 import org.springframework.integration.support.channel.ChannelResolutionException;
+import org.springframework.integration.test.util.TestUtils;
 
 /**
  * @author Mark Fisher
@@ -220,6 +221,17 @@ public class ChannelAdapterParserTests {
 		BeanFactoryChannelResolver channelResolver = new BeanFactoryChannelResolver(this.applicationContext);
 		channelResolver.resolveChannelName("methodInvokingSource");
 	}
+	
+	@Test
+	public void methodInvokingSourceWithSendTimeout() throws Exception{
+		String beanName = "methodInvokingSourceWithTimeout";
+		
+		SourcePollingChannelAdapter adapter = 
+			this.applicationContext.getBean(beanName, SourcePollingChannelAdapter.class);
+		assertNotNull(adapter);
+		long sendTimeout = TestUtils.getPropertyValue(adapter, "messagingTemplate.sendTimeout", Long.class);
+		assertEquals(999, sendTimeout);
+	}
 
 	public static class SampleBean{
 		private String message = "hello";