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

Refactored PollingDispatcher into separate ChannelPoller and SourcePoller implementations with an AbstractPoller base class.

Browse Source
This commit is contained in:
Mark Fisher
2008-09-07 15:23:17 +00:00
parent f0079d97f2
commit 15f9875b5b
25 changed files with 625 additions and 601 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -44,13 +44,14 @@ import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.DefaultChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.endpoint.AbstractPoller;
import org.springframework.integration.endpoint.ChannelPoller;
import org.springframework.integration.endpoint.DefaultEndpointRegistry;
import org.springframework.integration.endpoint.EndpointRegistry;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.MessagingGateway;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.SubscribableSource;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
@@ -77,7 +78,7 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
private final EndpointRegistry endpointRegistry = new DefaultEndpointRegistry();
private final Set<PollingDispatcher> pollingDispatchers = new CopyOnWriteArraySet<PollingDispatcher>();
private final Set<AbstractPoller> pollers = new CopyOnWriteArraySet<AbstractPoller>();
private volatile Schedule defaultPollerSchedule = new PollingSchedule(0);
@@ -273,17 +274,17 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
}
if (source instanceof SubscribableSource) {
((SubscribableSource) source).subscribe(endpoint);
if (source instanceof PollingDispatcher) {
PollingDispatcher poller = (PollingDispatcher) source;
this.pollingDispatchers.add(poller);
if (source instanceof AbstractPoller) {
AbstractPoller poller = (AbstractPoller) source;
this.pollers.add(poller);
this.taskScheduler.schedule(poller);
}
return;
}
else if (source instanceof PollableSource) {
PollingDispatcher poller = new PollingDispatcher((PollableSource<?>) source, this.defaultPollerSchedule);
else if (source instanceof PollableChannel) {
ChannelPoller poller = new ChannelPoller((PollableChannel) source, this.defaultPollerSchedule);
poller.subscribe(endpoint);
this.pollingDispatchers.add(poller);
this.pollers.add(poller);
this.taskScheduler.schedule(poller);
}
if (logger.isInfoEnabled()) {
@@ -306,10 +307,10 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A
public void deactivateEndpoint(MessageEndpoint endpoint) {
Assert.notNull(endpoint, "'endpoint' must not be null");
for (PollingDispatcher poller : this.pollingDispatchers) {
boolean removed = poller.unsubscribe(endpoint);
for (AbstractPoller poller : this.pollers) {
boolean removed = ((AbstractPoller) poller).unsubscribe(endpoint);
if (removed && this.logger.isInfoEnabled()) {
logger.info("removed endpoint '" + endpoint + "' from dispatcher '" + poller + "'");
logger.info("unsubscribed endpoint '" + endpoint + "' from poller '" + poller + "'");
}
}
if (endpoint instanceof Lifecycle) {

2
org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java
View File

@@ -43,7 +43,7 @@ public class DirectChannel extends AbstractMessageChannel implements Subscribabl
@Override
protected boolean doSend(Message<?> message, long timeout) {
return this.dispatcher.send(message);
return this.dispatcher.dispatch(message);
}
}

2
org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java
View File

@@ -58,7 +58,7 @@ public class PublishSubscribeChannel extends AbstractMessageChannel implements S
@Override
protected boolean doSend(Message<?> message, long timeout) {
return this.dispatcher.send(message);
return this.dispatcher.dispatch(message);
}
}

2
org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java
View File

@@ -89,7 +89,7 @@ public abstract class AbstractEndpointParser extends AbstractSingleBeanDefinitio
}
Element pollerElement = DomUtils.getChildElementByTagName(element, POLLER_ELEMENT);
if (pollerElement != null) {
String pollerBeanName = IntegrationNamespaceUtils.parsePoller(inputChannel, pollerElement, parserContext);
String pollerBeanName = IntegrationNamespaceUtils.parseChannelPoller(inputChannel, pollerElement, parserContext);
builder.addPropertyReference("source", pollerBeanName);
}
else {

4
org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java
View File

@@ -74,7 +74,7 @@ public class ChannelAdapterParser extends AbstractBeanDefinitionParser {
}
adapterBuilder = BeanDefinitionBuilder.genericBeanDefinition(InboundChannelAdapter.class);
if (pollerElement != null) {
String pollerBeanName = IntegrationNamespaceUtils.parsePoller(source, pollerElement, parserContext);
String pollerBeanName = IntegrationNamespaceUtils.parseSourcePoller(source, pollerElement, parserContext);
adapterBuilder.addPropertyReference("source", pollerBeanName);
}
else {
@@ -100,7 +100,7 @@ public class ChannelAdapterParser extends AbstractBeanDefinitionParser {
if (!StringUtils.hasText(channelName)) {
throw new ConfigurationException("outbound channel-adapter with a 'poller' requires a 'channel' to poll");
}
String pollerBeanName = IntegrationNamespaceUtils.parsePoller(channelName, pollerElement, parserContext);
String pollerBeanName = IntegrationNamespaceUtils.parseChannelPoller(channelName, pollerElement, parserContext);
adapterBuilder.addPropertyReference("source", pollerBeanName);
}
else if (StringUtils.hasText(channelName)) {

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

@@ -26,11 +26,14 @@ import org.springframework.beans.factory.xml.BeanDefinitionParserDelegate;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.core.Conventions;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.endpoint.ChannelPoller;
import org.springframework.integration.endpoint.SourcePoller;
import org.springframework.integration.scheduling.CronSchedule;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.springframework.util.xml.DomUtils;
/**
* Shared utility methods for integration namespace parsers.
@@ -135,15 +138,32 @@ public abstract class IntegrationNamespaceUtils {
}
/**
* Parse a "poller" element and return the bean name of the poller instance.
* Parse a "poller" element to create a ChannelPoller and return the bean name of the poller instance.
*
* @param channelBeanName the name of the PollableChannel bean
* @param element the "poller" element to parse
* @param parserContext the parserContext for registering a newly created bean definition
* @return the name of the ChannelPoller bean definition
*/
public static String parseChannelPoller(String channelBeanName, Element element, ParserContext parserContext) {
return parsePoller(channelBeanName, element, parserContext, true);
}
/**
* Parse a "poller" element to create a SourcePoller and return the bean name of the poller instance.
*
* @param sourceBeanName the name of the PollableSource bean
* @param element the "poller" element to parse
* @param parserContext the parserContext for registering a newly created bean definition
* @return the name of the poller bean definition
*/
public static String parsePoller(String sourceBeanName, Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(PollingDispatcherFactoryBean.class);
public static String parseSourcePoller(String sourceBeanName, Element element, ParserContext parserContext) {
return parsePoller(sourceBeanName, element, parserContext, false);
}
private static String parsePoller(String sourceBeanName, Element element, ParserContext parserContext, boolean isChannel) {
Class<?> beanClass = isChannel ? ChannelPoller.class : SourcePoller.class;
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(beanClass);
Schedule schedule = null;
if (!(StringUtils.hasText(element.getAttribute("period")) ^ StringUtils.hasText(element.getAttribute("cron")))) {
throw new ConfigurationException("A <poller> element must define either a period "
@@ -170,13 +190,15 @@ public abstract class IntegrationNamespaceUtils {
Element txElement = DomUtils.getChildElementByTagName(element, "transactional");
if (txElement != null) {
builder.addPropertyReference("transactionManager", txElement.getAttribute("transaction-manager"));
builder.addPropertyValue("propagationBehaviorName", txElement.getAttribute("propagation"));
builder.addPropertyValue("isolationLevelName", txElement.getAttribute("isolation"));
builder.addPropertyValue("propagationBehaviorName",
DefaultTransactionDefinition.PREFIX_PROPAGATION + txElement.getAttribute("propagation"));
builder.addPropertyValue("isolationLevelName",
DefaultTransactionDefinition.PREFIX_ISOLATION + txElement.getAttribute("isolation"));
builder.addPropertyValue("transactionTimeout", txElement.getAttribute("timeout"));
builder.addPropertyValue("transactionReadOnly", txElement.getAttribute("read-only"));
}
builder.addPropertyReference("source", sourceBeanName);
builder.addPropertyValue("schedule", schedule);
builder.addConstructorArgReference(sourceBeanName);
builder.addConstructorArgValue(schedule);
setValueIfAttributeDefined(builder, element, "receive-timeout");
setValueIfAttributeDefined(builder, element, "send-timeout");
setValueIfAttributeDefined(builder, element, "max-messages-per-poll");

174
org.springframework.integration/src/main/java/org/springframework/integration/config/PollingDispatcherFactoryBean.java
View File

@@ -1,174 +0,0 @@
/*
* Copyright 2002-2008 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.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.dispatcher.SimpleDispatcher;
import org.springframework.integration.message.AsyncMessageExchangeTemplate;
import org.springframework.integration.message.MessageExchangeTemplate;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.support.DefaultTransactionDefinition;
/**
* @author Mark Fisher
*/
public class PollingDispatcherFactoryBean implements FactoryBean, InitializingBean {
private volatile PollingDispatcher poller;
private volatile MessageSource<?> source;
private volatile Schedule schedule;
private volatile long receiveTimeout = -1;
private volatile long sendTimeout = -1;
private volatile int maxMessagesPerPoll = -1;
private volatile TaskExecutor taskExecutor;
private volatile PlatformTransactionManager transactionManager;
private volatile String propagationBehaviorName;
private volatile String isolationLevelName;
private volatile int transactionTimeout;
private volatile boolean transactionReadOnly;
private volatile boolean validated;
private volatile boolean initialized;
private final Object initializationMonitor = new Object();
public void setSource(MessageSource<?> source) {
this.source = source;
}
public void setSchedule(Schedule schedule) {
this.schedule = schedule;
}
public void setReceiveTimeout(long receiveTimeout) {
this.receiveTimeout = receiveTimeout;
}
public void setSendTimeout(long sendTimeout) {
this.sendTimeout = sendTimeout;
}
public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
this.maxMessagesPerPoll = maxMessagesPerPoll;
}
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
public void setPropagationBehaviorName(String propagationBehaviorName) {
this.propagationBehaviorName = propagationBehaviorName;
}
public void setIsolationLevelName(String isolationLevelName) {
this.isolationLevelName = isolationLevelName;
}
public void setTransactionTimeout(int transactionTimeout) {
this.transactionTimeout = transactionTimeout;
}
public void setTransactionReadOnly(boolean transactionReadOnly) {
this.transactionReadOnly = transactionReadOnly;
}
public void afterPropertiesSet() {
synchronized (this.initializationMonitor) {
if (this.source == null) {
throw new ConfigurationException("source is required");
}
if (!(this.source instanceof PollableSource)) {
throw new BeanCreationException("Poller requires a PollableSource, but actual type of '"
+ this.source + "' is [" + this.source.getClass() + "]");
}
this.validated = true;
}
}
public Object getObject() throws Exception {
if (!this.initialized) {
this.initPoller();
}
return this.poller;
}
public Class<?> getObjectType() {
return PollingDispatcher.class;
}
public boolean isSingleton() {
return true;
}
private void initPoller() {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;
}
if (!this.validated) {
this.afterPropertiesSet();
}
if (this.schedule == null) {
this.schedule = new PollingSchedule(0);
}
MessageExchangeTemplate template = this.createMessageExchangeTemplate();
this.poller = new PollingDispatcher((PollableSource<?>) this.source, this.schedule, new SimpleDispatcher(), template);
this.poller.setMaxMessagesPerPoll(this.maxMessagesPerPoll);
this.initialized = true;
}
}
private MessageExchangeTemplate createMessageExchangeTemplate() {
MessageExchangeTemplate template = (this.taskExecutor != null) ?
new AsyncMessageExchangeTemplate(this.taskExecutor) : new MessageExchangeTemplate();
template.setTransactionManager(this.transactionManager);
template.setPropagationBehaviorName(DefaultTransactionDefinition.PREFIX_PROPAGATION + this.propagationBehaviorName);
template.setIsolationLevelName(DefaultTransactionDefinition.PREFIX_ISOLATION + this.isolationLevelName);
template.setTransactionTimeout(this.transactionTimeout);
template.setTransactionReadOnly(this.transactionReadOnly);
template.setReceiveTimeout(this.receiveTimeout);
template.setSendTimeout(this.sendTimeout);
return template;
}
}

4
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java
View File

@@ -26,9 +26,9 @@ import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.AbstractInOutEndpoint;
import org.springframework.integration.endpoint.ChannelPoller;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
@@ -93,7 +93,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
schedule.setInitialDelay(pollerAnnotation.initialDelay());
schedule.setFixedRate(pollerAnnotation.fixedRate());
schedule.setTimeUnit(pollerAnnotation.timeUnit());
PollingDispatcher poller = new PollingDispatcher((PollableChannel) inputChannel, schedule);
ChannelPoller poller = new ChannelPoller((PollableChannel) inputChannel, schedule);
poller.setMaxMessagesPerPoll(pollerAnnotation.maxMessagesPerPoll());
endpoint.setSource(poller);
}

34
org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/ChannelAdapterAnnotationPostProcessor.java
View File

@@ -26,14 +26,15 @@ import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.endpoint.ChannelPoller;
import org.springframework.integration.endpoint.InboundChannelAdapter;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.endpoint.OutboundChannelAdapter;
import org.springframework.integration.endpoint.SourcePoller;
import org.springframework.integration.handler.MethodInvokingTarget;
import org.springframework.integration.message.MethodInvokingSource;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.Assert;
/**
@@ -88,7 +89,15 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
throw new ConfigurationException("The @Poller annotation is required (at method-level) "
+ "when using the @ChannelAdapter annotation with a no-arg method.");
}
PollingDispatcher poller = this.createPoller(source, pollerAnnotation);
Schedule schedule = this.createSchedule(pollerAnnotation);
SourcePoller poller = new SourcePoller(source, schedule);
int maxMessagesPerPoll = pollerAnnotation.maxMessagesPerPoll();
if (maxMessagesPerPoll == -1) {
// the default is 1 since a MethodInvokingSource might return a non-null value
// every time it is invoked, thus producing an infinite number of messages per poll
maxMessagesPerPoll = 1;
}
poller.setMaxMessagesPerPoll(maxMessagesPerPoll);
InboundChannelAdapter adapter = new InboundChannelAdapter();
adapter.setSource(poller);
adapter.setChannel(channel);
@@ -99,9 +108,10 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
private OutboundChannelAdapter createOutboundChannelAdapter(MethodInvokingTarget target, MessageChannel channel, Poller pollerAnnotation) {
OutboundChannelAdapter adapter = new OutboundChannelAdapter(target);
if (channel instanceof PollableChannel) {
PollingDispatcher poller = (pollerAnnotation != null)
? this.createPoller((PollableChannel) channel, pollerAnnotation)
: new PollingDispatcher((PollableSource<?>) channel, new PollingSchedule(0));
Schedule schedule = (pollerAnnotation != null)
? this.createSchedule(pollerAnnotation)
: new PollingSchedule(0);
ChannelPoller poller = new ChannelPoller((PollableChannel) channel, schedule);
adapter.setSource(poller);
}
else {
@@ -111,20 +121,12 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
return adapter;
}
private PollingDispatcher createPoller(PollableSource<?> source, Poller pollerAnnotation) {
private Schedule createSchedule(Poller pollerAnnotation) {
PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period());
schedule.setInitialDelay(pollerAnnotation.initialDelay());
schedule.setFixedRate(pollerAnnotation.fixedRate());
schedule.setTimeUnit(pollerAnnotation.timeUnit());
PollingDispatcher poller = new PollingDispatcher((PollableSource<?>) source, schedule);
int maxMessagesPerPoll = pollerAnnotation.maxMessagesPerPoll();
if (maxMessagesPerPoll == -1) {
// the default is 1 since a MethodInvokingSource might return a non-null value
// every time it is invoked, thus producing an infinite number of messages per poll
maxMessagesPerPoll = 1;
}
poller.setMaxMessagesPerPoll(maxMessagesPerPoll);
return poller;
return schedule;
}
private boolean hasReturnValue(Method method) {

6
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java
View File

@@ -40,12 +40,6 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
private volatile TaskExecutor taskExecutor;
// TODO: dispatcher should not implement channel, need to move TX support into the poller
// so that the messageExchangeTemplate is not required for sending to a dispatcher
public String getName() {
return "dispatcher";
}
public boolean subscribe(MessageEndpoint endpoint) {
return this.endpoints.add(endpoint);
}

2
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java
View File

@@ -43,7 +43,7 @@ public class BroadcastingDispatcher extends AbstractDispatcher {
this.applySequence = applySequence;
}
public boolean send(Message<?> message) {
public boolean dispatch(Message<?> message) {
int sequenceNumber = 1;
int sequenceSize = this.endpoints.size();
for (final MessageEndpoint endpoint : this.endpoints) {

10
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java
View File

@@ -16,8 +16,6 @@
package org.springframework.integration.dispatcher;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.SubscribableSource;
@@ -26,12 +24,8 @@ import org.springframework.integration.message.SubscribableSource;
*
* @author Mark Fisher
*/
public interface MessageDispatcher extends MessageChannel, SubscribableSource {
public interface MessageDispatcher extends SubscribableSource {
boolean send(Message<?> message);
boolean subscribe(MessageEndpoint endpoint);
boolean unsubscribe(MessageEndpoint endpoint);
boolean dispatch(Message<?> message);
}

141
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java
View File

@@ -1,141 +0,0 @@
/*
* Copyright 2002-2008 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.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.message.BlockingSource;
import org.springframework.integration.message.MessageExchangeTemplate;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.SubscribableSource;
import org.springframework.integration.scheduling.SchedulableTask;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.Assert;
/**
* @author Mark Fisher
*/
public class PollingDispatcher implements SchedulableTask, SubscribableSource {
public final static int MAX_MESSAGES_UNBOUNDED = -1;
public final static long DEFAULT_RECEIVE_TIMEOUT = 1000;
private final Log logger = LogFactory.getLog(this.getClass());
private final PollableSource<?> source;
private final MessageDispatcher dispatcher;
private final Schedule schedule;
private final MessageExchangeTemplate messageExchangeTemplate;
private volatile int maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED;
/**
* Create a PollingDispatcher for the provided {@link PollableSource}.
* It can be scheduled according to the specified {@link Schedule}.
*/
public PollingDispatcher(PollableSource<?> source, Schedule schedule) {
this(source, schedule, null, null);
}
public PollingDispatcher(PollableSource<?> source, Schedule schedule, MessageDispatcher dispatcher) {
this(source, schedule, dispatcher, null);
}
public PollingDispatcher(PollableSource<?> source, Schedule schedule, MessageDispatcher dispatcher, MessageExchangeTemplate messageExchangeTemplate) {
Assert.notNull(source, "source must not be null");
this.source = source;
this.schedule = schedule;
this.dispatcher = (dispatcher != null)
? dispatcher : new SimpleDispatcher();
this.messageExchangeTemplate = (messageExchangeTemplate != null)
? messageExchangeTemplate : createDefaultTemplate();
}
/**
* Specify the timeout to use when receiving from the source (in milliseconds).
* Note that this value will only be applicable if the source is an instance
* of {@link BlockingSource}.
* <p/>
* A negative value indicates that receive calls should block indefinitely,
* and that is the default behavior.
*/
public void setReceiveTimeout(long receiveTimeout) {
this.messageExchangeTemplate.setReceiveTimeout(receiveTimeout);
}
/**
* Set the maximum number of messages to receive for each poll.
* A non-positive value indicates that polling should repeat as long
* as non-null messages are being received and successfully sent.
*
* <p>The default is unbounded.
*
* @see #MAX_MESSAGES_UNBOUNDED
*/
public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
this.maxMessagesPerPoll = maxMessagesPerPoll;
}
public boolean subscribe(MessageEndpoint endpoint) {
return this.dispatcher.subscribe(endpoint);
}
public boolean unsubscribe(MessageEndpoint endpoint) {
return this.dispatcher.unsubscribe(endpoint);
}
public Schedule getSchedule() {
return this.schedule;
}
public void run() {
int count = 0;
while (this.maxMessagesPerPoll < 0 || count < this.maxMessagesPerPoll) {
if (!this.messageExchangeTemplate.receiveAndForward(this.source, this.dispatcher)) {
break;
}
count++;
}
if (this.logger.isTraceEnabled()) {
this.logger.trace("poller for source '" + this.source + "' sent " + count
+ " messages to target '" + this.dispatcher + "'");
}
return;
}
public String toString() {
return this.getClass().getSimpleName() + " [source = " + this.source
+ ", dispatcher = [" + this.dispatcher + "]";
}
private MessageExchangeTemplate createDefaultTemplate() {
MessageExchangeTemplate template = new MessageExchangeTemplate();
template.setReceiveTimeout(DEFAULT_RECEIVE_TIMEOUT);
template.setSendTimeout(-1);
return template;
}
}

2
org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java
View File

@@ -35,7 +35,7 @@ import org.springframework.integration.message.MessageRejectedException;
*/
public class SimpleDispatcher extends AbstractDispatcher {
public boolean send(Message<?> message) {
public boolean dispatch(Message<?> message) {
if (this.endpoints.size() == 0) {
throw new MessageDeliveryException(message, "Dispatcher has no subscribers.");
}

174
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPoller.java Normal file
View File

@@ -0,0 +1,174 @@
/*
* Copyright 2002-2008 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.endpoint;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.message.SubscribableSource;
import org.springframework.integration.scheduling.SchedulableTask;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.Assert;
/**
* @author Mark Fisher
*/
public abstract class AbstractPoller implements SubscribableSource, SchedulableTask, InitializingBean {
public static final int MAX_MESSAGES_UNBOUNDED = -1;
private final Schedule schedule;
private volatile long maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED;
private volatile TaskExecutor taskExecutor;
private volatile PlatformTransactionManager transactionManager;
private volatile TransactionTemplate transactionTemplate;
private volatile String propagationBehaviorName = "PROPAGATION_REQUIRED";
private volatile String isolationLevelName = "ISOLATION_DEFAULT";
private volatile int transactionTimeout = -1;
private volatile boolean readOnly = false;
private volatile boolean initialized;
private final Object initializationMonitor = new Object();
public AbstractPoller(Schedule schedule) {
Assert.notNull(schedule, "schedule must not be null");
this.schedule = schedule;
}
public Schedule getSchedule() {
return this.schedule;
}
/**
* Set the maximum number of messages to receive for each poll.
* A non-positive value indicates that polling should repeat as long
* as non-null messages are being received and successfully sent.
*
* <p>The default is unbounded.
*
* @see #MAX_MESSAGES_UNBOUNDED
*/
public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
this.maxMessagesPerPoll = maxMessagesPerPoll;
}
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
/**
* Specify a transaction manager to use for all exchange operations.
* If none is provided, then the operations will occur without any
* transactional behavior (i.e. there is no default transaction manager).
*/
public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
public void setPropagationBehaviorName(String propagationBehaviorName) {
this.propagationBehaviorName = propagationBehaviorName;
}
public void setIsolationLevelName(String isolationLevelName) {
this.isolationLevelName = isolationLevelName;
}
public void setTransactionTimeout(int transactionTimeout) {
this.transactionTimeout = transactionTimeout;
}
public void setTransactionReadOnly(boolean readOnly) {
this.readOnly = readOnly;
}
private TransactionTemplate getTransactionTemplate() {
if (!this.initialized) {
this.afterPropertiesSet();
}
return this.transactionTemplate;
}
public void afterPropertiesSet() {
synchronized (this.initializationMonitor) {
if (this.initialized) {
return;
}
if (this.transactionManager != null) {
TransactionTemplate template = new TransactionTemplate(this.transactionManager);
template.setPropagationBehaviorName(this.propagationBehaviorName);
template.setIsolationLevelName(this.isolationLevelName);
template.setTimeout(this.transactionTimeout);
template.setReadOnly(this.readOnly);
this.transactionTemplate = template;
}
this.initialized = true;
}
}
public void run() {
if (this.taskExecutor != null) {
this.taskExecutor.execute(new Runnable() {
public void run() {
poll();
}
});
}
else {
poll();
}
}
private void poll() {
int count = 0;
while (this.maxMessagesPerPoll < 0 || count < this.maxMessagesPerPoll) {
if (!this.pollWithinTransaction()) {
break;
}
count++;
}
}
private boolean pollWithinTransaction() {
TransactionTemplate txTemplate = this.getTransactionTemplate();
if (txTemplate != null) {
return (Boolean) txTemplate.execute(new TransactionCallback() {
public Object doInTransaction(TransactionStatus status) {
return doPoll();
}
});
}
return doPoll();
}
protected abstract boolean doPoll();
}

72
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ChannelPoller.java Normal file
View File

@@ -0,0 +1,72 @@
/*
* Copyright 2002-2008 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.endpoint;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.dispatcher.SimpleDispatcher;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.SubscribableSource;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.Assert;
/**
* @author Mark Fisher
*/
public class ChannelPoller extends AbstractPoller implements SubscribableSource {
private final PollableChannel channel;
private volatile long receiveTimeout = 1000;
private final SimpleDispatcher dispatcher = new SimpleDispatcher();
public ChannelPoller(PollableChannel channel, Schedule schedule) {
super(schedule);
Assert.notNull(channel, "channel must not be null");
this.channel = channel;
}
/**
* Specify the timeout to use when receiving from the channel (in milliseconds).
* A negative value indicates that receive calls should block indefinitely.
* The default value is 1000 (1 second).
*/
public void setReceiveTimeout(long receiveTimeout) {
this.receiveTimeout = receiveTimeout;
}
public boolean subscribe(MessageEndpoint endpoint) {
return this.dispatcher.subscribe(endpoint);
}
public boolean unsubscribe(MessageEndpoint endpoint) {
return this.dispatcher.unsubscribe(endpoint);
}
@Override
protected boolean doPoll() {
Message<?> message = (this.receiveTimeout >= 0)
? this.channel.receive(this.receiveTimeout)
: this.channel.receive();
if (message == null) {
return false;
}
return this.dispatcher.dispatch(message);
}
}

76
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePoller.java Normal file
View File

@@ -0,0 +1,76 @@
/*
* Copyright 2002-2008 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.endpoint;
import org.springframework.integration.dispatcher.SimpleDispatcher;
import org.springframework.integration.message.BlockingSource;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.SubscribableSource;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.util.Assert;
/**
* @author Mark Fisher
*/
public class SourcePoller extends AbstractPoller implements SubscribableSource {
private final PollableSource<?> source;
private final SimpleDispatcher dispatcher = new SimpleDispatcher();
private volatile long receiveTimeout = 1000;
public SourcePoller(PollableSource<?> source, Schedule schedule) {
super(schedule);
Assert.notNull(source, "source must not be null");
this.source = source;
}
/**
* Specify the timeout to use when receiving from the source (in milliseconds).
* This value will only apply if the source is a {@link BlockingSource}.
* <p>
* A negative value indicates that receive calls should block indefinitely.
* The default value is 1000 (1 second).
*/
public void setReceiveTimeout(long receiveTimeout) {
this.receiveTimeout = receiveTimeout;
}
public boolean subscribe(MessageEndpoint endpoint) {
return this.dispatcher.subscribe(endpoint);
}
public boolean unsubscribe(MessageEndpoint endpoint) {
return this.dispatcher.unsubscribe(endpoint);
}
@Override
protected boolean doPoll() {
Message<?> message = (this.receiveTimeout >= 0 && this.source instanceof BlockingSource)
? ((BlockingSource<?>) this.source).receive(this.receiveTimeout)
: this.source.receive();
if (message == null) {
return false;
}
return this.dispatcher.dispatch(message);
}
}

14
org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java
View File

@@ -34,13 +34,14 @@ import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.AbstractInOutEndpoint;
import org.springframework.integration.endpoint.InboundChannelAdapter;
import org.springframework.integration.endpoint.SourcePoller;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.scheduling.PollingSchedule;
/**
* @author Mark Fisher
@@ -189,20 +190,23 @@ public class DefaultMessageBusTests {
@Test
public void testErrorChannelWithFailedDispatch() throws InterruptedException {
MessageBus bus = new DefaultMessageBus();
QueueChannel errorChannel = new QueueChannel();
errorChannel.setBeanName("errorChannel");
bus.registerChannel(errorChannel);
CountDownLatch latch = new CountDownLatch(1);
InboundChannelAdapter channelAdapter = new InboundChannelAdapter();
channelAdapter.setSource(new FailingSource(latch));
SourcePoller poller = new SourcePoller(new FailingSource(latch), new PollingSchedule(1000));
channelAdapter.setSource(poller);
channelAdapter.setBeanName("testChannel");
bus.registerEndpoint(channelAdapter);
bus.start();
latch.await(2000, TimeUnit.MILLISECONDS);
Message<?> message = ((PollableChannel) bus.getErrorChannel()).receive(5000);
Message<?> message = errorChannel.receive(5000);
bus.stop();
assertNotNull("message should not be null", message);
assertTrue(message instanceof ErrorMessage);
Throwable exception = ((ErrorMessage) message).getPayload();
assertTrue(exception instanceof MessagingException);
assertEquals("intentional test failure", exception.getCause().getMessage());
assertEquals("intentional test failure", exception.getMessage());
}
@Test(expected = BeanCreationException.class)

28
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java
View File

@@ -77,7 +77,7 @@ public class BroadcastingDispatcherTests {
dispatcher.subscribe(targetMock1);
expect(targetMock1.send(messageMock)).andReturn(true);
replay(globalMocks);
dispatcher.send(messageMock);
dispatcher.dispatch(messageMock);
verify(globalMocks);
}
@@ -86,7 +86,7 @@ public class BroadcastingDispatcherTests {
dispatcher.subscribe(targetMock1);
expect(targetMock1.send(messageMock)).andReturn(true);
replay(globalMocks);
dispatcher.send(messageMock);
dispatcher.dispatch(messageMock);
verify(globalMocks);
}
@@ -100,7 +100,7 @@ public class BroadcastingDispatcherTests {
expect(targetMock2.send(messageMock)).andReturn(true);
expect(targetMock3.send(messageMock)).andReturn(true);
replay(globalMocks);
dispatcher.send(messageMock);
dispatcher.dispatch(messageMock);
verify(globalMocks);
}
@@ -113,7 +113,7 @@ public class BroadcastingDispatcherTests {
expect(targetMock2.send(messageMock)).andReturn(true);
expect(targetMock3.send(messageMock)).andReturn(true);
replay(globalMocks);
dispatcher.send(messageMock);
dispatcher.dispatch(messageMock);
verify(globalMocks);
}
@@ -127,7 +127,7 @@ public class BroadcastingDispatcherTests {
expect(targetMock2.send(messageMock)).andReturn(true);
expect(targetMock3.send(messageMock)).andReturn(true);
replay(globalMocks);
dispatcher.send(messageMock);
dispatcher.dispatch(messageMock);
verify(globalMocks);
}
@@ -141,7 +141,7 @@ public class BroadcastingDispatcherTests {
expect(targetMock1.send(messageMock)).andReturn(true);
expect(targetMock3.send(messageMock)).andReturn(true);
replay(globalMocks);
dispatcher.send(messageMock);
dispatcher.dispatch(messageMock);
verify(globalMocks);
}
@@ -155,7 +155,7 @@ public class BroadcastingDispatcherTests {
expect(targetMock1.send(messageMock)).andReturn(true);
expect(targetMock2.send(messageMock)).andReturn(true);
replay(globalMocks);
dispatcher.send(messageMock);
dispatcher.dispatch(messageMock);
verify(globalMocks);
}
@@ -167,7 +167,7 @@ public class BroadcastingDispatcherTests {
dispatcher.subscribe(targetMock3);
partialFailingExecutorMock(false, false, false);
replay(globalMocks);
dispatcher.send(messageMock);
dispatcher.dispatch(messageMock);
verify(globalMocks);
}
@@ -178,7 +178,7 @@ public class BroadcastingDispatcherTests {
dispatcher.subscribe(targetMock1);
expect(targetMock1.send(messageMock)).andReturn(true);
replay(globalMocks);
dispatcher.send(messageMock);
dispatcher.dispatch(messageMock);
verify(globalMocks);
}
@@ -191,7 +191,7 @@ public class BroadcastingDispatcherTests {
expect(targetMock1.send(messageMock)).andReturn(true);
expect(targetMock3.send(messageMock)).andReturn(true);
replay(globalMocks);
dispatcher.send(messageMock);
dispatcher.dispatch(messageMock);
verify(globalMocks);
}
@@ -204,9 +204,9 @@ public class BroadcastingDispatcherTests {
expect(targetMock2.send(messageMock)).andReturn(true);
expect(targetMock3.send(messageMock)).andReturn(true).times(2);
replay(globalMocks);
dispatcher.send(messageMock);
dispatcher.dispatch(messageMock);
dispatcher.unsubscribe(targetMock2);
dispatcher.send(messageMock);
dispatcher.dispatch(messageMock);
verify(globalMocks);
}
@@ -218,7 +218,7 @@ public class BroadcastingDispatcherTests {
MessageEndpoint target2 = new MessageStoringTestEndpoint(messages);
dispatcher.subscribe(target1);
dispatcher.subscribe(target2);
dispatcher.send(new StringMessage("test"));
dispatcher.dispatch(new StringMessage("test"));
assertEquals(2, messages.size());
assertEquals(0, (int) messages.get(0).getHeaders().getSequenceNumber());
assertEquals(0, (int) messages.get(0).getHeaders().getSequenceSize());
@@ -237,7 +237,7 @@ public class BroadcastingDispatcherTests {
dispatcher.subscribe(target1);
dispatcher.subscribe(target2);
dispatcher.subscribe(target3);
dispatcher.send(new StringMessage("test"));
dispatcher.dispatch(new StringMessage("test"));
assertEquals(3, messages.size());
assertEquals(1, (int) messages.get(0).getHeaders().getSequenceNumber());
assertEquals(3, (int) messages.get(0).getHeaders().getSequenceSize());

128
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTests.java
View File

@@ -1,128 +0,0 @@
/*
* Copyright 2002-2008 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 static org.easymock.EasyMock.createMock;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.reset;
import static org.easymock.EasyMock.verify;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.dispatcher.MessageDispatcher;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.message.BlockingSource;
import org.springframework.integration.message.Message;
import org.springframework.integration.scheduling.Schedule;
/**
* @author Iwein Fuld
*/
@SuppressWarnings("unchecked")
public class PollingDispatcherTests {
private PollingDispatcher pollingDispatcher;
private Schedule scheduleMock = createMock(Schedule.class);
private MessageDispatcher dispatcherMock = createMock(MessageDispatcher.class);
private BlockingSource sourceMock = createMock(BlockingSource.class);
private Message messageMock = createMock(Message.class);
private Object[] globalMocks = new Object[] { scheduleMock, dispatcherMock, sourceMock, messageMock };
@Before
public void init() {
pollingDispatcher = new PollingDispatcher(sourceMock, scheduleMock, dispatcherMock);
pollingDispatcher.setReceiveTimeout(-1);
reset(globalMocks);
}
@Test
public void singleMessage() {
expect(sourceMock.receive()).andReturn(messageMock);
expect(dispatcherMock.send(messageMock)).andReturn(true);
replay(globalMocks);
pollingDispatcher.setMaxMessagesPerPoll(1);
pollingDispatcher.run();
verify(globalMocks);
}
@Test
public void multipleMessages() {
expect(sourceMock.receive()).andReturn(messageMock).times(5);
expect(dispatcherMock.send(messageMock)).andReturn(true).times(5);
replay(globalMocks);
pollingDispatcher.setMaxMessagesPerPoll(5);
pollingDispatcher.run();
verify(globalMocks);
}
@Test
public void multipleMessages_underrun() {
expect(sourceMock.receive()).andReturn(messageMock).times(5);
expect(sourceMock.receive()).andReturn(null);
expect(dispatcherMock.send(messageMock)).andReturn(true).times(5);
replay(globalMocks);
pollingDispatcher.setMaxMessagesPerPoll(6);
pollingDispatcher.run();
verify(globalMocks);
}
@Test
public void droppedMessage() {
expect(sourceMock.receive()).andReturn(messageMock);
expect(dispatcherMock.send(messageMock)).andReturn(false);
replay(globalMocks);
pollingDispatcher.run();
verify(globalMocks);
}
@Test
public void droppedMessage_onePerPoll() {
expect(sourceMock.receive()).andReturn(messageMock).times(1);
expect(dispatcherMock.send(messageMock)).andReturn(false).anyTimes();
replay(globalMocks);
pollingDispatcher.setMaxMessagesPerPoll(10);
pollingDispatcher.run();
verify(globalMocks);
}
@Test
public void blockingSourceTimedOut() {
pollingDispatcher = new PollingDispatcher(sourceMock, scheduleMock, dispatcherMock);
// we don't need to await the timeout, returning null suffices
expect(sourceMock.receive(1)).andReturn(null);
replay(globalMocks);
pollingDispatcher.setReceiveTimeout(1);
pollingDispatcher.run();
verify(globalMocks);
}
@Test
public void blockingSourceNotTimedOut() {
pollingDispatcher = new PollingDispatcher(sourceMock, scheduleMock, dispatcherMock);
expect(sourceMock.receive(1)).andReturn(messageMock);
expect(dispatcherMock.send(messageMock)).andReturn(false);
replay(globalMocks);
pollingDispatcher.setReceiveTimeout(1);
pollingDispatcher.run();
verify(globalMocks);
}
}

28
org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java
View File

@@ -47,7 +47,7 @@ public class SimpleDispatcherTests {
SimpleDispatcher dispatcher = new SimpleDispatcher();
final CountDownLatch latch = new CountDownLatch(1);
dispatcher.subscribe(createEndpoint(TestHandlers.countDownHandler(latch)));
dispatcher.send(new StringMessage("test"));
dispatcher.dispatch(new StringMessage("test"));
latch.await(500, TimeUnit.MILLISECONDS);
assertEquals(0, latch.getCount());
}
@@ -60,7 +60,7 @@ public class SimpleDispatcherTests {
final AtomicInteger counter2 = new AtomicInteger();
dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter1, latch)));
dispatcher.subscribe(createEndpoint(TestHandlers.countingCountDownHandler(counter2, latch)));
dispatcher.send(new StringMessage("test"));
dispatcher.dispatch(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());
@@ -73,7 +73,7 @@ public class SimpleDispatcherTests {
MessageEndpoint target = new CountingTestEndpoint(counter, false);
dispatcher.subscribe(target);
dispatcher.subscribe(target);
dispatcher.send(new StringMessage("test"));
dispatcher.dispatch(new StringMessage("test"));
assertEquals("target should not have duplicate subscriptions", 1, counter.get());
}
@@ -88,7 +88,7 @@ public class SimpleDispatcherTests {
dispatcher.subscribe(target2);
dispatcher.subscribe(target3);
dispatcher.unsubscribe(target2);
dispatcher.send(new StringMessage("test"));
dispatcher.dispatch(new StringMessage("test"));
assertEquals(2, counter.get());
}
@@ -102,13 +102,13 @@ public class SimpleDispatcherTests {
dispatcher.subscribe(target1);
dispatcher.subscribe(target2);
dispatcher.subscribe(target3);
dispatcher.send(new StringMessage("test1"));
dispatcher.dispatch(new StringMessage("test1"));
assertEquals(3, counter.get());
dispatcher.unsubscribe(target2);
dispatcher.send(new StringMessage("test2"));
dispatcher.dispatch(new StringMessage("test2"));
assertEquals(5, counter.get());
dispatcher.unsubscribe(target1);
dispatcher.send(new StringMessage("test3"));
dispatcher.dispatch(new StringMessage("test3"));
assertEquals(6, counter.get());
}
@@ -118,10 +118,10 @@ public class SimpleDispatcherTests {
final AtomicInteger counter = new AtomicInteger();
MessageEndpoint target = new CountingTestEndpoint(counter, false);
dispatcher.subscribe(target);
dispatcher.send(new StringMessage("test1"));
dispatcher.dispatch(new StringMessage("test1"));
assertEquals(1, counter.get());
dispatcher.unsubscribe(target);
dispatcher.send(new StringMessage("test2"));
dispatcher.dispatch(new StringMessage("test2"));
}
@Test
@@ -141,7 +141,7 @@ public class SimpleDispatcherTests {
dispatcher.subscribe(endpoint1);
dispatcher.subscribe(endpoint2);
dispatcher.subscribe(endpoint3);
dispatcher.send(new StringMessage("test"));
dispatcher.dispatch(new StringMessage("test"));
assertEquals(0, latch.getCount());
assertEquals("selectors should have been invoked one time each", 3, selectorCounter.get());
assertEquals("handler with rejecting selector should not have received the message", 0, counter1.get());
@@ -168,7 +168,7 @@ public class SimpleDispatcherTests {
dispatcher.subscribe(endpoint3);
boolean exceptionThrown = false;
try {
dispatcher.send(new StringMessage("test"));
dispatcher.dispatch(new StringMessage("test"));
}
catch (MessageRejectedException e) {
exceptionThrown = true;
@@ -190,7 +190,7 @@ public class SimpleDispatcherTests {
dispatcher.subscribe(target1);
dispatcher.subscribe(target2);
dispatcher.subscribe(target3);
assertTrue(dispatcher.send(new StringMessage("test")));
assertTrue(dispatcher.dispatch(new StringMessage("test")));
assertEquals("only the first target should have been invoked", 1, counter.get());
}
@@ -204,7 +204,7 @@ public class SimpleDispatcherTests {
dispatcher.subscribe(target1);
dispatcher.subscribe(target2);
dispatcher.subscribe(target3);
assertTrue(dispatcher.send(new StringMessage("test")));
assertTrue(dispatcher.dispatch(new StringMessage("test")));
assertEquals("first two targets should have been invoked", 2, counter.get());
}
@@ -218,7 +218,7 @@ public class SimpleDispatcherTests {
dispatcher.subscribe(target1);
dispatcher.subscribe(target2);
dispatcher.subscribe(target3);
assertFalse(dispatcher.send(new StringMessage("test")));
assertFalse(dispatcher.dispatch(new StringMessage("test")));
assertEquals("each target should have been invoked", 3, counter.get());
}

131
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ChannelPollerTests.java Normal file
View File

@@ -0,0 +1,131 @@
/*
* Copyright 2002-2008 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.endpoint;
import static org.easymock.EasyMock.createMock;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.replay;
import static org.easymock.EasyMock.reset;
import static org.easymock.EasyMock.verify;
import org.junit.Before;
import org.junit.Test;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.endpoint.ChannelPoller;
import org.springframework.integration.endpoint.MessageEndpoint;
import org.springframework.integration.message.Message;
import org.springframework.integration.scheduling.Schedule;
/**
* @author Iwein Fuld
*/
@SuppressWarnings("unchecked")
public class ChannelPollerTests {
private ChannelPoller poller;
private Schedule scheduleMock = createMock(Schedule.class);
private PollableChannel channelMock = createMock(PollableChannel.class);
private MessageEndpoint endpointMock = createMock(MessageEndpoint.class);
private Message messageMock = createMock(Message.class);
private Object[] globalMocks = new Object[] { scheduleMock, channelMock, endpointMock, messageMock };
@Before
public void init() {
poller = new ChannelPoller(channelMock, scheduleMock);
poller.subscribe(endpointMock);
poller.setReceiveTimeout(-1);
reset(globalMocks);
}
@Test
public void singleMessage() {
expect(channelMock.receive()).andReturn(messageMock);
expect(endpointMock.send(messageMock)).andReturn(true);
replay(globalMocks);
poller.setMaxMessagesPerPoll(1);
poller.run();
verify(globalMocks);
}
@Test
public void multipleMessages() {
expect(channelMock.receive()).andReturn(messageMock).times(5);
expect(endpointMock.send(messageMock)).andReturn(true).times(5);
replay(globalMocks);
poller.setMaxMessagesPerPoll(5);
poller.run();
verify(globalMocks);
}
@Test
public void multipleMessages_underrun() {
expect(channelMock.receive()).andReturn(messageMock).times(5);
expect(channelMock.receive()).andReturn(null);
expect(endpointMock.send(messageMock)).andReturn(true).times(5);
replay(globalMocks);
poller.setMaxMessagesPerPoll(6);
poller.run();
verify(globalMocks);
}
@Test
public void droppedMessage() {
expect(channelMock.receive()).andReturn(messageMock);
expect(endpointMock.send(messageMock)).andReturn(false);
replay(globalMocks);
poller.run();
verify(globalMocks);
}
@Test
public void droppedMessage_onePerPoll() {
expect(channelMock.receive()).andReturn(messageMock).times(1);
expect(endpointMock.send(messageMock)).andReturn(false).anyTimes();
replay(globalMocks);
poller.setMaxMessagesPerPoll(10);
poller.run();
verify(globalMocks);
}
@Test
public void blockingSourceTimedOut() {
poller = new ChannelPoller(channelMock, scheduleMock);
poller.subscribe(endpointMock);
// we don't need to await the timeout, returning null suffices
expect(channelMock.receive(1)).andReturn(null);
replay(globalMocks);
poller.setReceiveTimeout(1);
poller.run();
verify(globalMocks);
}
@Test
public void blockingSourceNotTimedOut() {
poller = new ChannelPoller(channelMock, scheduleMock);
poller.subscribe(endpointMock);
expect(channelMock.receive(1)).andReturn(messageMock);
expect(endpointMock.send(messageMock)).andReturn(false);
replay(globalMocks);
poller.setReceiveTimeout(1);
poller.run();
verify(globalMocks);
}
}

3
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/MessagingBridgeTests.java
View File

@@ -24,7 +24,6 @@ import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.springframework.integration.bus.DefaultMessageBus;
import org.springframework.integration.dispatcher.PollingDispatcher;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.message.PollableSource;
@@ -52,7 +51,7 @@ public class MessagingBridgeTests {
return new StringMessage("test");
}
};
PollingDispatcher poller = new PollingDispatcher(source, new PollingSchedule(1000));
SourcePoller poller = new SourcePoller(source, new PollingSchedule(1000));
poller.setMaxMessagesPerPoll(1);
bridge.setSource(poller);
bus.registerEndpoint(bridge);