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

Added 'autoStartup' to message-bus (INT-101), 'initialDelay' and 'fixedRate' attributes for @Polled (INT-100), <handler-chain> element (INT-58), <selector> sub-element (INT-102), and <interceptor> sub-element (INT-106).

Browse Source
This commit is contained in:
Mark Fisher
2008-02-13 08:08:26 +00:00
parent 405311d9b9
commit ecce093f6e
32 changed files with 696 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-core/src/main/java/org/springframework/integration/annotation/Polled.java
View File

@@ -23,6 +23,8 @@ import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.integration.scheduling.PollingSchedule;
/**
* Indicates that a method is capable of providing messages. The method must not
* accept any parameters but can return either a single object or collection.
@@ -39,4 +41,8 @@ public @interface Polled {
int period() default 1000;
long initialDelay() default PollingSchedule.DEFAULT_INITIAL_DELAY;
boolean fixedRate() default PollingSchedule.DEFAULT_FIXED_RATE;
}

21
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -73,6 +73,8 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
private boolean autoCreateChannels;
private volatile boolean autoStartup = true;
private volatile boolean initialized;
private volatile boolean starting;
@@ -84,9 +86,16 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
Assert.notNull(applicationContext, "'applicationContext' must not be null");
if (applicationContext.getBeanNamesForType(this.getClass()).length > 1) {
throw new MessagingConfigurationException("Only one instance of '" + this.getClass().getSimpleName()
+ "' is allowed per ApplicationContext.");
}
this.registerChannels(applicationContext);
this.registerEndpoints(applicationContext);
this.registerSourceAdapters(applicationContext);
if (this.autoStartup) {
this.start();
}
}
public void setMessagingTaskScheduler(MessagingTaskScheduler taskScheduler) {
@@ -108,6 +117,15 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
}
}
/**
* Set whether to automatically start the bus after initialization.
* <p>Default is 'true'; set this to 'false' to allow for manual startup
* through the {@link #start()} method.
*/
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
/**
* Set whether the bus should automatically create a channel when a
* subscription contains the name of a previously unregistered channel.
@@ -226,6 +244,9 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
((ChannelRegistryAware) endpoint).setChannelRegistry(this.channelRegistry);
}
this.endpoints.put(name, endpoint);
if (this.isRunning()) {
activateEndpoint(endpoint);
}
if (logger.isInfoEnabled()) {
logger.info("registered endpoint '" + name + "'");
}

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

@@ -52,6 +52,8 @@ public class ChannelParser implements BeanDefinitionParser {
private static final String DATATYPE_ATTRIBUTE = "datatype";
private static final String INTERCEPTOR_ELEMENT = "interceptor";
private static final String INTERCEPTORS_PROPERTY = "interceptors";
@@ -60,6 +62,7 @@ public class ChannelParser implements BeanDefinitionParser {
channelDef.setSource(parserContext.extractSource(element));
boolean isPublishSubscribe = "true".equals(element.getAttribute(PUBLISH_SUBSCRIBE_ATTRIBUTE));
DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(isPublishSubscribe);
ManagedList interceptors = new ManagedList();
NodeList childNodes = element.getChildNodes();
for (int i = 0; i < childNodes.getLength(); i++) {
Node child = childNodes.item(i);
@@ -68,13 +71,16 @@ public class ChannelParser implements BeanDefinitionParser {
if (DISPATCHER_POLICY_ELEMENT.equals(localName)) {
configureDispatcherPolicy((Element) child, dispatcherPolicy);
}
else if (INTERCEPTOR_ELEMENT.equals(localName)) {
String ref = ((Element) child).getAttribute("ref");
interceptors.add(new RuntimeBeanReference(ref));
}
}
}
String capAttr = element.getAttribute(CAPACITY_ATTRIBUTE);
int capacity = (StringUtils.hasText(capAttr)) ? Integer.parseInt(capAttr) : SimpleChannel.DEFAULT_CAPACITY;
channelDef.getConstructorArgumentValues().addIndexedArgumentValue(0, capacity);
channelDef.getConstructorArgumentValues().addIndexedArgumentValue(1, dispatcherPolicy);
ManagedList interceptors = new ManagedList();
String datatypeAttr = element.getAttribute(DATATYPE_ATTRIBUTE);
if (StringUtils.hasText(datatypeAttr)) {
String[] datatypes = StringUtils.commaDelimitedListToStringArray(datatypeAttr);
@@ -90,7 +96,6 @@ public class ChannelParser implements BeanDefinitionParser {
parserContext.registerBeanComponent(interceptorComponent);
interceptors.add(new RuntimeBeanReference(interceptorBeanName));
}
// TODO: parse interceptor sub-elements
channelDef.getPropertyValues().addPropertyValue(INTERCEPTORS_PROPERTY, interceptors);
String beanName = element.getAttribute(ID_ATTRIBUTE);
parserContext.registerBeanComponent(new BeanComponentDefinition(channelDef, beanName));

19
spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java
View File

@@ -58,6 +58,10 @@ public class EndpointParser implements BeanDefinitionParser {
private static final String DEFAULT_OUTPUT_CHANNEL_PROPERTY = "defaultOutputChannelName";
private static final String SELECTOR_ELEMENT = "selector";
private static final String SELECTORS_PROPERTY = "messageSelectors";
private static final String HANDLER_ELEMENT = "handler";
private static final String REF_ATTRIBUTE = "ref";
@@ -107,6 +111,7 @@ public class EndpointParser implements BeanDefinitionParser {
if (StringUtils.hasText(defaultOutputChannel)) {
endpointDef.getPropertyValues().addPropertyValue(DEFAULT_OUTPUT_CHANNEL_PROPERTY, defaultOutputChannel);
}
ManagedList selectors = new ManagedList();
List<String> childHandlerRefs = new ArrayList<String>();
NodeList childNodes = element.getChildNodes();
for (int i = 0; i < childNodes.getLength(); i++) {
@@ -116,10 +121,19 @@ public class EndpointParser implements BeanDefinitionParser {
if (CONCURRENCY_ELEMENT.equals(localName)) {
parseConcurrencyPolicy((Element) child, endpointDef);
}
else if (SELECTOR_ELEMENT.equals(localName)) {
String ref = ((Element) child).getAttribute(REF_ATTRIBUTE);
selectors.add(new RuntimeBeanReference(ref));
}
else if (HANDLER_ELEMENT.equals(localName)) {
String ref = ((Element) child).getAttribute(REF_ATTRIBUTE);
String method = ((Element) child).getAttribute(METHOD_ATTRIBUTE);
childHandlerRefs.add(this.parseHandlerAdapter(ref, method, parserContext));
if (StringUtils.hasText(method)) {
childHandlerRefs.add(this.parseHandlerAdapter(ref, method, parserContext));
}
else {
childHandlerRefs.add(ref);
}
}
else if (SCHEDULE_ELEMENT.equals(localName)) {
this.parseSchedule((Element) child, subscriptionDef);
@@ -129,6 +143,9 @@ public class EndpointParser implements BeanDefinitionParser {
String subscriptionBeanName = parserContext.getReaderContext().generateBeanName(subscriptionDef);
parserContext.registerBeanComponent(new BeanComponentDefinition(subscriptionDef, subscriptionBeanName));
endpointDef.getPropertyValues().addPropertyValue(SUBSCRIPTION_PROPERTY, new RuntimeBeanReference(subscriptionBeanName));
if (selectors.size() > 0) {
endpointDef.getPropertyValues().addPropertyValue(SELECTORS_PROPERTY, selectors);
}
if (childHandlerRefs.size() > 0) {
if (childHandlerRefs.size() == 1) {
endpointDef.getPropertyValues().addPropertyValue(

127
spring-integration-core/src/main/java/org/springframework/integration/config/HandlerParser.java Normal file
View File

@@ -0,0 +1,127 @@
/*
* Copyright 2002-2007 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
*
* 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.
*/
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.BeanDefinition;
import org.springframework.beans.factory.config.BeanDefinitionHolder;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.beans.factory.xml.BeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.handler.DefaultMessageHandlerAdapter;
import org.springframework.integration.handler.MessageHandlerChain;
import org.springframework.util.StringUtils;
/**
* Parser for the &lt;handler/&gt; element.
*
* @author Mark Fisher
*/
public class HandlerParser implements BeanDefinitionParser {
private static final String HANDLER_CHAIN_ELEMENT = "handler-chain";
private static final String HANDLER_ELEMENT = "handler";
private static final String HANDLERS_PROPERTY = "handlers";
private static final String OBJECT_PROPERTY = "object";
private static final String METHOD_NAME_PROPERTY = "methodName";
public BeanDefinition parse(Element element, ParserContext parserContext) {
if (HANDLER_CHAIN_ELEMENT.equals(element.getLocalName())) {
return this.parseHandlerChain(element, parserContext);
}
else if (HANDLER_ELEMENT.equals(element.getLocalName())) {
return this.parseHandler(element, parserContext, null);
}
return null;
}
private BeanDefinition parseHandlerChain(Element element, ParserContext parserContext) {
RootBeanDefinition beanDefinition = new RootBeanDefinition(MessageHandlerChain.class);
ManagedList handlers = new ManagedList();
NodeList childNodes = element.getChildNodes();
for (int i = 0; i < childNodes.getLength(); i++) {
Node child = childNodes.item(i);
if (child.getNodeType() == Node.ELEMENT_NODE) {
String localName = child.getLocalName();
if (HANDLER_ELEMENT.equals(localName)) {
parseHandler((Element) child, parserContext, handlers);
}
}
}
beanDefinition.getPropertyValues().addPropertyValue(HANDLERS_PROPERTY, handlers);
String id = element.getAttribute("id");
String beanName = (StringUtils.hasText(id)) ? id : parserContext.getReaderContext().generateBeanName(beanDefinition);
parserContext.registerBeanComponent(new BeanComponentDefinition(beanDefinition, beanName));
return beanDefinition;
}
private BeanDefinition parseHandler(Element element, ParserContext parserContext, ManagedList handlers) {
boolean isInnerHandler = (handlers != null);
String ref = element.getAttribute("ref");
String method = element.getAttribute("method");
String id = element.getAttribute("id");
if (!isInnerHandler && (!StringUtils.hasText(id) || !StringUtils.hasText(ref) || !StringUtils.hasText(method))) {
parserContext.getReaderContext().error("Top-level <handler> elements must provide 'id', 'ref', and 'method' attributes.",
parserContext.extractSource(element));
}
if (isInnerHandler && StringUtils.hasText(id)) {
parserContext.getReaderContext().error("The 'id' attribute is only supported for top-level <handler> elements.",
parserContext.extractSource(element));
}
if (StringUtils.hasText(method)) {
BeanDefinitionHolder bdh = this.parseHandlerAdapter(id, ref, method, parserContext, isInnerHandler);
if (handlers != null) {
handlers.add(bdh.getBeanDefinition());
return null;
}
return bdh.getBeanDefinition();
}
if (StringUtils.hasText(id)) {
parserContext.getReaderContext().error("The 'id' attribute is only supported for handler adapters (when 'method' is also provided).",
parserContext.extractSource(element));
}
if (handlers != null) {
handlers.add(new RuntimeBeanReference(ref));
}
return null;
}
private BeanDefinitionHolder parseHandlerAdapter(String id, String handlerRef, String handlerMethod, ParserContext parserContext, boolean isInnerHandler) {
BeanDefinition handlerAdapterDef = new RootBeanDefinition(DefaultMessageHandlerAdapter.class);
handlerAdapterDef.getPropertyValues().addPropertyValue(OBJECT_PROPERTY, new RuntimeBeanReference(handlerRef));
handlerAdapterDef.getPropertyValues().addPropertyValue(METHOD_NAME_PROPERTY, handlerMethod);
String adapterBeanName = (StringUtils.hasText(id)) ? id :
BeanDefinitionReaderUtils.generateBeanName(handlerAdapterDef, parserContext.getRegistry(), isInnerHandler);
if (!isInnerHandler) {
parserContext.registerBeanComponent(new BeanComponentDefinition(handlerAdapterDef, adapterBeanName));
}
return new BeanDefinitionHolder(handlerAdapterDef, adapterBeanName);
}
}

