diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ChainParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ChainParser.java index e18aa5c8c4..4eb4774158 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/xml/ChainParser.java +++ b/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; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java index c108886c3c..8202ab183b 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java +++ b/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 null, 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 handlers = this.handlers; + Assert.isTrue(this.handlers.size() == new HashSet(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"); diff --git a/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-1.0.xsd b/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-1.0.xsd index 1c66979a44..93c462881a 100644 --- a/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-1.0.xsd +++ b/org.springframework.integration/src/main/resources/org/springframework/integration/config/xml/spring-integration-1.0.xsd @@ -479,14 +479,6 @@ - - - - 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. - - - @@ -513,6 +505,14 @@ + + + + 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. + + + @@ -620,14 +620,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/org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests-context.xml b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests-context.xml index 23935bc934..f07794b300 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests-context.xml +++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests-context.xml @@ -73,6 +73,11 @@ + + + + + diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests.java index 53a33ec41e..b38f674079 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests.java +++ b/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 diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java index 8f1bed943e..8b8f48c80d 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/handler/MessageHandlerChainTests.java +++ b/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 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 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); @@ -66,9 +79,9 @@ 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); @@ -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 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); @@ -93,15 +110,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); @@ -111,16 +132,20 @@ 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); @@ -128,6 +153,23 @@ public class MessageHandlerChainTests { 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); + 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;