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 54acaf017a..d25cf3e036 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 @@ -36,7 +36,7 @@ import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.dispatcher.DefaultMessageDispatcher; -import org.springframework.integration.dispatcher.MessageDispatcher; +import org.springframework.integration.dispatcher.SchedulingMessageDispatcher; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.DefaultMessageEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; @@ -63,7 +63,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif private Map endpoints = new ConcurrentHashMap(); - private Map dispatchers = new ConcurrentHashMap(); + private Map dispatchers = new ConcurrentHashMap(); private List lifecycleSourceAdapters = new CopyOnWriteArrayList(); @@ -194,7 +194,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif public MessageChannel unregisterChannel(String name) { MessageChannel removedChannel = this.channelRegistry.unregisterChannel(name); if (removedChannel != null) { - MessageDispatcher removedDispatcher = this.dispatchers.remove(removedChannel); + SchedulingMessageDispatcher removedDispatcher = this.dispatchers.remove(removedChannel); if (removedDispatcher != null && removedDispatcher.isRunning()) { removedDispatcher.stop(); } @@ -300,7 +300,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } private void registerWithDispatcher(MessageChannel channel, MessageHandler handler, Schedule schedule) { - MessageDispatcher dispatcher = dispatchers.get(channel); + SchedulingMessageDispatcher dispatcher = dispatchers.get(channel); if (dispatcher == null) { if (logger.isWarnEnabled()) { logger.warn("no dispatcher available for channel '" + channel.getName() + "', be sure to register the channel"); @@ -329,7 +329,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif synchronized (this.lifecycleMonitor) { this.activateEndpoints(); this.taskScheduler.start(); - for (MessageDispatcher dispatcher : this.dispatchers.values()) { + for (SchedulingMessageDispatcher dispatcher : this.dispatchers.values()) { dispatcher.start(); if (logger.isInfoEnabled()) { logger.info("started dispatcher '" + dispatcher + "'"); @@ -362,7 +362,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif logger.info("stopped source adapter '" + adapter + "'"); } } - for (MessageDispatcher dispatcher : this.dispatchers.values()) { + for (SchedulingMessageDispatcher dispatcher : this.dispatchers.values()) { dispatcher.stop(); if (logger.isInfoEnabled()) { logger.info("stopped dispatcher '" + dispatcher + "'"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptor.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptor.java index 000c4517e4..32ec9d1f7b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptor.java @@ -27,12 +27,12 @@ import org.springframework.integration.message.Message; */ public interface ChannelInterceptor { - boolean preSend(Message message, MessageChannel channel); + boolean preSend(Message message, MessageChannel channel); - void postSend(Message message, MessageChannel channel, boolean sent); + void postSend(Message message, MessageChannel channel, boolean sent); boolean preReceive(MessageChannel channel); - void postReceive(Message message, MessageChannel channel); + void postReceive(Message message, MessageChannel channel); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelPurger.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelPurger.java new file mode 100644 index 0000000000..a72b8b89cc --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelPurger.java @@ -0,0 +1,64 @@ +/* + * 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.channel; + +import java.util.List; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.selector.MessageSelector; + +/** + * A utility class for purging {@link Message Messages} from a + * {@link MessageChannel}. Any message that does not match the + * provided {@link MessageSelector} will be removed from the channel. If no + * {@link MessageSelector} is provided, then all messages will be + * cleared from the channel. + *

+ * Note that the {@link #purge()} method operates on a snapshot of the messages + * within a channel at the time that the method is invoked. It is therefore + * possible that new messages will arrive on the channel during the purge + * operation and thus will not be removed. Likewise, messages to be + * purged may have been removed from the channel while the operation is taking + * place. Such messages will not be included in the returned list. + * + * @author Mark Fisher + */ +public class ChannelPurger { + + private MessageChannel channel; + + private MessageSelector selector; + + + public ChannelPurger(MessageChannel channel) { + this.channel = channel; + } + + public ChannelPurger(MessageChannel channel, MessageSelector selector) { + this(channel); + this.selector = selector; + } + + + public final List> purge() { + if (this.selector == null) { + return this.channel.clear(); + } + return this.channel.purge(this.selector); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java index 5f0688cfe3..95794478a8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java @@ -20,6 +20,7 @@ import java.util.List; import org.springframework.integration.dispatcher.DispatcherPolicy; import org.springframework.integration.message.Message; +import org.springframework.integration.message.selector.MessageSelector; /** * Base channel interface defining common behavior for message sending and receiving. @@ -88,11 +89,11 @@ public interface MessageChannel { /** * Remove all {@link Message Messages} from this channel. */ - List clear(); + List> clear(); /** - * Remove any expired {@link Message Messages} from this channel. + * Remove any {@link Message Messages} that are not accepted by the provided selector. */ - List purge(); + List> purge(MessageSelector selector); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java index 0a43c02a59..67997bdeb9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java @@ -25,6 +25,7 @@ import java.util.concurrent.TimeUnit; import org.springframework.integration.dispatcher.DispatcherPolicy; import org.springframework.integration.message.Message; +import org.springframework.integration.message.selector.MessageSelector; import org.springframework.util.Assert; /** @@ -111,20 +112,21 @@ public class SimpleChannel extends AbstractMessageChannel { } } - public List clear() { - List clearedMessages = new ArrayList(); + public List> clear() { + List> clearedMessages = new ArrayList>(); this.queue.drainTo(clearedMessages); return clearedMessages; } - public List purge() { - List purgedMessages = new ArrayList(); - // take a snapshot + public List> purge(MessageSelector selector) { + if (selector == null) { + return this.clear(); + } + List> purgedMessages = new ArrayList>(); Object[] array = this.queue.toArray(); for (Object o : array) { - Message message = (Message) o; - if (message.isExpired() && this.queue.remove(message)) { - // message was still in the queue + Message message = (Message) o; + if (!selector.accept(message) && this.queue.remove(message)) { purgedMessages.add(message); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptorAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ChannelInterceptorAdapter.java similarity index 71% rename from spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptorAdapter.java rename to spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ChannelInterceptorAdapter.java index 7ef293efe0..07cf53b07b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelInterceptorAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/ChannelInterceptorAdapter.java @@ -14,8 +14,10 @@ * limitations under the License. */ -package org.springframework.integration.channel; +package org.springframework.integration.channel.interceptor; +import org.springframework.integration.channel.ChannelInterceptor; +import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.Message; /** @@ -26,18 +28,18 @@ import org.springframework.integration.message.Message; */ public class ChannelInterceptorAdapter implements ChannelInterceptor { - public boolean preSend(Message message, MessageChannel channel) { + public boolean preSend(Message message, MessageChannel channel) { return true; } - public void postSend(Message message, MessageChannel channel, boolean sent) { + public void postSend(Message message, MessageChannel channel, boolean sent) { } public boolean preReceive(MessageChannel channel) { return true; } - public void postReceive(Message message, MessageChannel channel) { + public void postReceive(Message message, MessageChannel channel) { } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/MessageSelectingInterceptor.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/MessageSelectingInterceptor.java new file mode 100644 index 0000000000..0502eefaf2 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/interceptor/MessageSelectingInterceptor.java @@ -0,0 +1,55 @@ +/* + * 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.channel.interceptor; + +import java.util.Arrays; +import java.util.List; + +import org.springframework.integration.MessageDeliveryException; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.selector.MessageSelector; + +/** + * A {@link org.springframework.integration.channel.ChannelInterceptor} that + * delegates to a list of {@link MessageSelector MessageSelectors} to decide + * whether a {@link Message} should be accepted on the {@link MessageChannel}. + * + * @author Mark Fisher + */ +public class MessageSelectingInterceptor extends ChannelInterceptorAdapter { + + private final List selectors; + + + public MessageSelectingInterceptor(MessageSelector... selectors) { + this.selectors = Arrays.asList(selectors); + } + + + @Override + public boolean preSend(Message message, MessageChannel channel) { + for (MessageSelector selector : this.selectors) { + if (!selector.accept(message)) { + throw new MessageDeliveryException( + "selector '" + selector + "' did not accept message '" + message + "'"); + } + } + return true; + } + +} 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 800ff0d7bd..b28606eab5 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 @@ -21,12 +21,18 @@ import org.w3c.dom.Node; import org.w3c.dom.NodeList; 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.GenericBeanDefinition; +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.channel.SimpleChannel; +import org.springframework.integration.channel.interceptor.MessageSelectingInterceptor; import org.springframework.integration.dispatcher.DispatcherPolicy; +import org.springframework.integration.message.selector.MessageSelector; +import org.springframework.integration.message.selector.PayloadTypeSelector; import org.springframework.util.StringUtils; /** @@ -44,6 +50,10 @@ public class ChannelParser implements BeanDefinitionParser { private static final String DISPATCHER_POLICY_ELEMENT = "dispatcher-policy"; + private static final String DATATYPE_ATTRIBUTE = "datatype"; + + private static final String INTERCEPTORS_PROPERTY = "interceptors"; + public BeanDefinition parse(Element element, ParserContext parserContext) { RootBeanDefinition channelDef = new RootBeanDefinition(SimpleChannel.class); @@ -64,6 +74,24 @@ public class ChannelParser implements BeanDefinitionParser { 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); + RootBeanDefinition selectorDef = new RootBeanDefinition(PayloadTypeSelector.class); + selectorDef.getConstructorArgumentValues().addGenericArgumentValue(datatypes); + String selectorBeanName = parserContext.getReaderContext().generateBeanName(selectorDef); + BeanComponentDefinition selectorComponent = new BeanComponentDefinition(selectorDef, selectorBeanName); + parserContext.registerBeanComponent(selectorComponent); + RootBeanDefinition interceptorDef = new RootBeanDefinition(MessageSelectingInterceptor.class); + interceptorDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference(selectorBeanName)); + String interceptorBeanName = parserContext.getReaderContext().generateBeanName(interceptorDef); + BeanComponentDefinition interceptorComponent = new BeanComponentDefinition(interceptorDef, interceptorBeanName); + 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)); return channelDef; 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 cdc6e7c3c2..519c85f580 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 @@ -53,6 +53,7 @@ + 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 150db459b8..db405335f3 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 @@ -16,11 +16,12 @@ 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.CopyOnWriteArrayList; -import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -28,6 +29,8 @@ import org.apache.commons.logging.LogFactory; import org.springframework.context.Lifecycle; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Message; +import org.springframework.integration.scheduling.MessagingTask; import org.springframework.integration.scheduling.MessagingTaskScheduler; import org.springframework.integration.scheduling.MessagingTaskSchedulerAware; import org.springframework.integration.scheduling.PollingSchedule; @@ -44,11 +47,13 @@ import org.springframework.util.Assert; * * @author Mark Fisher */ -public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTaskSchedulerAware { +public class DefaultMessageDispatcher implements SchedulingMessageDispatcher, MessagingTaskSchedulerAware { protected Log logger = LogFactory.getLog(this.getClass()); - private MessageChannel channel; + private final MessageChannel channel; + + private final MessageRetriever retriever; private MessagingTaskScheduler scheduler; @@ -56,7 +61,7 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas private Map> scheduledHandlers = new ConcurrentHashMap>(); - private List> futures = new CopyOnWriteArrayList>(); + private AtomicLong totalMessagesProcessed = new AtomicLong(); private volatile boolean running; @@ -66,6 +71,7 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas public DefaultMessageDispatcher(MessageChannel channel) { Assert.notNull(channel, "'channel' must not be null"); this.channel = channel; + this.retriever = new ChannelPollingMessageRetriever(this.channel); } @@ -127,21 +133,14 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas return; } synchronized (this.lifecycleMonitor) { - for (Map.Entry> entry : this.scheduledHandlers.entrySet()) { - Schedule schedule = entry.getKey(); - List handlers = entry.getValue(); - DispatcherTask task = new DispatcherTask(channel); - task.setSchedule(schedule); + for (Schedule schedule : this.scheduledHandlers.keySet()) { + List handlers = this.scheduledHandlers.get(schedule); for (MessageHandler handler : handlers) { if (handler instanceof Lifecycle) { ((Lifecycle) handler).start(); } - task.addHandler(handler); - } - ScheduledFuture future = this.scheduler.schedule(task); - if (future != null) { - futures.add(future); } + this.scheduler.schedule(new DispatcherTask(schedule)); } this.running = true; } @@ -152,13 +151,10 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas return; } synchronized (this.lifecycleMonitor) { - for (ScheduledFuture future : this.futures) { - future.cancel(true); - for (List handlerList : scheduledHandlers.values()) { - for (MessageHandler handler : handlerList) { - if (handler instanceof Lifecycle) { - ((Lifecycle) handler).stop(); - } + for (List handlerList : scheduledHandlers.values()) { + for (MessageHandler handler : handlerList) { + if (handler instanceof Lifecycle) { + ((Lifecycle) handler).stop(); } } } @@ -166,4 +162,57 @@ public class DefaultMessageDispatcher implements MessageDispatcher, MessagingTas } } + public int dispatch() { + MessageDistributor distributor = this.getDistributor(this.defaultSchedule); + return this.doDispatch(distributor); + } + + private int doDispatch(MessageDistributor distributor) { + int messagesProcessed = 0; + Collection> messages = this.retriever.retrieveMessages(); + if (messages == null) { + return 0; + } + for (Message message : messages) { + if (distributor.distribute(message)) { + messagesProcessed++; + } + } + totalMessagesProcessed.addAndGet(messagesProcessed); + return messagesProcessed; + } + + private MessageDistributor getDistributor(Schedule schedule) { + if (schedule == null) { + schedule = this.defaultSchedule; + } + MessageDistributor distributor = new DefaultMessageDistributor(this.channel.getDispatcherPolicy()); + for (MessageHandler handler : this.scheduledHandlers.get(schedule)) { + distributor.addHandler(handler); + } + return distributor; + } + + + private class DispatcherTask implements MessagingTask { + + private Schedule schedule; + + private MessageDistributor distributor; + + + public DispatcherTask(Schedule schedule) { + this.schedule = (schedule != null) ? schedule : defaultSchedule; + this.distributor = getDistributor(this.schedule); + } + + public Schedule getSchedule() { + return this.schedule; + } + + public void run() { + doDispatch(this.distributor); + } + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java similarity index 63% rename from spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java rename to spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java index 945e4e4417..353199e554 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DispatcherTask.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultMessageDistributor.java @@ -17,7 +17,6 @@ package org.springframework.integration.dispatcher; import java.util.ArrayList; -import java.util.Collection; import java.util.Iterator; import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; @@ -27,87 +26,44 @@ import org.apache.commons.logging.LogFactory; import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.MessageHandlingException; -import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.MessageHandlerNotRunningException; import org.springframework.integration.message.Message; import org.springframework.integration.message.selector.MessageSelectorRejectedException; -import org.springframework.integration.scheduling.MessagingTask; -import org.springframework.integration.scheduling.Schedule; import org.springframework.util.Assert; /** - * A task for polling {@link MessageDispatcher MessageDispatchers}. If - * {@link #broadcast} is set to false (the default), each message - * will be sent to a single {@link MessageHandler}. Otherwise, each - * retrieved {@link Message} will be sent to all of the handlers. + * Default implementation of the {@link MessageDistributor} interface. * * @author Mark Fisher */ -public class DispatcherTask implements MessagingTask { +public class DefaultMessageDistributor implements MessageDistributor { - private Log logger = LogFactory.getLog(this.getClass()); + private final Log logger = LogFactory.getLog(this.getClass()); - private Schedule schedule; + private final List handlers = new CopyOnWriteArrayList(); - private DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - - private MessageRetriever retriever; - - private List handlers = new CopyOnWriteArrayList(); + private final DispatcherPolicy dispatcherPolicy; - public DispatcherTask(MessageChannel channel) { - Assert.notNull(channel, "'channel' must not be null"); - this.retriever = new ChannelPollingMessageRetriever(channel); - DispatcherPolicy dispatcherPolicy = channel.getDispatcherPolicy(); - if (dispatcherPolicy != null) { - this.dispatcherPolicy = dispatcherPolicy; - } + public DefaultMessageDistributor(DispatcherPolicy dispatcherPolicy) { + Assert.notNull(dispatcherPolicy, "'dispatcherPolicy' must not be null"); + this.dispatcherPolicy = dispatcherPolicy; } - public void setSchedule(Schedule schedule) { - Assert.notNull(schedule, "'schedule' must not be null"); - this.schedule = schedule; - } - - public Schedule getSchedule() { - return this.schedule; - } - public void addHandler(MessageHandler handler) { - Assert.notNull(handler, "'handler' must not be null"); this.handlers.add(handler); } - /** - * Retrieves messages and dispatches to the executors. - * - * @return the number of messages processed - */ - public int dispatch() { - int messagesProcessed = 0; - Collection> messages = this.retriever.retrieveMessages(); - if (messages == null) { - return 0; - } - for (Message message : messages) { - if (dispatchMessage(message)) { - messagesProcessed++; - } - } - return messagesProcessed; - } - - protected boolean dispatchMessage(Message message) { + public boolean distribute(Message message) { int attempts = 0; List targets = new ArrayList(this.handlers); while (attempts < this.dispatcherPolicy.getRejectionLimit()) { if (attempts > 0) { if (logger.isDebugEnabled()) { - logger.debug("handler(s) rejected message after " + attempts - + " attempt(s), will try again after 'retryInterval' of " + + logger.debug("handler(s) rejected message after " + attempts + + " attempt(s), will try again after 'retryInterval' of " + this.dispatcherPolicy.getRetryInterval() + " milliseconds"); } try { @@ -164,8 +120,4 @@ public class DispatcherTask implements MessagingTask { return false; } - public void run() { - this.dispatch(); - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java index 154df946f1..0d979fc358 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java @@ -16,19 +16,17 @@ package org.springframework.integration.dispatcher; -import org.springframework.context.Lifecycle; import org.springframework.integration.handler.MessageHandler; -import org.springframework.integration.scheduling.Schedule; /** * Strategy interface for dispatching messages. * * @author Mark Fisher */ -public interface MessageDispatcher extends Lifecycle { +public interface MessageDispatcher { void addHandler(MessageHandler handler); - void addHandler(MessageHandler handler, Schedule schedule); + int dispatch(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDistributor.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDistributor.java new file mode 100644 index 0000000000..0819728a5e --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDistributor.java @@ -0,0 +1,34 @@ +/* + * 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.dispatcher; + +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Message; + +/** + * Strategy interface for distributing a {@link Message} to one or more + * {@link MessageHandler MessageHandlers}. + * + * @author Mark Fisher + */ +public interface MessageDistributor { + + void addHandler(MessageHandler handler); + + boolean distribute(Message message); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SchedulingMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SchedulingMessageDispatcher.java new file mode 100644 index 0000000000..c1239b22cd --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SchedulingMessageDispatcher.java @@ -0,0 +1,35 @@ +/* + * 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.dispatcher; + +import org.springframework.context.Lifecycle; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.scheduling.Schedule; + +/** + * An extension to the {@link MessageDispatcher} strategy for handlers that may + * be scheduled. + * + * @author Mark Fisher + */ +public interface SchedulingMessageDispatcher extends MessageDispatcher, Lifecycle { + + void setDefaultSchedule(Schedule defaultSchedule); + + void addHandler(MessageHandler handler, Schedule schedule); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/message/selector/UnexpiredMessageSelector.java b/spring-integration-core/src/main/java/org/springframework/integration/message/selector/UnexpiredMessageSelector.java new file mode 100644 index 0000000000..3faf8fcc44 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/message/selector/UnexpiredMessageSelector.java @@ -0,0 +1,33 @@ +/* + * 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.message.selector; + +import org.springframework.integration.message.Message; + +/** + * A {@link MessageSelector} that accepts {@link Message Messages} that are + * not expired. + * + * @author Mark Fisher + */ +public class UnexpiredMessageSelector implements MessageSelector { + + public boolean accept(Message message) { + return (!message.isExpired()); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java index 05d3cdd68b..1485b30f75 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java @@ -23,10 +23,10 @@ import java.io.IOException; import org.junit.Test; -import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.dispatcher.DefaultMessageDispatcher; import org.springframework.integration.dispatcher.DispatcherPolicy; -import org.springframework.integration.dispatcher.DispatcherTask; +import org.springframework.integration.dispatcher.MessageDispatcher; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; @@ -38,13 +38,8 @@ public class ByteStreamTargetAdapterTests { @Test public void testSingleByteArray() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); - MessageChannel channel = new SimpleChannel(); ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream); - DispatcherTask dispatcherTask = new DispatcherTask(channel); - dispatcherTask.addHandler(adapter); - channel.send(new GenericMessage(new byte[] {1,2,3})); - int count = dispatcherTask.dispatch(); - assertEquals(1, count); + adapter.handle(new GenericMessage(new byte[] {1,2,3})); byte[] result = stream.toByteArray(); assertEquals(3, result.length); assertEquals(1, result[0]); @@ -55,13 +50,8 @@ public class ByteStreamTargetAdapterTests { @Test public void testSingleString() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); - MessageChannel channel = new SimpleChannel(); ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream); - DispatcherTask dispatcherTask = new DispatcherTask(channel); - dispatcherTask.addHandler(adapter); - channel.send(new StringMessage("foo")); - int count = dispatcherTask.dispatch(); - assertEquals(1, count); + adapter.handle(new StringMessage("foo")); byte[] result = stream.toByteArray(); assertEquals(3, result.length); assertEquals("foo", new String(result)); @@ -74,12 +64,12 @@ public class ByteStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(3); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - DispatcherTask dispatcherTask = new DispatcherTask(channel); - dispatcherTask.addHandler(adapter); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + dispatcher.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - assertEquals(3, dispatcherTask.dispatch()); + assertEquals(3, dispatcher.dispatch()); byte[] result = stream.toByteArray(); assertEquals(9, result.length); assertEquals(1, result[0]); @@ -93,12 +83,12 @@ public class ByteStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(2); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - DispatcherTask dispatcherTask = new DispatcherTask(channel); - dispatcherTask.addHandler(adapter); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + dispatcher.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - assertEquals(2, dispatcherTask.dispatch()); + assertEquals(2, dispatcher.dispatch()); byte[] result = stream.toByteArray(); assertEquals(6, result.length); assertEquals(1, result[0]); @@ -112,12 +102,12 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(5); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - DispatcherTask dispatcherTask = new DispatcherTask(channel); - dispatcherTask.addHandler(adapter); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + dispatcher.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - assertEquals(3, dispatcherTask.dispatch()); + assertEquals(3, dispatcher.dispatch()); byte[] result = stream.toByteArray(); assertEquals(9, result.length); assertEquals(1, result[0]); @@ -131,16 +121,16 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(2); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - DispatcherTask dispatcherTask = new DispatcherTask(channel); - dispatcherTask.addHandler(adapter); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + dispatcher.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - assertEquals(2, dispatcherTask.dispatch()); + assertEquals(2, dispatcher.dispatch()); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); assertEquals(1, result1[0]); - assertEquals(1, dispatcherTask.dispatch()); + assertEquals(1, dispatcher.dispatch()); byte[] result2 = stream.toByteArray(); assertEquals(9, result2.length); assertEquals(1, result2[0]); @@ -155,16 +145,16 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(5); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - DispatcherTask dispatcherTask = new DispatcherTask(channel); - dispatcherTask.addHandler(adapter); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + dispatcher.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - assertEquals(3, dispatcherTask.dispatch()); + assertEquals(3, dispatcher.dispatch()); byte[] result1 = stream.toByteArray(); assertEquals(9, result1.length); assertEquals(1, result1[0]); - assertEquals(0, dispatcherTask.dispatch()); + assertEquals(0, dispatcher.dispatch()); byte[] result2 = stream.toByteArray(); assertEquals(9, result2.length); assertEquals(1, result2[0]); @@ -178,16 +168,16 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(2); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - DispatcherTask dispatcherTask = new DispatcherTask(channel); - dispatcherTask.addHandler(adapter); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + dispatcher.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - assertEquals(2, dispatcherTask.dispatch()); + assertEquals(2, dispatcher.dispatch()); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); stream.reset(); - assertEquals(1, dispatcherTask.dispatch()); + assertEquals(1, dispatcher.dispatch()); byte[] result2 = stream.toByteArray(); assertEquals(3, result2.length); assertEquals(7, result2[0]); @@ -201,17 +191,17 @@ public class ByteStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(2); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - DispatcherTask dispatcherTask = new DispatcherTask(channel); - dispatcherTask.addHandler(adapter); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + dispatcher.addHandler(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - assertEquals(2, dispatcherTask.dispatch()); + assertEquals(2, dispatcher.dispatch()); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); stream.write(new byte[] {123}); stream.flush(); - assertEquals(1, dispatcherTask.dispatch()); + assertEquals(1, dispatcher.dispatch()); byte[] result2 = stream.toByteArray(); assertEquals(10, result2.length); assertEquals(1, result2[0]); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java index 5449db7c90..6b3241362d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java @@ -24,8 +24,9 @@ import org.junit.Test; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.dispatcher.DefaultMessageDispatcher; import org.springframework.integration.dispatcher.DispatcherPolicy; -import org.springframework.integration.dispatcher.DispatcherTask; +import org.springframework.integration.dispatcher.MessageDispatcher; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; @@ -37,13 +38,8 @@ public class CharacterStreamTargetAdapterTests { @Test public void testSingleString() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); - MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); - DispatcherTask dispatcherTask = new DispatcherTask(channel); - dispatcherTask.addHandler(adapter); - channel.send(new StringMessage("foo")); - int count = dispatcherTask.dispatch(); - assertEquals(1, count); + adapter.handle(new StringMessage("foo")); String result = new String(stream.toByteArray()); assertEquals("foo", result); } @@ -53,14 +49,14 @@ public class CharacterStreamTargetAdapterTests { ByteArrayOutputStream stream = new ByteArrayOutputStream(); MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); - DispatcherTask dispatcherTask = new DispatcherTask(channel); - dispatcherTask.addHandler(adapter); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + dispatcher.addHandler(adapter); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - assertEquals(1, dispatcherTask.dispatch()); + assertEquals(1, dispatcher.dispatch()); String result1 = new String(stream.toByteArray()); assertEquals("foo", result1); - assertEquals(1, dispatcherTask.dispatch()); + assertEquals(1, dispatcher.dispatch()); String result2 = new String(stream.toByteArray()); assertEquals("foobar", result2); } @@ -71,15 +67,15 @@ public class CharacterStreamTargetAdapterTests { MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); adapter.setShouldAppendNewLine(true); - DispatcherTask dispatcherTask = new DispatcherTask(channel); - dispatcherTask.addHandler(adapter); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + dispatcher.addHandler(adapter); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - assertEquals(1, dispatcherTask.dispatch()); + assertEquals(1, dispatcher.dispatch()); String result1 = new String(stream.toByteArray()); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine, result1); - assertEquals(1, dispatcherTask.dispatch()); + assertEquals(1, dispatcher.dispatch()); String result2 = new String(stream.toByteArray()); assertEquals("foo" + newLine + "bar" + newLine, result2); } @@ -91,11 +87,11 @@ public class CharacterStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(2); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - DispatcherTask dispatcherTask = new DispatcherTask(channel); - dispatcherTask.addHandler(adapter); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + dispatcher.addHandler(adapter); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - assertEquals(2, dispatcherTask.dispatch()); + assertEquals(2, dispatcher.dispatch()); String result = new String(stream.toByteArray()); assertEquals("foobar", result); } @@ -108,12 +104,12 @@ public class CharacterStreamTargetAdapterTests { dispatcherPolicy.setMaxMessagesPerTask(10); dispatcherPolicy.setReceiveTimeout(0); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - DispatcherTask dispatcherTask = new DispatcherTask(channel); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); adapter.setShouldAppendNewLine(true); - dispatcherTask.addHandler(adapter); + dispatcher.addHandler(adapter); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - assertEquals(2, dispatcherTask.dispatch()); + assertEquals(2, dispatcher.dispatch()); String result = new String(stream.toByteArray()); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine + "bar" + newLine, result); @@ -124,11 +120,11 @@ public class CharacterStreamTargetAdapterTests { ByteArrayOutputStream stream = new ByteArrayOutputStream(); MessageChannel channel = new SimpleChannel(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream); - DispatcherTask dispatcherTask = new DispatcherTask(channel); - dispatcherTask.addHandler(adapter); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + dispatcher.addHandler(adapter); TestObject testObject = new TestObject("foo"); channel.send(new GenericMessage(testObject)); - int count = dispatcherTask.dispatch(); + int count = dispatcher.dispatch(); assertEquals(1, count); String result = new String(stream.toByteArray()); assertEquals("foo", result); @@ -142,13 +138,13 @@ public class CharacterStreamTargetAdapterTests { dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(2); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - DispatcherTask dispatcherTask = new DispatcherTask(channel); - dispatcherTask.addHandler(adapter); + MessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); + dispatcher.addHandler(adapter); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); channel.send(new GenericMessage(testObject1), 0); channel.send(new GenericMessage(testObject2), 0); - assertEquals(2, dispatcherTask.dispatch()); + assertEquals(2, dispatcher.dispatch()); String result = new String(stream.toByteArray()); assertEquals("foobar", result); } @@ -161,14 +157,14 @@ public class CharacterStreamTargetAdapterTests { dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(2); SimpleChannel channel = new SimpleChannel(dispatcherPolicy); - DispatcherTask dispatcherTask = new DispatcherTask(channel); + DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel); adapter.setShouldAppendNewLine(true); - dispatcherTask.addHandler(adapter); + dispatcher.addHandler(adapter); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); channel.send(new GenericMessage(testObject1), 0); channel.send(new GenericMessage(testObject2), 0); - assertEquals(2, dispatcherTask.dispatch()); + dispatcher.dispatch(); String result = new String(stream.toByteArray()); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine + "bar" + newLine, result); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelPurgerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelPurgerTests.java new file mode 100644 index 0000000000..be14820a0b --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelPurgerTests.java @@ -0,0 +1,101 @@ +/* + * 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.channel; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import java.util.List; + +import org.junit.Test; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; +import org.springframework.integration.message.selector.MessageSelector; + +/** + * @author Mark Fisher + */ +public class ChannelPurgerTests { + + @Test + public void testPurgeAllWithoutSelector() { + MessageChannel channel = new SimpleChannel(); + channel.send(new StringMessage("test1")); + channel.send(new StringMessage("test2")); + channel.send(new StringMessage("test3")); + ChannelPurger purger = new ChannelPurger(channel); + List> purgedMessages = purger.purge(); + assertEquals(3, purgedMessages.size()); + assertNull(channel.receive(0)); + } + + @Test + public void testPurgeAllWithSelector() { + MessageChannel channel = new SimpleChannel(); + channel.send(new StringMessage("test1")); + channel.send(new StringMessage("test2")); + channel.send(new StringMessage("test3")); + ChannelPurger purger = new ChannelPurger(channel, new MessageSelector() { + public boolean accept(Message message) { + return false; + } + }); + List> purgedMessages = purger.purge(); + assertEquals(3, purgedMessages.size()); + assertNull(channel.receive(0)); + } + + @Test + public void testPurgeNoneWithSelector() { + MessageChannel channel = new SimpleChannel(); + channel.send(new StringMessage("test1")); + channel.send(new StringMessage("test2")); + channel.send(new StringMessage("test3")); + ChannelPurger purger = new ChannelPurger(channel, new MessageSelector() { + public boolean accept(Message message) { + return true; + } + }); + List> purgedMessages = purger.purge(); + assertEquals(0, purgedMessages.size()); + assertNotNull(channel.receive(0)); + assertNotNull(channel.receive(0)); + assertNotNull(channel.receive(0)); + } + + @Test + public void testPurgeSubsetWithSelector() { + MessageChannel channel = new SimpleChannel(); + channel.send(new StringMessage("test1")); + channel.send(new StringMessage("test2")); + channel.send(new StringMessage("test3")); + ChannelPurger purger = new ChannelPurger(channel, new MessageSelector() { + public boolean accept(Message message) { + return (message.getPayload().equals("test2")); + } + }); + List> purgedMessages = purger.purge(); + assertEquals(2, purgedMessages.size()); + Message message = channel.receive(0); + assertNotNull(message); + assertEquals("test2", message.getPayload()); + assertNull(channel.receive(0)); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/SimpleChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/SimpleChannelTests.java index 38782d927b..64bc8419d4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/SimpleChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/SimpleChannelTests.java @@ -34,6 +34,7 @@ import org.junit.Test; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; +import org.springframework.integration.message.selector.UnexpiredMessageSelector; /** * @author Mark Fisher @@ -203,7 +204,7 @@ public class SimpleChannelTests { assertTrue(channel.send(message1)); assertTrue(channel.send(message2)); assertFalse(channel.send(message3, 0)); - List clearedMessages = channel.clear(); + List> clearedMessages = channel.clear(); assertNotNull(clearedMessages); assertEquals(2, clearedMessages.size()); assertTrue(channel.send(message3)); @@ -212,7 +213,7 @@ public class SimpleChannelTests { @Test public void testClearEmptyChannel() { SimpleChannel channel = new SimpleChannel(); - List clearedMessages = channel.clear(); + List> clearedMessages = channel.clear(); assertNotNull(clearedMessages); assertEquals(0, clearedMessages.size()); } @@ -231,7 +232,7 @@ public class SimpleChannelTests { assertTrue(channel.send(expiredMessage, 0)); assertTrue(channel.send(unexpiredMessage, 0)); assertFalse(channel.send(new StringMessage("atCapacity"), 0)); - List purgedMessages = channel.purge(); + List> purgedMessages = channel.purge(new UnexpiredMessageSelector()); assertNotNull(purgedMessages); assertEquals(1, purgedMessages.size()); assertTrue(channel.send(new StringMessage("roomAvailable"), 0)); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelInterceptorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java similarity index 97% rename from spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelInterceptorTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java index 94b9ce3373..414552da8b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelInterceptorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java @@ -14,10 +14,9 @@ * limitations under the License. */ -package org.springframework.integration.channel; +package org.springframework.integration.channel.interceptor; 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.assertSame; @@ -28,6 +27,8 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/MessageSelectingInterceptorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/MessageSelectingInterceptorTests.java new file mode 100644 index 0000000000..c93d0c9e1d --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/MessageSelectingInterceptorTests.java @@ -0,0 +1,110 @@ +/* + * 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.channel.interceptor; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Test; + +import org.springframework.integration.MessageDeliveryException; +import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; +import org.springframework.integration.message.selector.MessageSelector; + +/** + * @author Mark Fisher + */ +public class MessageSelectingInterceptorTests { + + @Test + public void testSingleSelectorAccepts() { + final AtomicInteger counter = new AtomicInteger(); + MessageSelector selector = new TestMessageSelector(true, counter); + MessageSelectingInterceptor interceptor = new MessageSelectingInterceptor(selector); + SimpleChannel channel = new SimpleChannel(); + channel.addInterceptor(interceptor); + assertTrue(channel.send(new StringMessage("test1"))); + } + + @Test(expected=MessageDeliveryException.class) + public void testSingleSelectorRejects() { + final AtomicInteger counter = new AtomicInteger(); + MessageSelector selector = new TestMessageSelector(false, counter); + MessageSelectingInterceptor interceptor = new MessageSelectingInterceptor(selector); + SimpleChannel channel = new SimpleChannel(); + channel.addInterceptor(interceptor); + channel.send(new StringMessage("test1")); + } + + @Test + public void testMultipleSelectorsAccept() { + final AtomicInteger counter = new AtomicInteger(); + MessageSelector selector1 = new TestMessageSelector(true, counter); + MessageSelector selector2 = new TestMessageSelector(true, counter); + MessageSelectingInterceptor interceptor = new MessageSelectingInterceptor(selector1, selector2); + SimpleChannel channel = new SimpleChannel(); + channel.addInterceptor(interceptor); + assertTrue(channel.send(new StringMessage("test1"))); + assertEquals(2, counter.get()); + } + + @Test + public void testMultipleSelectorsReject() { + boolean exceptionThrown = false; + final AtomicInteger counter = new AtomicInteger(); + MessageSelector selector1 = new TestMessageSelector(true, counter); + MessageSelector selector2 = new TestMessageSelector(false, counter); + MessageSelector selector3 = new TestMessageSelector(false, counter); + MessageSelector selector4 = new TestMessageSelector(true, counter); + MessageSelectingInterceptor interceptor = new MessageSelectingInterceptor(selector1, selector2, selector3, selector4); + SimpleChannel channel = new SimpleChannel(); + channel.addInterceptor(interceptor); + try { + channel.send(new StringMessage("test1")); + } + catch (MessageDeliveryException e) { + exceptionThrown = true; + } + assertTrue(exceptionThrown); + assertEquals(2, counter.get()); + } + + + private static class TestMessageSelector implements MessageSelector { + + private final boolean shouldAccept; + + private final AtomicInteger counter; + + + public TestMessageSelector(boolean shouldAccept, AtomicInteger counter) { + this.shouldAccept = shouldAccept; + this.counter = counter; + } + + + public boolean accept(Message message) { + this.counter.incrementAndGet(); + return this.shouldAccept; + } + } + +} 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 197f8cbc11..ed92d31b45 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 @@ -28,6 +28,7 @@ import org.junit.Test; import org.springframework.beans.FatalBeanException; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.MessageDeliveryException; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.dispatcher.DefaultMessageDispatcher; import org.springframework.integration.dispatcher.DispatcherPolicy; @@ -143,6 +144,48 @@ public class ChannelParserTests { assertFalse(dispatcherPolicy.getShouldFailOnRejectionLimit()); } + @Test + public void testDatatypeChannelWithCorrectType() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "channelParserTests.xml", this.getClass()); + MessageChannel channel = (MessageChannel) context.getBean("integerChannel"); + assertTrue(channel.send(new GenericMessage(123))); + } + + @Test(expected=MessageDeliveryException.class) + public void testDatatypeChannelWithIncorrectType() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "channelParserTests.xml", this.getClass()); + MessageChannel channel = (MessageChannel) context.getBean("integerChannel"); + channel.send(new StringMessage("incorrect type")); + } + + @Test + public void testDatatypeChannelWithAssignableSubTypes() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "channelParserTests.xml", this.getClass()); + MessageChannel channel = (MessageChannel) context.getBean("numberChannel"); + assertTrue(channel.send(new GenericMessage(123))); + assertTrue(channel.send(new GenericMessage(123.45))); + } + + @Test + public void testMultipleDatatypeChannelWithCorrectTypes() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "channelParserTests.xml", this.getClass()); + MessageChannel channel = (MessageChannel) context.getBean("stringOrNumberChannel"); + assertTrue(channel.send(new GenericMessage(123))); + assertTrue(channel.send(new StringMessage("accepted type"))); + } + + @Test(expected=MessageDeliveryException.class) + public void testMultipleDatatypeChannelWithIncorrectType() { + ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( + "channelParserTests.xml", this.getClass()); + MessageChannel channel = (MessageChannel) context.getBean("stringOrNumberChannel"); + channel.send(new GenericMessage(true)); + } + private static class TestHandler implements MessageHandler { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/channelParserTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/channelParserTests.xml index 6e06361a2c..e1fc000364 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/channelParserTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/channelParserTests.xml @@ -23,4 +23,10 @@ should-fail-on-rejection-limit="false"/> + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DispatcherTaskTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDistributorTests.java similarity index 59% rename from spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DispatcherTaskTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDistributorTests.java index 994333794b..71e4685ff3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DispatcherTaskTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultMessageDistributorTests.java @@ -24,52 +24,47 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.handler.TestHandlers; import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher */ -public class DispatcherTaskTests { +public class DefaultMessageDistributorTests { @Test - public void testSimpleDispatch() throws InterruptedException { - MessageChannel channel = new SimpleChannel(); - DispatcherTask task = new DispatcherTask(channel); + public void testSingleMessage() throws InterruptedException { + MessageDistributor distributor = new DefaultMessageDistributor(new DispatcherPolicy()); final CountDownLatch latch = new CountDownLatch(1); - task.addHandler(TestHandlers.countDownHandler(latch)); - task.dispatchMessage(new StringMessage("test")); + distributor.addHandler(TestHandlers.countDownHandler(latch)); + distributor.distribute(new StringMessage("test")); latch.await(500, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); } @Test - public void testDispatchWithPointToPointChannel() throws InterruptedException { - MessageChannel channel = new SimpleChannel(new DispatcherPolicy(false)); - DispatcherTask task = new DispatcherTask(channel); + public void testPointToPoint() throws InterruptedException { + MessageDistributor distributor = new DefaultMessageDistributor(new DispatcherPolicy(false)); final CountDownLatch latch = new CountDownLatch(1); final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); - task.addHandler(TestHandlers.countingCountDownHandler(counter1, latch)); - task.addHandler(TestHandlers.countingCountDownHandler(counter2, latch)); - task.dispatchMessage(new StringMessage("test")); + distributor.addHandler(TestHandlers.countingCountDownHandler(counter1, latch)); + distributor.addHandler(TestHandlers.countingCountDownHandler(counter2, latch)); + distributor.distribute(new StringMessage("test")); latch.await(500, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); assertEquals("only 1 handler should have received the message", 1, counter1.get() + counter2.get()); } @Test - public void testDispatchWithPublishSubscribeChannel() throws InterruptedException { - MessageChannel channel = new SimpleChannel(new DispatcherPolicy(true)); - DispatcherTask task = new DispatcherTask(channel); + public void testPublishSubscribe() throws InterruptedException { + MessageDistributor distributor = new DefaultMessageDistributor(new DispatcherPolicy(true)); final CountDownLatch latch = new CountDownLatch(2); final AtomicInteger counter1 = new AtomicInteger(); final AtomicInteger counter2 = new AtomicInteger(); - task.addHandler(TestHandlers.countingCountDownHandler(counter1, latch)); - task.addHandler(TestHandlers.countingCountDownHandler(counter2, latch)); - task.dispatchMessage(new StringMessage("test")); + distributor.addHandler(TestHandlers.countingCountDownHandler(counter1, latch)); + distributor.addHandler(TestHandlers.countingCountDownHandler(counter2, latch)); + distributor.distribute(new StringMessage("test")); latch.await(500, TimeUnit.MILLISECONDS); assertEquals(0, latch.getCount()); assertEquals(1, counter1.get()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/message/selector/UnexpiredMessageSelectorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/message/selector/UnexpiredMessageSelectorTests.java new file mode 100644 index 0000000000..c95d26ddc8 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/message/selector/UnexpiredMessageSelectorTests.java @@ -0,0 +1,52 @@ +/* + * 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.message.selector; + +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.util.Date; + +import org.junit.Test; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class UnexpiredMessageSelectorTests { + + @Test + public void testExpiredMessageRejected() { + long past = System.currentTimeMillis() - 60000; + Message message = new StringMessage("expired"); + message.getHeader().setExpiration(new Date(past)); + UnexpiredMessageSelector selector = new UnexpiredMessageSelector(); + assertFalse(selector.accept(message)); + } + + @Test + public void testUnexpiredMessageAccepted() { + long future = System.currentTimeMillis() + 60000; + Message message = new StringMessage("unexpired"); + message.getHeader().setExpiration(new Date(future)); + UnexpiredMessageSelector selector = new UnexpiredMessageSelector(); + assertTrue(selector.accept(message)); + } + +}