INT-1165, INT-1175 The 'send-timeout' on a chain element (if provided explicitly) will now take precedence over the send-timeout of its last handler (assuming it's a producer in the first place). Also, duplicate handlers are now rejected in a chain.

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

@@ -60,6 +60,7 @@ public class ChainParser extends AbstractConsumerEndpointParser {
 			}
 		}
 		builder.addPropertyValue("handlers", handlerList);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout");
 		return builder;
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.handler;
 
+import java.util.HashSet;
 import java.util.List;
 
 import org.springframework.core.Ordered;
@@ -65,6 +66,13 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes
 
 	private volatile MessageChannel outputChannel;
 
+	/**
+	 * If the sendTimeout is configured explicitly on this chain instance, it will
+	 * take precedence over the actual settings on the final handler in the chain.
+	 * By default, it is <code>null</code>, so the actual handler configuration is used.
+	 */
+	private volatile Long sendTimeout = null;
+
 	private volatile int order = Ordered.LOWEST_PRECEDENCE;
 
 	private volatile boolean initialized;
@@ -80,6 +88,10 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes
 		this.outputChannel = outputChannel;
 	}
 
+	public void setSendTimeout(long sendTimeout) {
+		this.sendTimeout = sendTimeout;
+	}
+
 	public void setOrder(int order) {
 		this.order = order;
 	}
