diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ChainParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ChainParser.java index bc06ce109f..82d18e4b49 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ChainParser.java +++ b/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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java index bfe758f6e8..a157c3bebd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java +++ b/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 null, 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 handlers = this.handlers; - for (int i = 0; i < handlers.size(); i++) { + Assert.isTrue(this.handlers.size() == new HashSet(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"); diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd index 6c20c92071..c7efe1c4d0 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.0.xsd @@ -760,16 +760,6 @@ - - - - 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. - - - @@ -803,6 +793,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. + + + @@ -942,16 +940,6 @@ - - - - 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. - - - diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests-context.xml index 510b836e86..5247b60953 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests-context.xml @@ -74,6 +74,11 @@ + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests.java index 8be7be1146..8f40d7e5eb 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests.java +++ b/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) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java index b8a818d5b9..e433b6755e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java +++ b/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 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 handlers = new ArrayList(); - 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 handlers = new ArrayList(); - 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 handlers = new ArrayList(); - 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 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 handlers = new ArrayList(); - 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 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 handlers = new ArrayList(); - 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 message = MessageBuilder.withPayload("test").setReplyChannelName("testChannel").build(); + DefaultListableBeanFactory beanFactory = new DefaultListableBeanFactory(); + beanFactory.registerSingleton("testChannel", outputChannel); + List handlers = new ArrayList(); + handlers.add(producer1); + handlers.add(producer2); + handlers.add(producer1); MessageHandlerChain chain = new MessageHandlerChain(); chain.setBeanName("testChain"); chain.setHandlers(handlers);