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

INT-2867: Add Script Support to inbound-c-a

Browse Source 
* Add `<script>` and `<expression>` to `<int:inbound-channel-adapter>`
* Refactoring for `DefaultInboundChannelAdapterParser`
* Introduce `ScriptExecutingMessageSource` to allow scripts' `MessageProcessor`
implementations work as a source for polling endpoint

JIRA: https://jira.springsource.org/browse/INT-2867

INT-2867: documentation

In addition the tests for Lifecycle attributes

Polishing
This commit is contained in:
Artem Bilan
2013-09-17 21:59:41 +03:00
committed by Gary Russell
parent 63cbf736a2
commit 09be87551b
15 changed files with 344 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-integration-core/src/main/java/org/springframework/integration/config/ConsumerEndpointFactoryBean.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -66,6 +66,8 @@ public class ConsumerEndpointFactoryBean
private volatile boolean autoStartup = true;
private volatile int phase = 0;
private volatile MessageChannel inputChannel;
private volatile ConfigurableBeanFactory beanFactory;
@@ -112,6 +114,10 @@ public class ConsumerEndpointFactoryBean
this.autoStartup = autoStartup;
}
public void setPhase(int phase) {
this.phase = phase;
}
public void setBeanName(String beanName) {
this.beanName = beanName;
}
@@ -239,6 +245,7 @@ public class ConsumerEndpointFactoryBean
this.endpoint.setBeanName(this.beanName);
this.endpoint.setBeanFactory(this.beanFactory);
this.endpoint.setAutoStartup(this.autoStartup);
this.endpoint.setPhase(this.phase);
this.endpoint.afterPropertiesSet();
this.initialized = true;
}

