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

Replacing Schedule with Trigger.

Browse Source
This commit is contained in:
Mark Fisher
2008-09-24 18:29:23 +00:00
parent 8b6f7be72c
commit ee7a8d0725
19 changed files with 102 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
org.springframework.integration/src/main/java/org/springframework/integration/annotation/Poller.java
View File

@@ -39,7 +39,7 @@ import org.springframework.integration.scheduling.PollingSchedule;
@Documented
public @interface Poller {
int period();
int interval();
long initialDelay() default PollingSchedule.DEFAULT_INITIAL_DELAY;

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) {
IntegrationNamespaceUtils.configureSchedule(pollerElement, builder);
IntegrationNamespaceUtils.configureTrigger(pollerElement, builder);
Element txElement = DomUtils.getChildElementByTagName(pollerElement, "transactional");
if (txElement != null) {
IntegrationNamespaceUtils.configureTransactionAttributes(txElement, builder);

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

@@ -46,7 +46,7 @@ public abstract class AbstractOutboundChannelAdapterParser extends AbstractChann
if (!StringUtils.hasText(channelName)) {
throw new ConfigurationException("outbound channel adapter with a 'poller' requires a 'channel' to poll");
}
IntegrationNamespaceUtils.configureSchedule(pollerElement, adapterBuilder);
IntegrationNamespaceUtils.configureTrigger(pollerElement, adapterBuilder);
Element txElement = DomUtils.getChildElementByTagName(pollerElement, "transactional");
if (txElement != null) {
IntegrationNamespaceUtils.configureTransactionAttributes(txElement, adapterBuilder);

8
org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractPollingInboundChannelAdapterParser.java
View File

@@ -23,7 +23,7 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
@@ -45,7 +45,7 @@ public abstract class AbstractPollingInboundChannelAdapterParser extends Abstrac
adapterBuilder.addPropertyReference("source", source);
adapterBuilder.addPropertyReference("outputChannel", channelName);
if (pollerElement != null) {
IntegrationNamespaceUtils.configureSchedule(pollerElement, adapterBuilder);
IntegrationNamespaceUtils.configureTrigger(pollerElement, adapterBuilder);
IntegrationNamespaceUtils.setValueIfAttributeDefined(adapterBuilder, pollerElement, "max-messages-per-poll");
Element txElement = DomUtils.getChildElementByTagName(pollerElement, "transactional");
if (txElement != null) {
@@ -53,14 +53,14 @@ public abstract class AbstractPollingInboundChannelAdapterParser extends Abstrac
}
}
else {
adapterBuilder.addPropertyValue("schedule", new PollingSchedule(this.getDefaultPollInterval()));
adapterBuilder.addPropertyValue("trigger", new IntervalTrigger(this.getDefaultPollInterval()));
}
return adapterBuilder.getBeanDefinition();
}
/**
* Subclasses may override this to provide the default poll interval (when
* no 'schedule' is configured). Otherwise, the value will be 1 second.
* no 'trigger' is configured). Otherwise, the value will be 1 second.
*/
protected int getDefaultPollInterval() {
return 1000;

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

@@ -25,9 +25,9 @@ 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.scheduling.CronSchedule;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.CronTrigger;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.integration.scheduling.Trigger;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.util.StringUtils;
@@ -134,35 +134,33 @@ public abstract class IntegrationNamespaceUtils {
}
/**
* Parse a "poller" element to create a Schedule and add it to the property values of the target builder.
* Parse a "poller" element to create a Trigger and add it to the property values of the target builder.
*
* @param pollerElement the "poller" element to parse
* @param targetBuilder the builder that expects the "schedule" property
* @param targetBuilder the builder that expects the "trigger" property
*/
public static void configureSchedule(Element pollerElement, BeanDefinitionBuilder targetBuilder) {
Schedule schedule = null;
if (!(StringUtils.hasText(pollerElement.getAttribute("period")) ^ StringUtils.hasText(pollerElement.getAttribute("cron")))) {
throw new ConfigurationException("A <poller> element must define either a period "
+ "or a cron expression (but not both)");
public static void configureTrigger(Element pollerElement, BeanDefinitionBuilder targetBuilder) {
Trigger trigger = null;
String interval = pollerElement.getAttribute("period");
String cron = pollerElement.getAttribute("cron");
if (!(StringUtils.hasText(interval) ^ StringUtils.hasText(cron))) {
throw new ConfigurationException(
"A <poller> element must define either a period or a cron expression (but not both).");
}
if (StringUtils.hasText(pollerElement.getAttribute("period"))) {
Long period = Long.valueOf(pollerElement.getAttribute("period"));
schedule = new PollingSchedule(period);
if (StringUtils.hasText(interval)) {
Long period = Long.valueOf(interval);
IntervalTrigger intervalTrigger = new IntervalTrigger(period);
String initialDelay = pollerElement.getAttribute("initial-delay");
if (StringUtils.hasText(initialDelay)) {
((PollingSchedule)schedule).setInitialDelay(Long.valueOf(initialDelay));
}
if ("true".equals(pollerElement.getAttribute("fixed-rate").toLowerCase())) {
((PollingSchedule)schedule).setFixedRate(true);
}
else {
((PollingSchedule)schedule).setFixedRate(false);
intervalTrigger.setInitialDelay(Long.valueOf(initialDelay));
}
intervalTrigger.setFixedRate("true".equals(pollerElement.getAttribute("fixed-rate").toLowerCase()));
trigger = intervalTrigger;
}
if (StringUtils.hasText(pollerElement.getAttribute("cron"))) {
schedule = new CronSchedule(pollerElement.getAttribute("cron"));
trigger = new CronTrigger(pollerElement.getAttribute("cron"));
}
targetBuilder.addPropertyValue("schedule", schedule);
targetBuilder.addPropertyValue("trigger", trigger);
}
/**

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

@@ -27,9 +27,9 @@ import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.AbstractMessageHandlingEndpoint;
import org.springframework.integration.endpoint.AbstractMessageConsumingEndpoint;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.endpoint.AbstractMessageHandlingEndpoint;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -87,11 +87,11 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
AbstractMessageConsumingEndpoint consumingEndpoint = (AbstractMessageConsumingEndpoint) endpoint;
if (pollerAnnotation != null) {
if (inputChannel instanceof PollableChannel) {
PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period());
schedule.setInitialDelay(pollerAnnotation.initialDelay());
schedule.setFixedRate(pollerAnnotation.fixedRate());
schedule.setTimeUnit(pollerAnnotation.timeUnit());
consumingEndpoint.setSchedule(schedule);
IntervalTrigger trigger = new IntervalTrigger(
pollerAnnotation.interval(), pollerAnnotation.timeUnit());
trigger.setInitialDelay(pollerAnnotation.initialDelay(), pollerAnnotation.timeUnit());
trigger.setFixedRate(pollerAnnotation.fixedRate());
consumingEndpoint.setTrigger(trigger);
consumingEndpoint.setMaxMessagesPerPoll(pollerAnnotation.maxMessagesPerPoll());
}
else {

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

@@ -32,8 +32,8 @@ import org.springframework.integration.endpoint.OutboundChannelAdapter;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.message.MethodInvokingConsumer;
import org.springframework.integration.message.MethodInvokingSource;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.integration.scheduling.Trigger;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
@@ -96,11 +96,11 @@ 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.");
}
Schedule schedule = this.createSchedule(pollerAnnotation);
Trigger trigger = this.createTrigger(pollerAnnotation);
SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter();
adapter.setSource(source);
adapter.setOutputChannel(channel);
adapter.setSchedule(schedule);
adapter.setTrigger(trigger);
return adapter;
}
@@ -108,20 +108,20 @@ public class ChannelAdapterAnnotationPostProcessor implements MethodAnnotationPo
OutboundChannelAdapter adapter = new OutboundChannelAdapter(consumer);
adapter.setInputChannel(channel);
if (channel instanceof PollableChannel) {
Schedule schedule = (pollerAnnotation != null)
? this.createSchedule(pollerAnnotation)
: new PollingSchedule(0);
adapter.setSchedule(schedule);
Trigger trigger = (pollerAnnotation != null)
? this.createTrigger(pollerAnnotation)
: new IntervalTrigger(0);
adapter.setTrigger(trigger);
}
return adapter;
}
private Schedule createSchedule(Poller pollerAnnotation) {
PollingSchedule schedule = new PollingSchedule(pollerAnnotation.period());
schedule.setInitialDelay(pollerAnnotation.initialDelay());
schedule.setFixedRate(pollerAnnotation.fixedRate());
schedule.setTimeUnit(pollerAnnotation.timeUnit());
return schedule;
private Trigger createTrigger(Poller pollerAnnotation) {
IntervalTrigger trigger = new IntervalTrigger(
pollerAnnotation.interval(), pollerAnnotation.timeUnit());
trigger.setInitialDelay(pollerAnnotation.initialDelay());
trigger.setFixedRate(pollerAnnotation.fixedRate());
return trigger;
}
private boolean hasReturnValue(Method method) {

12
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractMessageConsumingEndpoint.java
View File

@@ -28,8 +28,8 @@ import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.Subscribable;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.integration.scheduling.Trigger;
/**
* The base class for Message Endpoint implementations that consume Messages.
@@ -40,7 +40,7 @@ public abstract class AbstractMessageConsumingEndpoint extends AbstractEndpoint
private volatile MessageChannel inputChannel;
private volatile Schedule schedule = new PollingSchedule(0);
private volatile Trigger trigger = new IntervalTrigger(0);
private volatile ChannelPoller poller;
@@ -61,8 +61,8 @@ public abstract class AbstractMessageConsumingEndpoint extends AbstractEndpoint
this.inputChannel = inputChannel;
}
public void setSchedule(Schedule schedule) {
this.schedule = schedule;
public void setTrigger(Trigger trigger) {
this.trigger = trigger;
}
public void setTaskExecutor(TaskExecutor taskExecutor) {
@@ -84,7 +84,7 @@ public abstract class AbstractMessageConsumingEndpoint extends AbstractEndpoint
protected void initialize() throws Exception {
synchronized (this.lifecycleMonitor) {
if (this.inputChannel instanceof PollableChannel && this.poller == null) {
this.poller = new ChannelPoller((PollableChannel) this.inputChannel, this.schedule);
this.poller = new ChannelPoller((PollableChannel) this.inputChannel, this.trigger);
this.poller.setMaxMessagesPerPoll(this.maxMessagesPerPoll);
this.configureTransactionSettingsForPoller(this.poller);
if (this.taskExecutor != null) {

27
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPoller.java
View File

@@ -18,12 +18,7 @@ package org.springframework.integration.endpoint;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.scheduling.CronSchedule;
import org.springframework.integration.scheduling.CronTrigger;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.SchedulableTask;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.Trigger;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
@@ -33,6 +28,8 @@ import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.Assert;
/**
* Base class for pollers.
*
* @author Mark Fisher
*/
public abstract class AbstractPoller implements SchedulableTask, InitializingBean {
@@ -40,7 +37,7 @@ public abstract class AbstractPoller implements SchedulableTask, InitializingBea
public static final int MAX_MESSAGES_UNBOUNDED = -1;
private final Schedule schedule;
private final Trigger trigger;
private volatile long maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED;
@@ -63,24 +60,14 @@ public abstract class AbstractPoller implements SchedulableTask, InitializingBea
private final Object initializationMonitor = new Object();
public AbstractPoller(Schedule schedule) {
Assert.notNull(schedule, "schedule must not be null");
this.schedule = schedule;
public AbstractPoller(Trigger trigger) {
Assert.notNull(trigger, "trigger must not be null");
this.trigger = trigger;
}
public Trigger getTrigger() {
if (schedule instanceof PollingSchedule) {
PollingSchedule pollingSchedule = (PollingSchedule) schedule;
IntervalTrigger trigger = new IntervalTrigger(pollingSchedule.getPeriod(), pollingSchedule.getTimeUnit());
trigger.setInitialDelay(pollingSchedule.getInitialDelay());
trigger.setFixedRate(pollingSchedule.getFixedRate());
return trigger;
}
if (schedule instanceof CronSchedule ) {
return new CronTrigger(((CronSchedule) schedule).getCronExpression());
}
return null;
return this.trigger;
}
/**

6
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ChannelPoller.java
View File

@@ -21,7 +21,7 @@ import org.springframework.integration.dispatcher.SimpleDispatcher;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.Subscribable;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.Trigger;
import org.springframework.util.Assert;
/**
@@ -36,8 +36,8 @@ public class ChannelPoller extends AbstractPoller implements Subscribable {
private final SimpleDispatcher dispatcher = new SimpleDispatcher();
public ChannelPoller(PollableChannel channel, Schedule schedule) {
super(schedule);
public ChannelPoller(PollableChannel channel, Trigger trigger) {
super(trigger);
Assert.notNull(channel, "channel must not be null");
this.channel = channel;
}

6
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePoller.java
View File

@@ -23,7 +23,7 @@ import org.springframework.integration.message.MessageDeliveryAware;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.Trigger;
import org.springframework.util.Assert;
/**
@@ -38,8 +38,8 @@ public class SourcePoller extends AbstractPoller {
private volatile long receiveTimeout = 1000;
public SourcePoller(PollableSource<?> source, MessageChannel channel, Schedule schedule) {
super(schedule);
public SourcePoller(PollableSource<?> source, MessageChannel channel, Trigger trigger) {
super(trigger);
Assert.notNull(source, "source must not be null");
Assert.notNull(channel, "channel must not be null");
this.source = source;

10
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java
View File

@@ -22,8 +22,8 @@ import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.MethodInvokingSource;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.integration.scheduling.Trigger;
/**
* A Channel Adapter implementation for connecting a
@@ -36,7 +36,7 @@ public class SourcePollingChannelAdapter extends AbstractMessageProducingEndpoin
private volatile PollableSource<?> source;
private volatile Schedule schedule;
private volatile Trigger trigger;
private volatile SourcePoller poller;
@@ -53,8 +53,8 @@ public class SourcePollingChannelAdapter extends AbstractMessageProducingEndpoin
this.source = source;
}
public void setSchedule(Schedule schedule) {
this.schedule = schedule;
public void setTrigger(Trigger trigger) {
this.trigger = trigger;
}
public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
@@ -73,7 +73,7 @@ public class SourcePollingChannelAdapter extends AbstractMessageProducingEndpoin
if (this.running) {
return;
}
this.poller = new SourcePoller(source, this.getOutputChannel(), schedule);
this.poller = new SourcePoller(source, this.getOutputChannel(), trigger);
if (maxMessagesPerPoll < 0 && source instanceof MethodInvokingSource) {
// 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

7
org.springframework.integration/src/main/java/org/springframework/integration/scheduling/IntervalTrigger.java
View File

@@ -63,6 +63,13 @@ public class IntervalTrigger implements Trigger {
this.initialDelay = initialDelay;
}
/**
* Specify the delay for the initial execution using the given time unit.
*/
public void setInitialDelay(long initialDelay, TimeUnit unit) {
this.initialDelay = unit.toMillis(initialDelay);
}
/**
* Specify whether the interval should be measured between the
* scheduled start times rather than between actual completion times

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

@@ -41,7 +41,7 @@ import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.PollableSource;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.IntervalTrigger;
/**
* @author Mark Fisher
@@ -206,7 +206,7 @@ public class DefaultMessageBusTests {
CountDownLatch latch = new CountDownLatch(1);
SourcePollingChannelAdapter channelAdapter = new SourcePollingChannelAdapter();
channelAdapter.setSource(new FailingSource(latch));
channelAdapter.setSchedule(new PollingSchedule(1000));
channelAdapter.setTrigger(new IntervalTrigger(1000));
channelAdapter.setOutputChannel(outputChannel);
channelAdapter.setBeanName("testChannel");
context.getBeanFactory().registerSingleton("testChannel", channelAdapter);

21
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java
View File

@@ -49,8 +49,8 @@ import org.springframework.integration.endpoint.ServiceActivatorEndpoint;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.integration.scheduling.Trigger;
import org.springframework.integration.util.MethodInvoker;
/**
@@ -329,13 +329,12 @@ public class MessagingAnnotationPostProcessorTests {
processedEndpoint.afterPropertiesSet();
DirectFieldAccessor accessor = new DirectFieldAccessor(processedEndpoint);
ChannelPoller poller = (ChannelPoller) accessor.getPropertyValue("poller");
Schedule schedule = (Schedule) new DirectFieldAccessor(poller).getPropertyValue("schedule");
assertEquals(PollingSchedule.class, schedule.getClass());
PollingSchedule pollingSchedule = (PollingSchedule) schedule;
assertEquals(1234, pollingSchedule.getPeriod());
assertEquals(5678, pollingSchedule.getInitialDelay());
assertEquals(true, pollingSchedule.getFixedRate());
assertEquals(TimeUnit.SECONDS, pollingSchedule.getTimeUnit());
Trigger trigger = (Trigger) new DirectFieldAccessor(poller).getPropertyValue("trigger");
assertEquals(IntervalTrigger.class, trigger.getClass());
DirectFieldAccessor triggerAccessor = new DirectFieldAccessor(trigger);
assertEquals(new Long(123000), triggerAccessor.getPropertyValue("interval"));
assertEquals(new Long(456000), triggerAccessor.getPropertyValue("initialDelay"));
assertEquals(true, triggerAccessor.getPropertyValue("fixedRate"));
}
@Test
@@ -448,7 +447,7 @@ public class MessagingAnnotationPostProcessorTests {
private static class AnnotatedEndpointWithPolledAnnotation {
@ServiceActivator(inputChannel="testChannel")
@Poller(period=1234, initialDelay=5678, fixedRate=true, timeUnit=TimeUnit.SECONDS)
@Poller(interval=123, initialDelay=456, fixedRate=true, timeUnit=TimeUnit.SECONDS)
public String prependFoo(String s) {
return "foo" + s;
}
@@ -470,7 +469,7 @@ public class MessagingAnnotationPostProcessorTests {
private static class ChannelAdapterAnnotationTestBean {
@ChannelAdapter("testChannel")
@Poller(period = 1000, initialDelay = 0, maxMessagesPerPoll = 1)
@Poller(interval=1000, initialDelay=0, maxMessagesPerPoll=1)
public String test() {
return "test";
}

12
org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ChannelPollerTests.java
View File

@@ -30,7 +30,7 @@ import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageRejectedException;
import org.springframework.integration.scheduling.Schedule;
import org.springframework.integration.scheduling.Trigger;
/**
* @author Iwein Fuld
@@ -39,16 +39,16 @@ import org.springframework.integration.scheduling.Schedule;
public class ChannelPollerTests {
private ChannelPoller poller;
private Schedule scheduleMock = createMock(Schedule.class);
private Trigger triggerMock = createMock(Trigger.class);
private PollableChannel channelMock = createMock(PollableChannel.class);
private MessageConsumer endpointMock = createMock(MessageConsumer.class);
private Message messageMock = createMock(Message.class);
private Object[] globalMocks = new Object[] { scheduleMock, channelMock, endpointMock, messageMock };
private Object[] globalMocks = new Object[] { triggerMock, channelMock, endpointMock, messageMock };
@Before
public void init() {
poller = new ChannelPoller(channelMock, scheduleMock);
poller = new ChannelPoller(channelMock, triggerMock);
poller.subscribe(endpointMock);
poller.setReceiveTimeout(-1);
reset(globalMocks);
@@ -112,7 +112,7 @@ public class ChannelPollerTests {
@Test
public void blockingSourceTimedOut() {
poller = new ChannelPoller(channelMock, scheduleMock);
poller = new ChannelPoller(channelMock, triggerMock);
poller.subscribe(endpointMock);
// we don't need to await the timeout, returning null suffices
expect(channelMock.receive(1)).andReturn(null);
@@ -124,7 +124,7 @@ public class ChannelPollerTests {
@Test
public void blockingSourceNotTimedOut() {
poller = new ChannelPoller(channelMock, scheduleMock);
poller = new ChannelPoller(channelMock, triggerMock);
poller.subscribe(endpointMock);
expect(channelMock.receive(1)).andReturn(messageMock);
endpointMock.onMessage(messageMock);