@@ -112,8 +124,9 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes
 	}
 
 	private void configureChain() {
-		List<MessageHandler> handlers = this.handlers;
-		for (int i = 0; i < handlers.size(); i++) {
+		Assert.isTrue(this.handlers.size() == new HashSet<MessageHandler>(this.handlers).size(),
+				"duplicate handlers are not allowed in a chain");
+		for (int i = 0; i < this.handlers.size(); i++) {
 			MessageHandler handler = handlers.get(i);
 			if (i < handlers.size() - 1) { // not the last handler
 				Assert.isTrue(handler instanceof MessageProducer, "All handlers except for " +
@@ -131,8 +144,7 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes
 				((MessageProducer) handler).setOutputChannel(nextChannel);
 			}
 			else if (handler instanceof MessageProducer) {
-				MessageChannel replyChannel = (this.outputChannel != null) ? this.outputChannel
-						: new ReplyForwardingMessageChannel();
+				MessageChannel replyChannel = new ReplyForwardingMessageChannel();
 				((MessageProducer) handler).setOutputChannel(replyChannel);
 			}
 			else {
@@ -151,6 +163,11 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes
 		}
 
 		public boolean send(Message<?> message, long timeout) {
+			timeout = (MessageHandlerChain.this.sendTimeout != null)
+					? MessageHandlerChain.this.sendTimeout : timeout;
+			if (MessageHandlerChain.this.outputChannel != null) {
+				return MessageHandlerChain.this.outputChannel.send(message, timeout);
+			}
 			Object replyChannelHeader = message.getHeaders().getReplyChannel();
 			if (replyChannelHeader == null) {
 				throw new MessageHandlingException(message, "no replyChannel header available");

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

@@ -760,16 +760,6 @@
 						</xsd:appinfo>
 					</xsd:annotation>
 				</xsd:attribute>
-				<xsd:attribute name="send-timeout" type="xsd:string">
-					<xsd:annotation>
-						<xsd:documentation>
-							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. 
-						</xsd:documentation>
-					</xsd:annotation>
-				</xsd:attribute>
 				<xsd:attribute name="method" type="xsd:string">
 					<xsd:annotation>
 						<xsd:appinfo>
@@ -803,6 +793,14 @@
 						</xsd:appinfo>
 					</xsd:annotation>
 				</xsd:attribute>
+				<xsd:attribute name="send-timeout" type="xsd:string">
+					<xsd:annotation>
+						<xsd:documentation>
+	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. 
+						</xsd:documentation>
+					</xsd:annotation>
+				</xsd:attribute>
 			</xsd:extension>
 		</xsd:complexContent>
 	</xsd:complexType>
@@ -942,16 +940,6 @@
 							</xsd:appinfo>
 						</xsd:annotation>
 					</xsd:attribute>
-					<xsd:attribute name="send-timeout" type="xsd:string">
-						<xsd:annotation>
-							<xsd:documentation>
-								Specify the maximum amount of time in milliseconds to wait when sending
-								the released
-								Messages (after delay) to the output channel. By default the send will
-								block indefinitely. 
-							</xsd:documentation>
-						</xsd:annotation>
-					</xsd:attribute>
 					<xsd:attribute name="wait-for-tasks-to-complete-on-shutdown"
 						type="xsd:string">
 						<xsd:annotation>

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

@@ -74,6 +74,11 @@
 		</payload-type-router>
 	</chain>
 
+	<chain id="chainWithSendTimeout" input-channel="chainWithSendTimeoutInput" output-channel="output" send-timeout="9876">
+		<filter ref="typeSelector" />
+		<service-activator ref="testHandler" />
+	</chain>
+
 	<channel id="strings">
 		<queue/>
 	</channel>

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

@@ -27,12 +27,16 @@ import org.hamcrest.Factory;
 import org.hamcrest.Matcher;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+
+import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.ApplicationContext;
 import org.springframework.integration.channel.PollableChannel;
 import org.springframework.integration.core.Message;
 import org.springframework.integration.core.MessageChannel;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
+import org.springframework.integration.handler.MessageHandlerChain;
 import org.springframework.integration.message.MessageBuilder;
 import org.springframework.integration.message.MessageMatcher;
 import org.springframework.test.context.ContextConfiguration;
@@ -88,6 +92,9 @@ public class ChainParserTests {
 	@Autowired
 	private PollableChannel numbers;
 
+	@Autowired
+	private ApplicationContext context;
+
 	public static Message<?> successMessage = MessageBuilder.withPayload("success").build();
 
 	@Factory
@@ -175,6 +182,15 @@ public class ChainParserTests {
 		assertEquals(123, reply2.getPayload());
 	}
 
+	@Test // INT-1165
+	public void chainWithSendTimeout() {
+		Object endpoint = this.context.getBean("chainWithSendTimeout");
+		MessageHandlerChain chain = (MessageHandlerChain) new DirectFieldAccessor(endpoint).getPropertyValue("handler");
+		long sendTimeout = ((Long) new DirectFieldAccessor(chain).getPropertyValue("sendTimeout")).longValue();
+		assertEquals(9876, sendTimeout);
+	}
+
+
 	public static class StubHandler extends AbstractReplyProducingMessageHandler {
 		@Override
 		protected Object handleRequestMessage(Message<?> requestMessage) {

spring-integration-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2008 the original author or authors.
+ * Copyright 2002-2010 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.
@@ -40,22 +40,34 @@ public class MessageHandlerChainTests {
 
 	private Message<String> message = MessageBuilder.withPayload("foo").build();
 
-	private MessageHandler handler = createMock(MessageHandler.class);
+	private MessageHandler handler1 = createMock(MessageHandler.class);
 
-	private ProducingHandlerStub producer = new ProducingHandlerStub(handler);
+	private MessageHandler handler2 = createMock(MessageHandler.class);
 
-	private Object[] allMocks = new Object[] { outputChannel, handler };
+	private MessageHandler handler3 = createMock(MessageHandler.class);
+
+	private ProducingHandlerStub producer1 = new ProducingHandlerStub(handler1);
+
+	private ProducingHandlerStub producer2 = new ProducingHandlerStub(handler2);
+
+	private ProducingHandlerStub producer3 = new ProducingHandlerStub(handler3);
+
+	private Object[] allMocks = new Object[] { outputChannel, handler1, handler2, handler3 };
 
 	@Test
 	public void chainWithOutputChannel() {
-		handler.handleMessage(message);
-		expectLastCall().times(3);
-		expect(outputChannel.send(eq(message))).andReturn(true);
+		handler1.handleMessage(message);
+		expectLastCall();
+		handler2.handleMessage(message);
+		expectLastCall();
+		handler3.handleMessage(message);
+		expectLastCall();
+		expect(outputChannel.send(eq(message), eq(-1L))).andReturn(true);
 		replay(allMocks);
 		List<MessageHandler> handlers = new ArrayList<MessageHandler>();
-		handlers.add(producer);
-		handlers.add(producer);
-		handlers.add(producer);
+		handlers.add(producer1);
+		handlers.add(producer2);
+		handlers.add(producer3);
 		MessageHandlerChain chain = new MessageHandlerChain();
 		chain.setBeanName("testChain");
 		chain.setHandlers(handlers);
@@ -67,24 +79,29 @@ public class MessageHandlerChainTests {
 	public void chainWithOutputChannelButLastHandlerDoesNotProduceReplies() {
 		replay(allMocks);
 		List<MessageHandler> handlers = new ArrayList<MessageHandler>();
-		handlers.add(producer);
-		handlers.add(producer);
-		handlers.add(handler);
+		handlers.add(producer1);
+		handlers.add(producer2);
+		handlers.add(handler3);
 		MessageHandlerChain chain = new MessageHandlerChain();
 		chain.setBeanName("testChain");
 		chain.setHandlers(handlers);
 		chain.setOutputChannel(outputChannel);
 		chain.handleMessage(message);
 	}
+
 	@Test
 	public void chainWithoutOutputChannelButLastHandlerDoesNotProduceReplies() {
-		handler.handleMessage(message);
-		expectLastCall().times(3);
+		handler1.handleMessage(message);
+		expectLastCall();
+		handler2.handleMessage(message);
+		expectLastCall();
+		handler3.handleMessage(message);
+		expectLastCall();
 		replay(allMocks);
 		List<MessageHandler> handlers = new ArrayList<MessageHandler>();
-		handlers.add(producer);
-		handlers.add(producer);
-		handlers.add(handler);
+		handlers.add(producer1);
+		handlers.add(producer2);
+		handlers.add(handler3);
 		MessageHandlerChain chain = new MessageHandlerChain();
 		chain.setBeanName("testChain");
 		chain.setHandlers(handlers);
@@ -94,15 +111,19 @@ public class MessageHandlerChainTests {
 	@Test
 	public void chainForwardsToReplyChannel() {
 		Message<String> message = MessageBuilder.withPayload("test").setReplyChannel(outputChannel).build();
-		handler.handleMessage(message);
-		expectLastCall().times(3);
+		handler1.handleMessage(message);
+		expectLastCall();
+		handler2.handleMessage(message);
+		expectLastCall();
+		handler3.handleMessage(message);
+		expectLastCall();
 		//equality is lost when recreating the message
 		expect(outputChannel.send(isA(Message.class))).andReturn(true);
 		replay(allMocks);
 		List<MessageHandler> handlers = new ArrayList<MessageHandler>();
-		handlers.add(producer);
-		handlers.add(producer);
-		handlers.add(producer);
+		handlers.add(producer1);
+		handlers.add(producer2);
+		handlers.add(producer3);
 		MessageHandlerChain chain = new MessageHandlerChain();
 		chain.setBeanName("testChain");
 		chain.setHandlers(handlers);
@@ -112,16 +133,36 @@ public class MessageHandlerChainTests {
 	@Test
 	public void chainResolvesReplyChannelName() {
 		Message<String> message = MessageBuilder.withPayload("test").setReplyChannelName("testChannel").build();
-		handler.handleMessage(message);
-		expectLastCall().times(3);
+		handler1.handleMessage(message);
+		expectLastCall();
+		handler2.handleMessage(message);
+		expectLastCall();
+		handler3.handleMessage(message);
+		expectLastCall();
 		expect(outputChannel.send(eq(message))).andReturn(true);
 		replay(allMocks);
 		DefaultListableBeanFactory beanFactory = new DefaultListableBeanFactory();
 		beanFactory.registerSingleton("testChannel", outputChannel);
 		List<MessageHandler> handlers = new ArrayList<MessageHandler>();
-		handlers.add(producer);
-		handlers.add(producer);
-		handlers.add(producer);
+		handlers.add(producer1);
+		handlers.add(producer2);
+		handlers.add(producer3);
+		MessageHandlerChain chain = new MessageHandlerChain();
+		chain.setBeanName("testChain");
+		chain.setHandlers(handlers);
+		chain.setBeanFactory(beanFactory);
+		chain.handleMessage(message);
+	}
+
+	@Test(expected = IllegalArgumentException.class) // INT-1175
+	public void chainRejectsDuplicateHandlers() {
+		Message<String> message = MessageBuilder.withPayload("test").setReplyChannelName("testChannel").build();
+		DefaultListableBeanFactory beanFactory = new DefaultListableBeanFactory();
+		beanFactory.registerSingleton("testChannel", outputChannel);
+		List<MessageHandler> handlers = new ArrayList<MessageHandler>();
+		handlers.add(producer1);
+		handlers.add(producer2);
+		handlers.add(producer1);
 		MessageHandlerChain chain = new MessageHandlerChain();
 		chain.setBeanName("testChain");
 		chain.setHandlers(handlers);