9
spring-integration-core/src/main/java/org/springframework/integration/config/SourcePollingChannelAdapterFactoryBean.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -48,6 +48,8 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
private volatile boolean autoStartup = true;
private volatile int phase = 0;
private volatile Long sendTimeout;
private volatile String beanName;
@@ -82,6 +84,10 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
this.autoStartup = autoStartup;
}
public void setPhase(int phase) {
this.phase = phase;
}
public void setBeanFactory(BeanFactory beanFactory) {
Assert.isInstanceOf(ConfigurableBeanFactory.class, beanFactory,
"a ConfigurableBeanFactory is required");
@@ -145,6 +151,7 @@ public class SourcePollingChannelAdapterFactoryBean implements FactoryBean<Sourc
spca.setErrorHandler(this.pollerMetadata.getErrorHandler());
spca.setBeanClassLoader(this.beanClassLoader);
spca.setAutoStartup(this.autoStartup);
spca.setPhase(this.phase);
spca.setBeanName(this.beanName);
spca.setBeanFactory(this.beanFactory);
spca.setTransactionSynchronizationFactory(this.pollerMetadata.getTransactionSynchronizationFactory());

4
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractOutboundChannelAdapterParser.java
View File

@@ -16,6 +16,8 @@
package org.springframework.integration.config.xml;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
@@ -27,7 +29,6 @@ import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
/**
* Base class for outbound Channel Adapter parsers.
@@ -71,6 +72,7 @@ public abstract class AbstractOutboundChannelAdapterParser extends AbstractChann
}
builder.addPropertyValue("inputChannelName", channelName);
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase");
this.configureRequestHandlerAdviceChain(element, parserContext, handlerBeanComponentDefinition.getBeanDefinition(), builder);

3
spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractPollingInboundChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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.
@@ -50,6 +50,7 @@ public abstract class AbstractPollingInboundChannelAdapterParser extends Abstrac
IntegrationNamespaceUtils.configurePollerMetadata(pollerElement, adapterBuilder, parserContext);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(adapterBuilder, element, "auto-startup");
IntegrationNamespaceUtils.setValueIfAttributeDefined(adapterBuilder, element, "phase");
return adapterBuilder.getBeanDefinition();
}

123
spring-integration-core/src/main/java/org/springframework/integration/config/xml/DefaultInboundChannelAdapterParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2013 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.
@@ -18,20 +18,20 @@ package org.springframework.integration.config.xml;
import java.util.List;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.config.ExpressionFactoryBean;
import org.springframework.integration.endpoint.ExpressionEvaluatingMessageSource;
import org.springframework.integration.endpoint.MethodInvokingMessageSource;
import org.w3c.dom.Element;
import org.springframework.beans.BeanMetadataElement;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.ManagedMap;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.ExpressionFactoryBean;
import org.springframework.integration.endpoint.ExpressionEvaluatingMessageSource;
import org.springframework.integration.endpoint.MethodInvokingMessageSource;
import org.springframework.integration.expression.DynamicExpression;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
@@ -46,23 +46,34 @@ public class DefaultInboundChannelAdapterParser extends AbstractPollingInboundCh
@Override
protected BeanMetadataElement parseSource(Element element, ParserContext parserContext) {
Object source = parserContext.extractSource(element);
BeanMetadataElement result = null;
BeanComponentDefinition innnerBeanDef = IntegrationNamespaceUtils.parseInnerHandlerDefinition(element, parserContext);
String sourceRef = element.getAttribute(IntegrationNamespaceUtils.REF_ATTRIBUTE);
String methodName = element.getAttribute(IntegrationNamespaceUtils.METHOD_ATTRIBUTE);
String expressionString = element.getAttribute(IntegrationNamespaceUtils.EXPRESSION_ATTRIBUTE);
Element scriptElement = DomUtils.getChildElementByTagName(element, "script");
Element expressionElement = DomUtils.getChildElementByTagName(element, "expression");
boolean isInnerDef = innnerBeanDef != null;
boolean isRef = StringUtils.hasText(sourceRef);
boolean isExpression = StringUtils.hasText(expressionString);
boolean hasInnerDef = innnerBeanDef != null;
boolean hasRef = StringUtils.hasText(sourceRef);
boolean hasExpression = StringUtils.hasText(expressionString);
boolean hasScriptElement = scriptElement != null;
boolean hasExpressionElement = expressionElement != null;
boolean hasMethod = StringUtils.hasText(methodName);
if (!(isInnerDef ^ (isRef ^ isExpression))) {
if (!hasInnerDef && !hasRef && !hasExpression && !hasScriptElement && !hasExpressionElement) {
parserContext.getReaderContext().error(
"Exactly one of the 'ref', 'expression' or inner bean is required.", element);
"Exactly one of the 'ref', 'expression', inner bean, <script> or <expression> is required.", element);
}
if (isInnerDef) {
if (hasInnerDef) {
if (hasRef || hasExpression) {
parserContext.getReaderContext().error(
"Neither 'ref' nor 'expression' are permitted when an inner bean (<bean/>) is configured on element " +
IntegrationNamespaceUtils.createElementDescription(element) + ".", source);
return null;
}
if (hasMethod) {
result = this.parseMethodInvokingSource(innnerBeanDef, methodName, element, parserContext);
}
@@ -70,15 +81,34 @@ public class DefaultInboundChannelAdapterParser extends AbstractPollingInboundCh
result = innnerBeanDef;
}
}
else if (isExpression) {
if (hasMethod) {
else if (hasScriptElement) {
if (hasRef || hasMethod || hasExpression) {
parserContext.getReaderContext().error(
"The 'method' attribute can't be used with 'expression' attribute.", element);
"Neither 'ref' and 'method' nor 'expression' are permitted when an inner script element is configured on element " +
IntegrationNamespaceUtils.createElementDescription(element) + ".", source);
return null;
}
String expressionBeanName = this.parseExpression(expressionString, element, parserContext);
result = new RuntimeBeanReference(expressionBeanName);
BeanDefinition scriptBeanDefinition = parserContext.getDelegate().parseCustomElement(scriptElement);
BeanDefinitionBuilder sourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(
IntegrationNamespaceUtils.BASE_PACKAGE + ".scripting.ScriptExecutingMessageSource");
sourceBuilder.addConstructorArgValue(scriptBeanDefinition);
this.parseHeaderExpressions(sourceBuilder, element, parserContext);
result = sourceBuilder.getBeanDefinition();
}
else if (isRef) {
else if (hasExpression || hasExpressionElement) {
if (hasRef || hasMethod) {
parserContext.getReaderContext().error(
"The 'ref' and 'method' attributes can't be used with 'expression' attribute or inner <expression>.", element);
return null;
}
if (hasExpression & hasExpressionElement) {
parserContext.getReaderContext().error(
"Exactly one of the 'expression' attribute or inner <expression> is required.", element);
return null;
}
result = this.parseExpression(expressionString, expressionElement, element, parserContext);
}
else if (hasRef) {
BeanMetadataElement sourceValue = new RuntimeBeanReference(sourceRef);
if (hasMethod) {
result = this.parseMethodInvokingSource(sourceValue, methodName, element, parserContext);
@@ -87,31 +117,44 @@ public class DefaultInboundChannelAdapterParser extends AbstractPollingInboundCh
result = sourceValue;
}
}
else {
parserContext.getReaderContext().error("One of the following is required: " +
"'ref' attribute, 'expression' attribute, or an inner-bean definition.", element);
}
return result;
}
private BeanMetadataElement parseMethodInvokingSource(BeanMetadataElement targetObject, String methodName, Element element, ParserContext parserContext) {
private BeanMetadataElement parseMethodInvokingSource(BeanMetadataElement targetObject, String methodName, Element element,
ParserContext parserContext) {
BeanDefinitionBuilder sourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(MethodInvokingMessageSource.class);
sourceBuilder.addPropertyValue("object", targetObject);
sourceBuilder.addPropertyValue("methodName", methodName);
this.parseHeaderExpressions(sourceBuilder, element, parserContext);
String sourceRef = BeanDefinitionReaderUtils.registerWithGeneratedName(
sourceBuilder.getBeanDefinition(), parserContext.getRegistry());
return new RuntimeBeanReference(sourceRef);
return sourceBuilder.getBeanDefinition();
}
private String parseExpression(String expressionString, Element element, ParserContext parserContext) {
private BeanMetadataElement parseExpression(String expressionString, Element expressionElement, Element element,
ParserContext parserContext) {
BeanDefinitionBuilder sourceBuilder = BeanDefinitionBuilder.genericBeanDefinition(ExpressionEvaluatingMessageSource.class);
RootBeanDefinition expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class);
expressionDef.getConstructorArgumentValues().addGenericArgumentValue(expressionString);
BeanDefinition expressionDef = null;
if (StringUtils.hasText(expressionString)) {
expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class);
expressionDef.getConstructorArgumentValues().addGenericArgumentValue(expressionString);
}
else {
BeanDefinitionBuilder dynamicExpressionBuilder = BeanDefinitionBuilder.genericBeanDefinition(
DynamicExpression.class);
String key = expressionElement.getAttribute("key");
String expressionSourceReference = expressionElement.getAttribute("source");
dynamicExpressionBuilder.addConstructorArgValue(key);
dynamicExpressionBuilder.addConstructorArgReference(expressionSourceReference);
expressionDef = dynamicExpressionBuilder.getBeanDefinition();
}
sourceBuilder.addConstructorArgValue(expressionDef);
sourceBuilder.addConstructorArgValue(null);
this.parseHeaderExpressions(sourceBuilder, element, parserContext);
return BeanDefinitionReaderUtils.registerWithGeneratedName(sourceBuilder.getBeanDefinition(), parserContext.getRegistry());
return sourceBuilder.getBeanDefinition();
}
private void parseHeaderExpressions(BeanDefinitionBuilder builder, Element element, ParserContext parserContext) {
@@ -120,24 +163,8 @@ public class DefaultInboundChannelAdapterParser extends AbstractPollingInboundCh
ManagedMap<String, Object> headerExpressions = new ManagedMap<String, Object>();
for (Element headerElement : headerElements) {
String headerName = headerElement.getAttribute("name");
String headerValue = headerElement.getAttribute("value");
String headerExpression = headerElement.getAttribute("expression");
boolean hasValue = StringUtils.hasText(headerValue);
boolean hasExpression = StringUtils.hasText(headerExpression);
if (!(hasValue ^ hasExpression)) {
parserContext.getReaderContext().error("exactly one of 'value' or 'expression' is required on a header sub-element",
parserContext.extractSource(headerElement));
continue;
}
RootBeanDefinition expressionDef = null;
if (hasValue) {
expressionDef = new RootBeanDefinition(LiteralExpression.class);
expressionDef.getConstructorArgumentValues().addGenericArgumentValue(headerValue);
}
else {
expressionDef = new RootBeanDefinition(ExpressionFactoryBean.class);
expressionDef.getConstructorArgumentValues().addGenericArgumentValue(headerExpression);
}
BeanDefinition expressionDef = IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("value",
"expression", parserContext, headerElement, true);
headerExpressions.put(headerName, expressionDef);
}
builder.addPropertyValue("headerExpressions", headerExpressions);

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

@@ -800,6 +800,9 @@
<xsd:documentation>
Defines a Channel Adapter that receives from a MessageSource and sends to a
MessageChannel.
Note, when using the 'expression' attribute, or 'expression' or 'script' sub-element,
there is not yet a root Message object and therefore
the 'payload' and 'headers' properties are not available in the expression or script.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
@@ -807,10 +810,16 @@
<xsd:choice minOccurs="0" maxOccurs="1" >
<xsd:sequence>
<xsd:element name="poller" type="basePollerType" />
<xsd:element ref="beans:bean" minOccurs="0" maxOccurs="1" />
<xsd:choice minOccurs="0" maxOccurs="1" >
<xsd:element name="expression" type="innerExpressionType"/>
<xsd:any namespace="##other"/>
</xsd:choice>
</xsd:sequence>
<xsd:sequence>
<xsd:element ref="beans:bean" />
<xsd:choice minOccurs="0" maxOccurs="1" >
<xsd:element name="expression" type="innerExpressionType"/>
<xsd:any namespace="##other"/>
</xsd:choice>
<xsd:element name="poller" type="basePollerType" minOccurs="0" maxOccurs="1" />
</xsd:sequence>
</xsd:choice>
@@ -1039,6 +1048,16 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<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:attributeGroup>
<xsd:complexType name="methodInvokingChannelAdapterType">

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

@@ -15,21 +15,22 @@
<queue capacity="1"/>
</channel>
<outbound-channel-adapter id="outboundWithImplicitChannel" ref="consumer"/>
<outbound-channel-adapter id="outboundWithImplicitChannel" ref="consumer" auto-startup="false" phase="-1"/>
<outbound-channel-adapter id="methodInvokingConsumer" ref="testBean" method="store"/>
<outbound-channel-adapter id="expressionConsumer" expression="@testBean.store(payload)"/>
<inbound-channel-adapter id="methodInvokingSource" ref="testBean" method="getMessage"
channel="queueChannel" auto-startup="false">
<inbound-channel-adapter id="methodInvokingSource" ref="testBean" method="getMessage"
channel="queueChannel" auto-startup="false" phase="-1">
<poller max-messages-per-poll="1" fixed-delay="10000"/>
</inbound-channel-adapter>
<channel id="withTimeoutChannel">
<queue/>
</channel>
<inbound-channel-adapter id="methodInvokingSourceWithTimeout" ref="testBean" method="getMessage"
<inbound-channel-adapter id="methodInvokingSourceWithTimeout" ref="testBean" method="getMessage"
channel="withTimeoutChannel" auto-startup="false"
send-timeout="999">
<poller max-messages-per-poll="1" fixed-rate="800"/>

21
spring-integration-core/src/test/java/org/springframework/integration/config/ChannelAdapterParserTests.java
View File

@@ -17,10 +17,14 @@
package org.springframework.integration.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import org.hamcrest.Matchers;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
@@ -30,6 +34,8 @@ import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.endpoint.EventDrivenConsumer;
@@ -73,6 +79,7 @@ public class ChannelAdapterParserTests {
Object adapter = this.applicationContext.getBean(beanName);
assertNotNull(adapter);
assertTrue(adapter instanceof SourcePollingChannelAdapter);
assertEquals(-1, ((SourcePollingChannelAdapter) adapter).getPhase());
this.applicationContext.start();
Message<?> message = channel.receive(1000);
assertNotNull(message);
@@ -110,10 +117,22 @@ public class ChannelAdapterParserTests {
Object adapter = this.applicationContext.getBean(beanName + ".adapter");
assertNotNull(adapter);
assertTrue(adapter instanceof EventDrivenConsumer);
assertFalse(((EventDrivenConsumer) adapter).isAutoStartup());
assertEquals(-1, ((EventDrivenConsumer) adapter).getPhase());
TestConsumer consumer = (TestConsumer) this.applicationContext.getBean("consumer");
assertNull(consumer.getLastMessage());
Message<?> message = new GenericMessage<String>("test");
assertTrue(((MessageChannel) channel).send(message));
try {
((MessageChannel) channel).send(message);
fail("MessageDispatchingException is expected.");
}
catch (Exception e) {
assertThat(e, Matchers.instanceOf(MessageDeliveryException.class));
assertThat(e.getCause(), Matchers.instanceOf(MessageDispatchingException.class));
}
((EventDrivenConsumer) adapter).start();
((MessageChannel) channel).send(message);
assertNotNull(consumer.getLastMessage());
assertEquals(message, consumer.getLastMessage());
}

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

@@ -41,6 +41,15 @@
<poller trigger="customTrigger"/>
</inbound-channel-adapter>
<inbound-channel-adapter id="expressionElement" channel="triggerRefChannel">
<poller fixed-delay="1000"/>
<expression key="test.greeting"/>
</inbound-channel-adapter>
<beans:bean id="expressionSource" class="org.springframework.integration.expression.ReloadableResourceBundleExpressionSource">
<beans:property name="basename" value="org/springframework/integration/expression/expressions"/>
</beans:bean>
<beans:bean id="customTrigger" class="org.springframework.scheduling.support.PeriodicTrigger">
<beans:constructor-arg value="9999"/>
</beans:bean>

13
spring-integration-core/src/test/java/org/springframework/integration/config/xml/InboundChannelAdapterExpressionTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2013 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.
@@ -38,6 +38,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Mark Fisher
* @author Artem Bilan
* @since 2.0
*/
@ContextConfiguration
@@ -109,7 +110,7 @@ public class InboundChannelAdapterExpressionTests {
public void headerExpressions() {
SourcePollingChannelAdapter adapter = context.getBean("headerExpressionsProducer", SourcePollingChannelAdapter.class);
assertFalse(adapter.isAutoStartup());
Map<String, Expression> headerExpressions = TestUtils.getPropertyValue(adapter, "source.headerExpressions", Map.class);
Map<String, Expression> headerExpressions = TestUtils.getPropertyValue(adapter, "source.headerExpressions", Map.class);
assertEquals(2, headerExpressions.size());
assertEquals("6 * 7", headerExpressions.get("foo").getExpressionString());
assertEquals("x", headerExpressions.get("bar").getExpressionString());
@@ -117,4 +118,12 @@ public class InboundChannelAdapterExpressionTests {
assertEquals("x", headerExpressions.get("bar").getValue());
}
@Test
public void testInt2867InnerExpression() {
SourcePollingChannelAdapter adapter = context.getBean("expressionElement", SourcePollingChannelAdapter.class);
Expression expression = TestUtils.getPropertyValue(adapter, "source.expression", Expression.class);
assertEquals("'Hello World!'", expression.getExpressionString());
}
}