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

PollingDispatchers are configured from the <poller/> element. The <schedule/> element has been removed. Instead the 'period' attribute is now available on <poller/>.

Browse Source
This commit is contained in:
Mark Fisher
2008-08-05 02:52:52 +00:00
parent 9f3e7d3d24
commit de207c9bda
13 changed files with 60 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java
View File

@@ -344,6 +344,11 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
}
if (source != null && source instanceof SubscribableSource) {
((SubscribableSource) source).subscribe(endpoint);
if (source instanceof PollingDispatcher) {
PollingDispatcher poller = (PollingDispatcher) source;
this.pollingDispatchers.add(poller);
this.taskScheduler.schedule(poller);
}
if (logger.isInfoEnabled()) {
logger.info("activated subscription to channel '"
+ source + "' for endpoint '" + endpoint + "'");

58
org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractHandlerEndpointParser.java
View File

@@ -17,8 +17,6 @@
package org.springframework.integration.config;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.config.RuntimeBeanReference;
@@ -31,9 +29,8 @@ import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
/**
* Base class parser for elements that create handler-invoking endpoints.
@@ -52,10 +49,6 @@ public abstract class AbstractHandlerEndpointParser extends AbstractSingleBeanDe
protected static final String RETURN_ADDRESS_OVERRIDES_ATTRIBUTE = "return-address-overrides";
private static final String PERIOD_ATTRIBUTE = "period";
private static final String SCHEDULE_ELEMENT = "schedule";
private static final String POLLER_ELEMENT = "poller";
private static final String SELECTOR_ATTRIBUTE = "selector";
@@ -96,32 +89,24 @@ public abstract class AbstractHandlerEndpointParser extends AbstractSingleBeanDe
else {
builder.addConstructorArgReference(ref);
}
Schedule schedule = null;
NodeList childNodes = element.getChildNodes();
for (int i = 0; i < childNodes.getLength(); i++) {
Node child = childNodes.item(i);
if (child.getNodeType() == Node.ELEMENT_NODE) {
Element childElement = (Element) child;
String localName = child.getLocalName();
if (SCHEDULE_ELEMENT.equals(localName)) {
schedule = this.parseSchedule(childElement);
}
else if (POLLER_ELEMENT.equals(localName)) {
builder.addPropertyReference("messageExchangeTemplate",
IntegrationNamespaceUtils.parsePoller(childElement, parserContext));
}
else if (INTERCEPTORS_ELEMENT.equals(localName)) {
EndpointInterceptorParser parser = new EndpointInterceptorParser();
ManagedList interceptors = parser.parseInterceptors(childElement, parserContext);
builder.addPropertyValue("interceptors", interceptors);
}
}
String inputChannel = element.getAttribute(INPUT_CHANNEL_ATTRIBUTE);
if (!StringUtils.hasText(inputChannel)) {
throw new ConfigurationException("the '" + INPUT_CHANNEL_ATTRIBUTE + "' attribute is required");
}
if (schedule != null) {
builder.addPropertyValue("schedule", schedule);
Element pollerElement = DomUtils.getChildElementByTagName(element, POLLER_ELEMENT);
if (pollerElement != null) {
String pollerBeanName = IntegrationNamespaceUtils.parsePoller(inputChannel, pollerElement, parserContext);
builder.addPropertyReference("source", pollerBeanName);
}
else {
builder.addPropertyReference("source", inputChannel);
}
Element interceptorsElement = DomUtils.getChildElementByTagName(element, INTERCEPTORS_ELEMENT);
if (interceptorsElement != null) {
EndpointInterceptorParser parser = new EndpointInterceptorParser();
ManagedList interceptors = parser.parseInterceptors(interceptorsElement, parserContext);
builder.addPropertyValue("interceptors", interceptors);
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(
builder, element, INPUT_CHANNEL_ATTRIBUTE, "inputChannelName");
IntegrationNamespaceUtils.setValueIfAttributeDefined(
builder, element, OUTPUT_CHANNEL_ATTRIBUTE, "outputChannelName");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, SELECTOR_ATTRIBUTE);
@@ -151,13 +136,4 @@ public abstract class AbstractHandlerEndpointParser extends AbstractSingleBeanDe
return adapterBeanName;
}
private Schedule parseSchedule(Element scheduleElement) {
PollingSchedule schedule = new PollingSchedule(0);
String period = scheduleElement.getAttribute(PERIOD_ATTRIBUTE);
if (StringUtils.hasText(period)) {
schedule.setPeriod(Integer.parseInt(period));
}
return schedule;
}
}

33
org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java
View File

@@ -25,8 +25,11 @@ import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.core.Conventions;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.dispatcher.SimpleDispatcher;
import org.springframework.integration.message.AsyncMessageExchangeTemplate;
import org.springframework.integration.message.MessageExchangeTemplate;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
@@ -135,26 +138,30 @@ public abstract class IntegrationNamespaceUtils {
/**
* Parse a "poller" element and return the bean name of the poller instance.
* The name will be generated for a newly created bean, or if the poller
* element simply provides a "ref", that value will be returned.
*
* @param sourceBeanName the name of the PollableSource bean
* @param element the "poller" element to parse
* @param parserContext the parserContext for registering a newly created bean definition
* @return the name of the poller bean definition
*/
public static String parsePoller(Element element, ParserContext parserContext) {
String ref = element.getAttribute("ref");
public static String parsePoller(String sourceBeanName, Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(PollingDispatcher.class);
Long period = Long.valueOf(element.getAttribute("period"));
PollingSchedule schedule = new PollingSchedule(period);
String templateBeanName = parseMessageExhangeTemplate(element, parserContext);
builder.addConstructorArgReference(sourceBeanName);
builder.addConstructorArgValue(schedule);
builder.addConstructorArgValue(new SimpleDispatcher());
builder.addConstructorArgReference(templateBeanName);
setValueIfAttributeDefined(builder, element, "receive-timeout");
setValueIfAttributeDefined(builder, element, "send-timeout");
setValueIfAttributeDefined(builder, element, "max-messages-per-poll");
return BeanDefinitionReaderUtils.registerWithGeneratedName(builder.getBeanDefinition(), parserContext.getRegistry());
}
private static String parseMessageExhangeTemplate(Element element, ParserContext parserContext) {
String taskExecutorRef = element.getAttribute("task-executor");
Element txElement = DomUtils.getChildElementByTagName(element, "transactional");
if (StringUtils.hasText(ref)) {
if (StringUtils.hasText(taskExecutorRef) || (txElement != null)) {
parserContext.getReaderContext().error(
"Neither the 'task-executor' attribute or 'transactional' sub-element "
+ "should be provided when using the 'ref' attribute.",
parserContext.extractSource(element));
}
return ref;
}
Class<?> beanClass = (StringUtils.hasText(taskExecutorRef)) ?
AsyncMessageExchangeTemplate.class : MessageExchangeTemplate.class;
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(beanClass);

17
org.springframework.integration/src/main/java/org/springframework/integration/config/spring-integration-core-1.0.xsd
View File

@@ -207,7 +207,6 @@
<xsd:complexContent>
<xsd:extension base="beans:identifiedType">
<xsd:all>
<xsd:element ref="schedule" minOccurs="0" maxOccurs="1"/>
<xsd:element ref="poller" minOccurs="0" maxOccurs="1"/>
<xsd:element name="interceptors" type="endpointInterceptorsType" minOccurs="0" maxOccurs="1"/>
</xsd:all>
@@ -219,17 +218,6 @@
</xsd:complexContent>
</xsd:complexType>
<xsd:element name="schedule">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a schedule.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="period" type="xsd:int"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="poller">
<xsd:complexType>
<xsd:annotation>
@@ -240,10 +228,11 @@
<xsd:sequence>
<xsd:element name="transactional" type="transactionalType" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="ref" type="xsd:string"/>
<xsd:attribute name="task-executor" type="xsd:string"/>
<xsd:attribute name="period" type="xsd:long" use="required"/>
<xsd:attribute name="receive-timeout" type="xsd:long"/>
<xsd:attribute name="send-timeout" type="xsd:long"/>
<xsd:attribute name="max-messages-per-poll" type="xsd:long"/>
<xsd:attribute name="task-executor" type="xsd:string"/>
</xsd:complexType>
</xsd:element>

4
org.springframework.integration/src/test/java/org/springframework/integration/config/endpointInterceptorTests.xml
View File

@@ -17,7 +17,7 @@
input-channel="testChannel"
ref="testHandler"
output-channel="replyChannel">
<schedule period="100"/>
<poller period="100"/>
<interceptors>
<beans:bean class="org.springframework.integration.config.TestPreSendInterceptor"/>
<beans:bean class="org.springframework.integration.config.TestAroundSendEndpointInterceptor"/>
@@ -28,7 +28,7 @@
input-channel="testChannel"
ref="testHandler"
output-channel="replyChannel">
<schedule period="100"/>
<poller period="100"/>
<interceptors>
<ref bean="preInterceptor"/>
<ref bean="aroundInterceptor"/>

2
org.springframework.integration/src/test/java/org/springframework/integration/config/endpointWithHandlerChainElement.xml
View File

@@ -20,7 +20,7 @@
</handler-chain>
<service-activator input-channel="testChannel" ref="chain" output-channel="replyChannel">
<schedule period="100"/>
<poller period="100"/>
</service-activator>
<beans:bean id="handler1" class="org.springframework.integration.config.TestConcatenatingHandler">

2
org.springframework.integration/src/test/java/org/springframework/integration/config/endpointWithSelector.xml
View File

@@ -13,7 +13,7 @@
<service-activator id="endpoint" input-channel="testChannel"
ref="testHandler" selector="typeSelector">
<schedule period="100"/>
<poller period="100"/>
</service-activator>
<beans:bean id="typeSelector" class="org.springframework.integration.message.selector.PayloadTypeSelector">

2
org.springframework.integration/src/test/java/org/springframework/integration/config/handlerAdapterEndpointTests.xml
View File

@@ -12,7 +12,7 @@
<queue-channel id="testChannel" capacity="50"/>
<service-activator input-channel="testChannel" ref="testBean" method="store">
<schedule period="100"/>
<poller period="100"/>
</service-activator>
<beans:bean id="testBean" class="org.springframework.integration.config.TestBean">

2
org.springframework.integration/src/test/java/org/springframework/integration/config/simpleEndpointTests.xml
View File

@@ -12,7 +12,7 @@
<queue-channel id="testChannel" capacity="50"/>
<service-activator input-channel="testChannel" ref="testHandler">
<schedule period="100"/>
<poller period="100"/>
</service-activator>
<beans:bean id="testHandler" class="org.springframework.integration.config.TestHandler">

10
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/transactionInterceptorPropagationTests.xml
View File

@@ -17,7 +17,7 @@
ref="testBean"
method="good"
output-channel="output">
<schedule period="10000"/>
<poller period="10000"/>
<interceptors>
<transaction-interceptor transaction-manager="txManager" propagation="REQUIRED"/>
</interceptors>
@@ -28,7 +28,7 @@
ref="testBean"
method="good"
output-channel="output">
<schedule period="10000"/>
<poller period="10000"/>
<interceptors>
<transaction-interceptor transaction-manager="txManager" propagation="REQUIRES_NEW"/>
</interceptors>
@@ -39,7 +39,7 @@
ref="testBean"
method="good"
output-channel="output">
<schedule period="10000"/>
<poller period="10000"/>
<interceptors>
<transaction-interceptor transaction-manager="txManager" propagation="SUPPORTS"/>
</interceptors>
@@ -50,7 +50,7 @@
ref="testBean"
method="good"
output-channel="output">
<schedule period="10000"/>
<poller period="10000"/>
<interceptors>
<transaction-interceptor transaction-manager="txManager" propagation="NOT_SUPPORTED"/>
</interceptors>
@@ -61,7 +61,7 @@
ref="testBean"
method="good"
output-channel="output">
<schedule period="10000"/>
<poller period="10000"/>
<interceptors>
<transaction-interceptor transaction-manager="txManager" propagation="MANDATORY"/>
</interceptors>

4
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/interceptor/transactionInterceptorTests.xml
View File

@@ -17,7 +17,7 @@
ref="testBean"
method="bad"
output-channel="output">
<schedule period="100"/>
<poller period="100"/>
<interceptors>
<transaction-interceptor transaction-manager="txManager"/>
</interceptors>
@@ -27,7 +27,7 @@
ref="testBean"
method="good"
output-channel="output">
<schedule period="100"/>
<poller period="100"/>
<interceptors>
<transaction-interceptor transaction-manager="txManager" propagation="REQUIRED"/>
</interceptors>