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.

org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ChainParser.java

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

org.springframework.integration/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2009 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.
@@ -16,6 +16,7 @@
 
 package org.springframework.integration.handler;
 
+import java.util.HashSet;
 import java.util.List;
 
 import org.springframework.beans.BeanWrapperImpl;
@@ -71,6 +72,8 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes
 
 	private volatile MessageChannel outputChannel;
 
+	private volatile Long sendTimeout = null;
+
 	private volatile int order = Ordered.LOWEST_PRECEDENCE;
 
 	private volatile boolean initialized;
@@ -86,6 +89,15 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes
 		this.outputChannel = 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.
+ 	 */
+	public void setSendTimeout(long sendTimeout) {
+		this.sendTimeout = sendTimeout;
+	}
+
 	public void setOrder(int order) {
 		this.order = order;
 	}
@@ -114,6 +126,8 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes
 	private void configureChain() {
 		DirectChannel channel = null;
 		List<MessageHandler> handlers = this.handlers;
+		Assert.isTrue(this.handlers.size() == new HashSet<MessageHandler>(this.handlers).size(),
+				"duplicate handlers are not allowed in a chain");
 		for (int i = 0; i < handlers.size(); i++) {
 			boolean first = (i == 0);
 			boolean last = (i == handlers.size() - 1);
@@ -132,8 +146,7 @@ public class MessageHandlerChain extends IntegrationObjectSupport implements Mes
 				accessor.setPropertyValue(OUTPUT_CHANNEL_PROPERTY, channel);
 			}
 			else if (accessor.getPropertyType(OUTPUT_CHANNEL_PROPERTY) != null) {
-				MessageChannel replyChannel = (this.outputChannel != null) ? this.outputChannel
-						: new ReplyForwardingMessageChannel();
+				MessageChannel replyChannel = new ReplyForwardingMessageChannel();
 				accessor.setPropertyValue(OUTPUT_CHANNEL_PROPERTY, replyChannel);
 			}
 			else {
@@ -156,6 +169,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");

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

@@ -479,14 +479,6 @@
 					</xsd:annotation>
 				</xsd:attribute>
 				<xsd:attribute name="method" type="xsd:string" />
-				<xsd:attribute name="send-timeout" type="xsd:string">
-					<xsd:annotation>
-						<xsd:documentation>
-	Specify the maximum amount of time in milliseconds to wait when sending reply
-	Messages 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>
@@ -513,6 +505,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 reply
+	Messages 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>
@@ -620,14 +620,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>
 							<xsd:documentation>

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

@@ -73,6 +73,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>

org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2009 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.
@@ -25,12 +25,15 @@ import java.util.List;
 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.handler.ReplyMessageHolder;
 import org.springframework.integration.message.MessageBuilder;
 import org.springframework.test.context.ContextConfiguration;
@@ -88,6 +91,10 @@ public class ChainParserTests  {
 	@Qualifier("aggregatorInput")
 	private MessageChannel aggregatorInput;
 
+	@Autowired
+	private ApplicationContext context;
+
+
 	public static Message<?> successMessage = MessageBuilder.withPayload("success").build();
 
 
@@ -172,6 +179,14 @@ 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

org.springframework.integration/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.
@@ -39,22 +39,35 @@ 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 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 };
 
-	private Object[] allMocks = new Object[] { outputChannel, handler };
 
 	@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);
@@ -66,9 +79,9 @@ 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);
@@ -77,13 +90,17 @@ public class MessageHandlerChainTests {
 	}
 	@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);
@@ -93,15 +110,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);
@@ -111,16 +132,20 @@ 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);
@@ -128,6 +153,23 @@ public class MessageHandlerChainTests {
 		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);
+		chain.setBeanFactory(beanFactory);
+		chain.handleMessage(message);
+	}
+
+
 	private class ProducingHandlerStub implements MessageHandler {
 		private MessageChannel output;
 		
@@ -136,7 +178,8 @@ public class MessageHandlerChainTests {
 		public ProducingHandlerStub(MessageHandler handler) {
 			messageHandler = handler;
 		}
-		
+
+		@SuppressWarnings("unused")
 		public void setOutputChannel(MessageChannel channel) {
 			this.output = channel;