Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #505 from artembilan/INT-2605

Browse Source
This commit is contained in:
Gary Russell
2012-06-21 10:09:48 -04:00
parent 6fea731809 de73c39488
commit 0a6e44ddc5
8 changed files with 210 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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 {

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

@@ -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.
* <p/>
*
*
* <pre>
* &lt;chain&gt;
* &lt;filter ref=&quot;someFilter&quot;/&gt;
@@ -62,13 +64,13 @@ import java.util.List;
* &lt;aggregator ... /&gt;
* &lt;/chain&gt;
* </pre>
*
*
* @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<MessageHandler> 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<MessageHandler> 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) {

104
spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-2.2.xsd
View File

@@ -1389,54 +1389,70 @@
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:complexContent>
<xsd:extension base="chain-type">
<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
</xsd:extension>
</xsd:complexContent>
<xsd:choice maxOccurs="unbounded">
<xsd:element name="poller" type="basePollerType" minOccurs="0"/>
<xsd:group ref="chain-elements-group" maxOccurs="unbounded"/>
</xsd:choice>
<xsd:attributeGroup ref="inputOutputChannelGroupWithId" />
<xsd:attribute name="phase" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<xsd:documentation>
The Lifecycle attribute determining the start/stop order
of the underlying MessageHandlerChain.
</xsd:documentation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="chain-type">
<xsd:sequence>
<xsd:choice minOccurs="0" maxOccurs="unbounded">
<xsd:any processContents="strict" namespace="##other" minOccurs="0" maxOccurs="unbounded"/>
<xsd:group name="chain-elements-group">
<xsd:sequence>
<xsd:choice minOccurs="0" maxOccurs="unbounded">
<xsd:any processContents="strict" namespace="##other" minOccurs="0" maxOccurs="unbounded"/>
<xsd:element name="service-activator" type="expressionOrInnerEndpointDefinitionAware"/>
<xsd:element name="splitter" type="splitter-type"/>
<xsd:element name="transformer" type="expressionOrInnerEndpointDefinitionAware"/>
<xsd:element name="header-enricher" type="header-enricher-type"/>
<xsd:element name="header-filter" type="header-filter-type"/>
<xsd:element name="enricher" type="enricher-type"/>
<xsd:element name="filter" type="filter-type"/>
<xsd:element name="aggregator" type="aggregator-type"/>
<xsd:element name="resequencer" type="resequencer-type"/>
<xsd:element name="router" type="routerTypeChain"/>
<xsd:element name="payload-type-router" type="payloadTypeRouterTypeChain"/>
<xsd:element name="recipient-list-router" type="recipientListRouterTypeChain"/>
<xsd:element name="exception-type-router" type="exceptionTypeRouterTypeChain"/>
<xsd:element name="header-value-router" type="headerValueRouterTypeChain"/>
<xsd:element name="delayer" type="delayer-type"/>
<xsd:element name="gateway" type="innerGatewayType"/>
<xsd:element name="poller" type="basePollerType"/>
<xsd:element name="payload-serializing-transformer" type="payload-serializing-transformer-type"/>
<xsd:element name="payload-deserializing-transformer" type="payload-deserializing-transformer-type"/>
<xsd:element name="object-to-string-transformer" type="specialized-transformer-type"/>
<xsd:element name="object-to-map-transformer" type="specialized-transformer-type"/>
<xsd:element name="map-to-object-transformer" type="map-to-object-transformer-type"/>
<xsd:element name="object-to-json-transformer" type="object-to-json-transformer-type"/>
<xsd:element name="json-to-object-transformer" type="json-to-object-transformer-type"/>
<xsd:element name="claim-check-in" type="claimCheckInTypeChain"/>
<xsd:element name="claim-check-out" type="claimCheckOutTypeChain"/>
<xsd:element name="control-bus" type="control-bus-type"/>
<xsd:element name="chain" type="chain-type"/>
</xsd:choice>
<xsd:choice minOccurs="0">
<xsd:element name="outbound-channel-adapter" type="methodInvokingChannelAdapterTypeChain"/>
<xsd:element name="logging-channel-adapter" type="loggingChannelAdapterTypeChain"/>
</xsd:choice>
</xsd:sequence>
</xsd:complexType>
<xsd:element name="service-activator" type="expressionOrInnerEndpointDefinitionAware"/>
<xsd:element name="splitter" type="splitter-type"/>
<xsd:element name="transformer" type="expressionOrInnerEndpointDefinitionAware"/>
<xsd:element name="header-enricher" type="header-enricher-type"/>
<xsd:element name="header-filter" type="header-filter-type"/>
<xsd:element name="enricher" type="enricher-type"/>
<xsd:element name="filter" type="filter-type"/>
<xsd:element name="aggregator" type="aggregator-type"/>
<xsd:element name="resequencer" type="resequencer-type"/>
<xsd:element name="router" type="routerTypeChain"/>
<xsd:element name="payload-type-router" type="payloadTypeRouterTypeChain"/>
<xsd:element name="recipient-list-router" type="recipientListRouterTypeChain"/>
<xsd:element name="exception-type-router" type="exceptionTypeRouterTypeChain"/>
<xsd:element name="header-value-router" type="headerValueRouterTypeChain"/>
<xsd:element name="delayer" type="delayer-type"/>
<xsd:element name="gateway" type="innerGatewayType"/>
<xsd:element name="payload-serializing-transformer" type="payload-serializing-transformer-type"/>
<xsd:element name="payload-deserializing-transformer"
type="payload-deserializing-transformer-type"/>
<xsd:element name="object-to-string-transformer" type="specialized-transformer-type"/>
<xsd:element name="object-to-map-transformer" type="specialized-transformer-type"/>
<xsd:element name="map-to-object-transformer" type="map-to-object-transformer-type"/>
<xsd:element name="object-to-json-transformer" type="object-to-json-transformer-type"/>
<xsd:element name="json-to-object-transformer" type="json-to-object-transformer-type"/>
<xsd:element name="claim-check-in" type="claimCheckInTypeChain"/>
<xsd:element name="claim-check-out" type="claimCheckOutTypeChain"/>
<xsd:element name="control-bus" type="control-bus-type"/>
<xsd:element name="chain">
<xsd:complexType>
<xsd:sequence>
<xsd:group ref="chain-elements-group"/>
</xsd:sequence>
</xsd:complexType>
</xsd:element>
</xsd:choice>
<xsd:choice minOccurs="0">
<xsd:element name="outbound-channel-adapter" type="methodInvokingChannelAdapterTypeChain"/>
<xsd:element name="logging-channel-adapter" type="loggingChannelAdapterTypeChain"/>
</xsd:choice>
</xsd:sequence>
</xsd:group>
<xsd:element name="poller">
<xsd:annotation>

18
spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserSmartLifecycleAttributesTest.xml Normal file
View File

@@ -0,0 +1,18 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd">
<chain id="chain"
input-channel="input"
auto-startup="false"
phase="256"
send-timeout="3000">
<service-activator expression="payload"/>
</chain>
</beans:beans>

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

@@ -39,14 +39,14 @@
</chain>
<chain input-channel="pollableInput1" output-channel="output">
<poller fixed-delay="10000" />
<filter ref="typeSelector" />
<poller fixed-delay="10000" />
<service-activator ref="testHandler" />
</chain>
<chain input-channel="pollableInput2" output-channel="output">
<poller ref="topLevelPoller"/>
<service-activator ref="testHandler" />
<poller ref="topLevelPoller"/>
</chain>
<poller id="topLevelPoller" fixed-delay="5000" />
@@ -86,7 +86,7 @@
<claim-check-in/>
<claim-check-out/>
</chain>
<channel id="claimCheckOutput">
<queue/>
</channel>
@@ -132,7 +132,7 @@
<beans:constructor-arg value="1" />
<beans:property name="replyMessageText" value="foo" />
</beans:bean>
<beans:bean id="messageStore" class="org.springframework.integration.store.SimpleMessageStore"/>
</beans:beans>

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

@@ -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) {

7
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpConfigOutboundGatewayTests.java
View File

@@ -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());

3
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java
View File

@@ -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);