2
spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java
View File

@@ -36,6 +36,8 @@ public class IntegrationNamespaceHandler extends NamespaceHandlerSupport {
registerBeanDefinitionParser("source-adapter", new ChannelAdapterParser(true));
registerBeanDefinitionParser("target-adapter", new ChannelAdapterParser(false));
registerBeanDefinitionParser("endpoint", new EndpointParser());
registerBeanDefinitionParser("handler", new HandlerParser());
registerBeanDefinitionParser("handler-chain", new HandlerParser());
registerBeanDefinitionParser("file-source", new FileSourceAdapterParser());
registerBeanDefinitionParser("file-target", new FileTargetAdapterParser());
registerBeanDefinitionParser("jms-source", new JmsSourceAdapterParser());

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

@@ -24,6 +24,7 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.core.Conventions;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.bus.MessageBus;
import org.springframework.util.StringUtils;
@@ -36,18 +37,24 @@ public class MessageBusParser extends AbstractSimpleBeanDefinitionParser {
public static final String MESSAGE_BUS_BEAN_NAME = "internal.MessageBus";
private static final Class<?> MESSAGE_BUS_CLASS = MessageBus.class;
private static final String ERROR_CHANNEL_ATTRIBUTE = "error-channel";
@Override
protected String resolveId(Element element, AbstractBeanDefinition definition, ParserContext parserContext)
throws BeanDefinitionStoreException {
if (parserContext.getRegistry().containsBeanDefinition(MESSAGE_BUS_BEAN_NAME)) {
throw new MessagingConfigurationException("Only one instance of '" + MESSAGE_BUS_CLASS.getSimpleName()
+ "' is allowed per ApplicationContext.");
}
return MESSAGE_BUS_BEAN_NAME;
}
@Override
protected Class<?> getBeanClass(Element element) {
return MessageBus.class;
return MESSAGE_BUS_CLASS;
}
@Override

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

@@ -134,7 +134,10 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
Annotation annotation = AnnotationUtils.getAnnotation(method, Polled.class);
if (annotation != null) {
int period = ((Polled) annotation).period();
Polled polledAnnotation = (Polled) annotation;
int period = polledAnnotation.period();
long initialDelay = polledAnnotation.initialDelay();
boolean fixedRate = polledAnnotation.fixedRate();
MethodInvokingSource<Object> source = new MethodInvokingSource<Object>();
source.setObject(bean);
source.setMethod(method.getName());
@@ -146,7 +149,9 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor
messageBus.registerChannel(channelName, channel);
messageBus.registerSourceAdapter(beanName + "-sourceAdapter", adapter);
Subscription subscription = new Subscription(channel);
Schedule schedule = new PollingSchedule(period);
PollingSchedule schedule = new PollingSchedule(period);
schedule.setInitialDelay(initialDelay);
schedule.setFixedRate(fixedRate);
subscription.setSchedule(schedule);
endpoint.setSubscription(subscription);
}

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

@@ -24,6 +24,7 @@
Defines a message bus.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="auto-startup" type="xsd:boolean"/>
<xsd:attribute name="auto-create-channels" type="xsd:boolean"/>
<xsd:attribute name="error-channel" type="xsd:string"/>
<xsd:attribute name="dispatcher-pool-size" type="xsd:int"/>
@@ -49,6 +50,7 @@
</xsd:annotation>
<xsd:sequence>
<xsd:element ref="dispatcher-policy" minOccurs="0" maxOccurs="1"/>
<xsd:element ref="interceptor" minOccurs="0" maxOccurs="unbounded"/>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:ID" use="required"/>
<xsd:attribute name="capacity" type="xsd:integer"/>
@@ -57,6 +59,17 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="interceptor">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Provides a channel interceptor reference.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="ref" type="xsd:string" use="required"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="source-adapter">
<xsd:complexType>
<xsd:annotation>
@@ -104,6 +117,7 @@
<xsd:sequence>
<xsd:element ref="schedule" minOccurs="0" maxOccurs="1"/>
<xsd:element ref="concurrency" minOccurs="0" maxOccurs="1"/>
<xsd:element ref="selector" minOccurs="0" maxOccurs="unbounded"/>
<xsd:element ref="handler" minOccurs="0" maxOccurs="unbounded"/>
</xsd:sequence>
<xsd:attribute name="input-channel" type="xsd:string"/>
@@ -140,6 +154,31 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="selector">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Provides a message selector reference.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="ref" type="xsd:string" use="required"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="handler-chain">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a MessageHandler chain.
</xsd:documentation>
</xsd:annotation>
<xsd:sequence>
<xsd:element ref="handler" minOccurs="1" maxOccurs="unbounded"/>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:ID"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="handler">
<xsd:complexType>
<xsd:annotation>
@@ -147,6 +186,7 @@
Defines a handler.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="id" type="xsd:ID"/>
<xsd:attribute name="ref" type="xsd:string" use="required"/>
<xsd:attribute name="method" type="xsd:string"/>
</xsd:complexType>

54
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java
View File

@@ -18,8 +18,8 @@ package org.springframework.integration.dispatcher;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicLong;
@@ -59,10 +59,12 @@ public class DefaultMessageDispatcher implements SchedulingMessageDispatcher, Me
private Schedule defaultSchedule = new PollingSchedule(5);
private Map<Schedule, List<MessageHandler>> scheduledHandlers = new ConcurrentHashMap<Schedule, List<MessageHandler>>();
private ConcurrentMap<Schedule, List<MessageHandler>> scheduledHandlers = new ConcurrentHashMap<Schedule, List<MessageHandler>>();
private AtomicLong totalMessagesProcessed = new AtomicLong();
private volatile boolean starting;
private volatile boolean running;
private Object lifecycleMonitor = new Object();
@@ -105,13 +107,13 @@ public class DefaultMessageDispatcher implements SchedulingMessageDispatcher, Me
if (this.isRunning() && handler instanceof Lifecycle) {
((Lifecycle) handler).start();
}
if (this.scheduledHandlers.containsKey(schedule)) {
this.scheduledHandlers.get(schedule).add(handler);
List<MessageHandler> handlers = this.scheduledHandlers.get(schedule);
if (handlers == null) {
handlers = this.scheduledHandlers.putIfAbsent(schedule, new CopyOnWriteArrayList<MessageHandler>());
}
else {
List<MessageHandler> handlerList = new CopyOnWriteArrayList<MessageHandler>();
handlerList.add(handler);
this.scheduledHandlers.put(schedule, handlerList);
this.scheduledHandlers.get(schedule).add(handler);
if (handlers == null && this.isRunning()) {
this.scheduleDispatcherTask(schedule);
}
}
@@ -120,6 +122,12 @@ public class DefaultMessageDispatcher implements SchedulingMessageDispatcher, Me
}
public void start() {
synchronized (this.lifecycleMonitor) {
if (this.isRunning() || this.starting) {
return;
}
this.starting = true;
}
if (this.scheduler == null) {
if (logger.isInfoEnabled()) {
logger.info("no scheduler was provided, will create one");
@@ -129,23 +137,28 @@ public class DefaultMessageDispatcher implements SchedulingMessageDispatcher, Me
if (!this.scheduler.isRunning()) {
this.scheduler.start();
}
if (this.isRunning()) {
return;
}
synchronized (this.lifecycleMonitor) {
for (Schedule schedule : this.scheduledHandlers.keySet()) {
List<MessageHandler> handlers = this.scheduledHandlers.get(schedule);
for (MessageHandler handler : handlers) {
if (handler instanceof Lifecycle) {
((Lifecycle) handler).start();
}
}
this.scheduler.schedule(new DispatcherTask(schedule));
scheduleDispatcherTask(schedule);
}
this.running = true;
this.starting = false;
}
}
private void scheduleDispatcherTask(Schedule schedule) {
if (!this.isRunning()) {
this.start();
}
List<MessageHandler> handlers = this.scheduledHandlers.get(schedule);
for (MessageHandler handler : handlers) {
if (handler instanceof Lifecycle) {
((Lifecycle) handler).start();
}
}
this.scheduler.schedule(new DispatcherTask(schedule));
}
public void stop() {
if (!this.isRunning()) {
return;
@@ -198,12 +211,9 @@ public class DefaultMessageDispatcher implements SchedulingMessageDispatcher, Me
private Schedule schedule;
private MessageDistributor distributor;
public DispatcherTask(Schedule schedule) {
this.schedule = (schedule != null) ? schedule : defaultSchedule;
this.distributor = getDistributor(this.schedule);
}
public Schedule getSchedule() {
@@ -211,7 +221,7 @@ public class DefaultMessageDispatcher implements SchedulingMessageDispatcher, Me
}
public void run() {
doDispatch(this.distributor);
doDispatch(getDistributor(this.schedule));
}
}

6
spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java
View File

@@ -102,6 +102,10 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
this.handler = handler;
}
public void setMessageSelectors(List<MessageSelector> selectors) {
this.selectors = new CopyOnWriteArrayList<MessageSelector>(selectors);
}
public void addMessageSelector(MessageSelector messageSelector) {
Assert.notNull(messageSelector, "'messageSelector' must not be null");
this.selectors.add(messageSelector);
@@ -210,7 +214,7 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA
return null;
}
try {
Message<?> replyMessage = handler.handle(message);
Message<?> replyMessage = this.handler.handle(message);
if (replyMessage != null) {
this.replyHandler.handle(replyMessage, message.getHeader());
}

9
spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollingSchedule.java
View File

@@ -27,13 +27,18 @@ import org.springframework.util.Assert;
*/
public class PollingSchedule implements Schedule {
public static final long DEFAULT_INITIAL_DELAY = 0;
public static final boolean DEFAULT_FIXED_RATE = false;
private long period;
private long initialDelay = 0;
private long initialDelay = DEFAULT_INITIAL_DELAY;
private TimeUnit timeUnit = TimeUnit.MILLISECONDS;
private boolean fixedRate = false;
private boolean fixedRate = DEFAULT_FIXED_RATE;
/**