From de73c39488415476cc627e35a51fbf4d825b0dac Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 19 Jun 2012 16:02:56 +0300 Subject: [PATCH] INT-2605: add SmartLifecycle support for 'chain' * polishing TCP test to use SmartLifecycle from 'chain' * polishing XSD to exclude using 'poller' element inside 'nested-chain' INT-2605: eliminate breaking change in the XSD --- .../integration/config/xml/ChainParser.java | 32 +++--- .../handler/MessageHandlerChain.java | 104 +++++++++++++++++- .../config/xml/spring-integration-2.2.xsd | 104 ++++++++++-------- ...hainParserSmartLifecycleAttributesTest.xml | 18 +++ .../config/ChainParserTests-context.xml | 8 +- .../integration/config/ChainParserTests.java | 11 ++ .../ip/tcp/TcpConfigOutboundGatewayTests.java | 7 -- .../ip/tcp/TcpSendingMessageHandlerTests.java | 3 - 8 files changed, 210 insertions(+), 77 deletions(-) create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserSmartLifecycleAttributesTest.xml 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 b9aeb5e211..0e8681e3c0 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 @@ -1,17 +1,14 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2012 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. - * You may obtain a copy of the License at + * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on + * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the + * specific language governing permissions and limitations under the License. */ package org.springframework.integration.config.xml; @@ -26,21 +23,22 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.support.BeanDefinitionReaderUtils; import org.springframework.beans.factory.support.ManagedList; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.handler.MessageHandlerChain; /** * Parser for the <chain> element. - * + * * @author Mark Fisher * @author Iwein Fuld * @author Oleg Zhurakousky + * @author Artem Bilan */ public class ChainParser extends AbstractConsumerEndpointParser { @Override - @SuppressWarnings({ "unchecked", "rawtypes" }) + @SuppressWarnings("unchecked") protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) { - BeanDefinitionBuilder builder = BeanDefinitionBuilder - .genericBeanDefinition(IntegrationNamespaceUtils.BASE_PACKAGE + ".handler.MessageHandlerChain"); + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(MessageHandlerChain.class); ManagedList handlerList = new ManagedList(); NodeList children = element.getChildNodes(); for (int i = 0; i < children.getLength(); i++) { @@ -55,17 +53,19 @@ public class ChainParser extends AbstractConsumerEndpointParser { } else { handlerList.add(holder); - } + } } } builder.addPropertyValue("handlers", handlerList); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase"); return builder; } private BeanDefinitionHolder parseChild(Element element, ParserContext parserContext, BeanDefinition parentDefinition) { BeanDefinitionHolder holder = null; - if (element.getLocalName().equals("bean")) { + if ("bean".equals(element.getLocalName())) { holder = parserContext.getDelegate().parseBeanDefinitionElement(element, parentDefinition); } else { 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 7f2ee38f91..4f877e2c62 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 @@ -20,6 +20,7 @@ import org.springframework.aop.framework.Advised; import org.springframework.aop.support.AopUtils; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; +import org.springframework.context.SmartLifecycle; import org.springframework.core.Ordered; import org.springframework.integration.Message; import org.springframework.integration.MessageChannel; @@ -34,6 +35,7 @@ import org.springframework.util.Assert; import java.util.HashSet; import java.util.List; +import java.util.concurrent.locks.ReentrantLock; /** * A composite {@link MessageHandler} implementation that invokes a chain of @@ -53,7 +55,7 @@ import java.util.List; * This component can be used from the namespace to improve the readability of * the configuration by removing channels that can be created implicitly. *

- * + * * 

  * <chain>
  *     <filter ref="someFilter"/>
@@ -62,13 +64,13 @@ import java.util.List;
  *     <aggregator ... />
  * </chain>
  *
