diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Polled.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Polled.java index 1917127091..c1ebbf2e2a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/Polled.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/Polled.java @@ -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; + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index d25cf3e036..a9c04fc814 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -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. + *

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 + "'"); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java index b28606eab5..3e3a43673a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java @@ -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)); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java index c37d62487f..fbeee52ab4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/EndpointParser.java @@ -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 childHandlerRefs = new ArrayList(); 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( diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/HandlerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/HandlerParser.java new file mode 100644 index 0000000000..e19bc808a3 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/HandlerParser.java @@ -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 <handler/> 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 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 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); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java index c3fe2fce70..182a59bb72 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationNamespaceHandler.java @@ -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()); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageBusParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageBusParser.java index 4e3038ba07..4643ede564 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageBusParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageBusParser.java @@ -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 diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java index 093beb9aaf..7d92c48c3f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java @@ -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 source = new MethodInvokingSource(); 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); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd index 519c85f580..d71b71a4f7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/spring-integration-1.0.xsd @@ -24,6 +24,7 @@ Defines a message bus. + @@ -49,6 +50,7 @@ + @@ -57,6 +59,17 @@ + + + + + Provides a channel interceptor reference. + + + + + + @@ -104,6 +117,7 @@ + @@ -140,6 +154,31 @@ + + + + + Provides a message selector reference. + + + + + + + + + + + Defines a MessageHandler chain. + + + + + + + + + @@ -147,6 +186,7 @@ Defines a handler. + diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java index db405335f3..472de39f54 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDispatcher.java @@ -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> scheduledHandlers = new ConcurrentHashMap>(); + private ConcurrentMap> scheduledHandlers = new ConcurrentHashMap>(); 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 handlers = this.scheduledHandlers.get(schedule); + if (handlers == null) { + handlers = this.scheduledHandlers.putIfAbsent(schedule, new CopyOnWriteArrayList()); } - else { - List handlerList = new CopyOnWriteArrayList(); - 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 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 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)); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java index 4df62c820b..f8dd682bed 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/DefaultMessageEndpoint.java @@ -102,6 +102,10 @@ public class DefaultMessageEndpoint implements MessageEndpoint, ChannelRegistryA this.handler = handler; } + public void setMessageSelectors(List selectors) { + this.selectors = new CopyOnWriteArrayList(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()); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollingSchedule.java b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollingSchedule.java index 1263213e0e..fdd8d160db 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollingSchedule.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scheduling/PollingSchedule.java @@ -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; /** diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml index 52dffeec47..d7ce264173 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml @@ -4,7 +4,9 @@ xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd"> - + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTestsWithNamespace.xml b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTestsWithNamespace.xml index 52322d548d..be2881ff27 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTestsWithNamespace.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTestsWithNamespace.xml @@ -7,7 +7,7 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java index e38278ba31..17bb6158dc 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -27,7 +27,9 @@ import java.util.concurrent.TimeUnit; import org.junit.Test; +import org.springframework.beans.factory.BeanCreationException; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.MessagingConfigurationException; import org.springframework.integration.adapter.PollableSource; import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.adapter.SourceAdapter; @@ -175,6 +177,19 @@ public class MessageBusTests { bus.stop(); } + @Test + public void testMultipleMessageBusBeans() { + boolean exceptionThrown = false; + try { + new ClassPathXmlApplicationContext("multipleMessageBusBeans.xml", this.getClass()); + } + catch (BeanCreationException e) { + exceptionThrown = true; + assertEquals(MessagingConfigurationException.class, e.getCause().getClass()); + } + assertTrue(exceptionThrown); + } + private static class FailingSource implements PollableSource { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/multipleMessageBusBeans.xml b/spring-integration-core/src/test/java/org/springframework/integration/bus/multipleMessageBusBeans.xml new file mode 100644 index 0000000000..cf612d688b --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/multipleMessageBusBeans.xml @@ -0,0 +1,11 @@ + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java index ed92d31b45..c5fa2929d7 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ChannelParserTests.java @@ -27,6 +27,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; import org.springframework.beans.FatalBeanException; +import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.channel.MessageChannel; @@ -186,6 +187,20 @@ public class ChannelParserTests { channel.send(new GenericMessage(true)); } + @Test + public void testChannelInteceptors() { + ApplicationContext context = new ClassPathXmlApplicationContext( + "channelInterceptorParserTests.xml", this.getClass()); + MessageChannel channel = (MessageChannel) context.getBean("channel"); + TestChannelInterceptor interceptor = (TestChannelInterceptor) context.getBean("interceptor"); + assertEquals(0, interceptor.getSendCount()); + channel.send(new StringMessage("test")); + assertEquals(1, interceptor.getSendCount()); + assertEquals(0, interceptor.getReceiveCount()); + channel.receive(); + assertEquals(1, interceptor.getReceiveCount()); + } + private static class TestHandler implements MessageHandler { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java index 77786271df..47d7a8781f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java @@ -17,17 +17,24 @@ package org.springframework.integration.config; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import java.util.concurrent.TimeUnit; import org.junit.Test; +import org.springframework.context.Lifecycle; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.DefaultMessageEndpoint; +import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; +import org.springframework.integration.message.selector.MessageSelectorRejectedException; /** * @author Mark Fisher @@ -47,6 +54,19 @@ public class EndpointParserTests { assertEquals("test", handler.getMessageString()); } + @Test + public void testEndpointWithChildHandler() throws InterruptedException { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "endpointWithHandlerChildElement.xml", this.getClass()); + context.start(); + MessageChannel channel = (MessageChannel) context.getBean("testChannel"); + TestHandler handler = (TestHandler) context.getBean("testHandler"); + assertNull(handler.getMessageString()); + channel.send(new GenericMessage(1, "test")); + handler.getLatch().await(50, TimeUnit.MILLISECONDS); + assertEquals("test", handler.getMessageString()); + } + @Test public void testHandlerAdapterEndpoint() throws InterruptedException { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( @@ -84,4 +104,28 @@ public class EndpointParserTests { assertEquals(7777, concurrencyPolicy.getKeepAliveSeconds()); } + @Test + public void testEndpointWithSelectorAccepts() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "endpointWithSelectors.xml", this.getClass()); + MessageHandler endpoint = (MessageHandler) context.getBean("endpoint"); + ((Lifecycle) endpoint).start(); + Message message = new StringMessage("test"); + MessageChannel replyChannel = new SimpleChannel(); + message.getHeader().setReplyChannel(replyChannel); + endpoint.handle(message); + Message reply = replyChannel.receive(500); + assertNotNull(reply); + assertEquals("foo", reply.getPayload()); + } + + @Test(expected=MessageSelectorRejectedException.class) + public void testEndpointWithSelectorRejects() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "endpointWithSelectors.xml", this.getClass()); + MessageHandler endpoint = (MessageHandler) context.getBean("endpoint"); + ((Lifecycle) endpoint).start(); + endpoint.handle(new GenericMessage(123)); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/HandlerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/HandlerParserTests.java new file mode 100644 index 0000000000..2ca42e6b87 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/HandlerParserTests.java @@ -0,0 +1,62 @@ +/* + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import org.junit.Test; + +import org.springframework.context.ApplicationContext; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class HandlerParserTests { + + @Test + public void testTopLevelHandlerAdapter() { + ApplicationContext context = new ClassPathXmlApplicationContext( + "handlerAdapterParserTests.xml", HandlerParserTests.class); + MessageHandler adapter = (MessageHandler) context.getBean("handlerAdapter"); + assertNotNull(adapter); + Message reply = adapter.handle(new StringMessage("foo")); + assertNotNull(reply); + assertEquals("bar", reply.getPayload()); + } + + @Test + public void testHandlerChain() { + ApplicationContext context = new ClassPathXmlApplicationContext( + "handlerChainParserTests.xml", HandlerParserTests.class); + TestBean testBean = (TestBean) context.getBean("testBean"); + assertNull(testBean.getMessage()); + MessageHandler handlerChain = (MessageHandler) context.getBean("handlerChain"); + assertNotNull(handlerChain); + Message reply = handlerChain.handle(new StringMessage("test")); + assertNotNull(reply); + assertEquals(0, testBean.getLatch().getCount()); + assertEquals("foo", testBean.getMessage()); + assertEquals("bar", reply.getPayload()); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java index 62b113d5a3..cf581b5876 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java @@ -18,9 +18,12 @@ package org.springframework.integration.config; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; import org.junit.Test; +import org.springframework.beans.factory.BeanCreationException; +import org.springframework.beans.factory.BeanDefinitionStoreException; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.MessagingConfigurationException; @@ -58,7 +61,6 @@ public class MessageBusParserTests { MessageBus bus = (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); Subscription subscription = new Subscription("unknownChannel"); bus.registerHandler("handler", TestHandlers.nullHandler(), subscription); - bus.start(); } @Test @@ -73,4 +75,39 @@ public class MessageBusParserTests { bus.stop(); } + @Test + public void testMultipleMessageBusElements() { + boolean exceptionThrown = false; + try { + new ClassPathXmlApplicationContext("multipleMessageBusElements.xml", this.getClass()); + } + catch (BeanDefinitionStoreException e) { + exceptionThrown = true; + assertEquals(MessagingConfigurationException.class, e.getCause().getClass()); + } + assertTrue(exceptionThrown); + } + + @Test + public void testMessageBusElementAndBean() { + boolean exceptionThrown = false; + try { + new ClassPathXmlApplicationContext("messageBusElementAndBean.xml", this.getClass()); + } + catch (BeanCreationException e) { + exceptionThrown = true; + assertEquals(MessagingConfigurationException.class, e.getCause().getClass()); + } + assertTrue(exceptionThrown); + } + + @Test + public void testAutoStartup() { + ApplicationContext context = new ClassPathXmlApplicationContext( + "messageBusWithAutoStartup.xml", this.getClass()); + MessageBus bus = (MessageBus) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME); + assertTrue(bus.isRunning()); + bus.stop(); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/TestBean.java b/spring-integration-core/src/test/java/org/springframework/integration/config/TestBean.java index 24e5eaf84b..83845b3ad6 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/TestBean.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/TestBean.java @@ -27,21 +27,30 @@ public class TestBean { private CountDownLatch latch; + private String replyMessageText = null; + public TestBean(int countdown) { this.latch = new CountDownLatch(countdown); } + + public void setReplyMessageText(String replyMessageText) { + this.replyMessageText = replyMessageText; + } + public CountDownLatch getLatch() { return this.latch; } - public void store(String message) { + public String store(String message) { this.message = message; latch.countDown(); + return this.replyMessageText; } public String getMessage() { return this.message; } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/TestChannelInterceptor.java b/spring-integration-core/src/test/java/org/springframework/integration/config/TestChannelInterceptor.java new file mode 100644 index 0000000000..6e31f9f038 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/TestChannelInterceptor.java @@ -0,0 +1,54 @@ +/* + * 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 java.util.concurrent.atomic.AtomicInteger; + +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter; +import org.springframework.integration.message.Message; + +/** + * @author Mark Fisher + */ +public class TestChannelInterceptor extends ChannelInterceptorAdapter { + + private final AtomicInteger sendCount = new AtomicInteger(); + + private final AtomicInteger receiveCount = new AtomicInteger(); + + + @Override + public boolean preSend(Message message, MessageChannel channel) { + sendCount.incrementAndGet(); + return true; + } + + @Override + public void postReceive(Message message, MessageChannel channel) { + receiveCount.incrementAndGet(); + } + + public int getSendCount() { + return this.sendCount.get(); + } + + public int getReceiveCount() { + return this.receiveCount.get(); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/TestHandler.java b/spring-integration-core/src/test/java/org/springframework/integration/config/TestHandler.java index 8c4c6274c2..32459ea1b0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/TestHandler.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/TestHandler.java @@ -20,6 +20,7 @@ import java.util.concurrent.CountDownLatch; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher @@ -30,15 +31,22 @@ public class TestHandler implements MessageHandler { private CountDownLatch latch; + private String replyMessageText = null; + public TestHandler(int countdown) { this.latch = new CountDownLatch(countdown); } + + public void setReplyMessageText(String replyMessageText) { + this.replyMessageText = replyMessageText; + } + public Message handle(Message message) { this.messageString = (String) message.getPayload(); this.latch.countDown(); - return null; + return (this.replyMessageText != null) ? new StringMessage(this.replyMessageText) : null; } public String getMessageString() { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/channelInterceptorParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/channelInterceptorParserTests.xml new file mode 100644 index 0000000000..f10bd1874d --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/channelInterceptorParserTests.xml @@ -0,0 +1,16 @@ + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/endpointWithHandlerChildElement.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/endpointWithHandlerChildElement.xml new file mode 100644 index 0000000000..0bea46b378 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/endpointWithHandlerChildElement.xml @@ -0,0 +1,23 @@ + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/endpointWithSelectors.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/endpointWithSelectors.xml new file mode 100644 index 0000000000..1d94e16c4f --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/endpointWithSelectors.xml @@ -0,0 +1,29 @@ + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/handlerAdapterParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/handlerAdapterParserTests.xml new file mode 100644 index 0000000000..b15b4bcbe5 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/handlerAdapterParserTests.xml @@ -0,0 +1,17 @@ + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/handlerChainParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/handlerChainParserTests.xml new file mode 100644 index 0000000000..492422d81e --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/handlerChainParserTests.xml @@ -0,0 +1,28 @@ + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/messageBusElementAndBean.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/messageBusElementAndBean.xml new file mode 100644 index 0000000000..d7178f4234 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/messageBusElementAndBean.xml @@ -0,0 +1,14 @@ + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/messageBusWithAutoStartup.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/messageBusWithAutoStartup.xml new file mode 100644 index 0000000000..88cf9fe4dc --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/messageBusWithAutoStartup.xml @@ -0,0 +1,12 @@ + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/multipleMessageBusElements.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/multipleMessageBusElements.xml new file mode 100644 index 0000000000..3ab2907afe --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/multipleMessageBusElements.xml @@ -0,0 +1,14 @@ + + + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/simpleEndpointTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/simpleEndpointTests.xml index 58fd2a9bd6..1b00328c80 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/simpleEndpointTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/simpleEndpointTests.xml @@ -14,7 +14,7 @@ - +