- * + * * @author Mark Fisher * @author Iwein Fuld * @author Gary Russell * @author Artem Bilan */ -public class MessageHandlerChain extends AbstractMessageHandler implements MessageProducer { +public class MessageHandlerChain extends AbstractMessageHandler implements MessageProducer, SmartLifecycle { private volatile List handlers; @@ -87,6 +89,13 @@ public class MessageHandlerChain extends AbstractMessageHandler implements Messa private final Object initializationMonitor = new Object(); + private volatile boolean autoStartup = true; + + private volatile int phase = Integer.MAX_VALUE; + + private volatile boolean running; + + private final ReentrantLock lifecycleLock = new ReentrantLock(); public void setHandlers(List handlers) { this.handlers = handlers; @@ -195,6 +204,95 @@ public class MessageHandlerChain extends AbstractMessageHandler implements Messa } } + /** + * SmartLifecycle implementation (delegates to the {@link #handlers}) + */ + + public final boolean isAutoStartup() { + return this.autoStartup; + } + + public final int getPhase() { + return this.phase; + } + + public final boolean isRunning() { + this.lifecycleLock.lock(); + try { + return this.running; + } + finally { + this.lifecycleLock.unlock(); + } + } + + public final void start() { + this.lifecycleLock.lock(); + try { + if (!this.running) { + this.doStart(); + this.running = true; + if (logger.isInfoEnabled()) { + logger.info("started " + this); + } + } + } + finally { + this.lifecycleLock.unlock(); + } + } + + public final void stop() { + this.lifecycleLock.lock(); + try { + if (this.running) { + this.doStop(); + this.running = false; + if (logger.isInfoEnabled()) { + logger.info("stopped " + this); + } + } + } + finally { + this.lifecycleLock.unlock(); + } + } + + public final void stop(Runnable callback) { + this.lifecycleLock.lock(); + try { + this.stop(); + callback.run(); + } + finally { + this.lifecycleLock.unlock(); + } + } + + public void setAutoStartup(boolean autoStartup) { + this.autoStartup = autoStartup; + } + + public void setPhase(int phase) { + this.phase = phase; + } + + private void doStop() { + for (MessageHandler handler : this.handlers) { + if (handler instanceof SmartLifecycle) { + ((SmartLifecycle) handler).stop(); + } + } + } + + private void doStart() { + for (MessageHandler handler : this.handlers) { + if (handler instanceof SmartLifecycle) { + ((SmartLifecycle) handler).start(); + } + } + } + private class ReplyForwardingMessageChannel implements MessageChannel { public boolean send(Message message) { diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd index 88cc7f138c..a5c67678f1 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd @@ -1389,54 +1389,70 @@ - - - - - + + + + + + + + + + The Lifecycle attribute determining the start/stop order + of the underlying MessageHandlerChain. + + + + - - - - + + + + - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserSmartLifecycleAttributesTest.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserSmartLifecycleAttributesTest.xml new file mode 100644 index 0000000000..dcf39c7da9 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserSmartLifecycleAttributesTest.xml @@ -0,0 +1,18 @@ + + + + + + + + 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 09fc426c63..99b3efbd0e 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 @@ -39,14 +39,14 @@ - + - + @@ -86,7 +86,7 @@ - + @@ -132,7 +132,7 @@ - + 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 5c7860cea5..782f073231 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 @@ -34,6 +34,7 @@ import org.springframework.integration.handler.AbstractReplyProducingMessageHand import org.springframework.integration.handler.MessageHandlerChain; import org.springframework.integration.message.MessageMatcher; import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; import org.springframework.util.StringUtils; @@ -289,6 +290,16 @@ public class ChainParserTests { } } + @Test //INT-2605 + public void checkSmartLifecycleConfig() { + ApplicationContext ctx = new ClassPathXmlApplicationContext("ChainParserSmartLifecycleAttributesTest.xml", this.getClass()); + MessageHandlerChain handlerChain = ctx.getBean(MessageHandlerChain.class); + assertEquals(false, handlerChain.isAutoStartup()); + assertEquals(256, handlerChain.getPhase()); + assertEquals(3000L, TestUtils.getPropertyValue(handlerChain, "sendTimeout")); + assertEquals(false, TestUtils.getPropertyValue(handlerChain, "running")); + } + public static class StubHandler extends AbstractReplyProducingMessageHandler { @Override protected Object handleRequestMessage(Message requestMessage) { diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigOutboundGatewayTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigOutboundGatewayTests.java index d1a9450ff2..411551fece 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigOutboundGatewayTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigOutboundGatewayTests.java @@ -29,7 +29,6 @@ import org.springframework.integration.MessageChannel; import org.springframework.integration.core.PollableChannel; import org.springframework.integration.core.SubscribableChannel; import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; -import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; import org.springframework.integration.support.MessageBuilder; import org.springframework.test.context.ContextConfiguration; @@ -112,9 +111,6 @@ public class TcpConfigOutboundGatewayTests { @Qualifier("requestChannelNio") SubscribableChannel requestChannelNio; - @Autowired - AbstractClientConnectionFactory crLfClient2; - @Autowired MessageChannel tcpOutboundGatewayInsideChain; @@ -177,9 +173,6 @@ public class TcpConfigOutboundGatewayTests { @Test //INT-1029 public void testOutboundInsideChain() throws Exception { -// TODO Lifecycle#start() isn't invoked within chain... - crLfClient2.start(); - tcpOutboundGatewayInsideChain.send(MessageBuilder.withPayload("test").build()); byte[] bytes = (byte[]) replyChannel.receive().getPayload(); assertEquals("echo:test", new String(bytes).trim()); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java index 62088483c2..a99138a6ce 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java @@ -1067,9 +1067,6 @@ public class TcpSendingMessageHandlerTests { public void testOutboundChannelAdapterWithinChain() throws Exception { ApplicationContext ctx = new ClassPathXmlApplicationContext( "TcpOutboundChannelAdapterWithinChainTests-context.xml", this.getClass()); - AbstractConnectionFactory ccf = ctx.getBean("ccf", AbstractConnectionFactory.class); -// TODO Lifecycle#start() isn't invoked within chain... - ccf.start(); AbstractServerConnectionFactory scf = ctx.getBean(AbstractServerConnectionFactory.class); TestingUtilities.waitListening(scf, null); MessageChannel channelAdapterWithinChain = ctx.getBean("tcpOutboundChannelAdapterWithinChain", MessageChannel.class);