diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetTests.java index ffd1ea558d..86914b5045 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetTests.java @@ -25,8 +25,7 @@ import org.junit.Before; import org.junit.Test; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.dispatcher.BroadcastingDispatcher; -import org.springframework.integration.dispatcher.PollingDispatcher; +import org.springframework.integration.endpoint.ChannelPoller; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; import org.springframework.integration.scheduling.PollingSchedule; @@ -38,13 +37,13 @@ public class ByteStreamTargetTests { private QueueChannel channel; - private PollingDispatcher dispatcher; + private ChannelPoller poller; @Before public void initialize() { this.channel = new QueueChannel(10); - this.dispatcher = new PollingDispatcher(channel, new PollingSchedule(0), new BroadcastingDispatcher()); + this.poller = new ChannelPoller(channel, new PollingSchedule(0)); } @@ -74,12 +73,12 @@ public class ByteStreamTargetTests { public void testMaxMessagesPerTaskSameAsMessageCount() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - dispatcher.setMaxMessagesPerPoll(3); - dispatcher.subscribe(target); + poller.setMaxMessagesPerPoll(3); + poller.subscribe(target); 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); - dispatcher.run(); + poller.run(); byte[] result = stream.toByteArray(); assertEquals(9, result.length); assertEquals(1, result[0]); @@ -90,12 +89,12 @@ public class ByteStreamTargetTests { public void testMaxMessagesPerTaskLessThanMessageCount() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - dispatcher.setMaxMessagesPerPoll(2); - dispatcher.subscribe(target); + poller.setMaxMessagesPerPoll(2); + poller.subscribe(target); 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); - dispatcher.run(); + poller.run(); byte[] result = stream.toByteArray(); assertEquals(6, result.length); assertEquals(1, result[0]); @@ -105,13 +104,13 @@ public class ByteStreamTargetTests { public void testMaxMessagesPerTaskExceedsMessageCount() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - dispatcher.setMaxMessagesPerPoll(5); - dispatcher.setReceiveTimeout(0); - dispatcher.subscribe(target); + poller.setMaxMessagesPerPoll(5); + poller.setReceiveTimeout(0); + poller.subscribe(target); 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); - dispatcher.run(); + poller.run(); byte[] result = stream.toByteArray(); assertEquals(9, result.length); assertEquals(1, result[0]); @@ -121,17 +120,17 @@ public class ByteStreamTargetTests { public void testMaxMessagesLessThanMessageCountWithMultipleDispatches() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - dispatcher.setMaxMessagesPerPoll(2); - dispatcher.setReceiveTimeout(0); - dispatcher.subscribe(target); + poller.setMaxMessagesPerPoll(2); + poller.setReceiveTimeout(0); + poller.subscribe(target); 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); - dispatcher.run(); + poller.run(); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); assertEquals(1, result1[0]); - dispatcher.run(); + poller.run(); byte[] result2 = stream.toByteArray(); assertEquals(9, result2.length); assertEquals(1, result2[0]); @@ -142,17 +141,17 @@ public class ByteStreamTargetTests { public void testMaxMessagesExceedsMessageCountWithMultipleDispatches() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - dispatcher.setMaxMessagesPerPoll(5); - dispatcher.setReceiveTimeout(0); - dispatcher.subscribe(target); + poller.setMaxMessagesPerPoll(5); + poller.setReceiveTimeout(0); + poller.subscribe(target); 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); - dispatcher.run(); + poller.run(); byte[] result1 = stream.toByteArray(); assertEquals(9, result1.length); assertEquals(1, result1[0]); - dispatcher.run(); + poller.run(); byte[] result2 = stream.toByteArray(); assertEquals(9, result2.length); assertEquals(1, result2[0]); @@ -162,17 +161,17 @@ public class ByteStreamTargetTests { public void testStreamResetBetweenDispatches() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - dispatcher.setMaxMessagesPerPoll(2); - dispatcher.setReceiveTimeout(0); - dispatcher.subscribe(target); + poller.setMaxMessagesPerPoll(2); + poller.setReceiveTimeout(0); + poller.subscribe(target); 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); - dispatcher.run(); + poller.run(); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); stream.reset(); - dispatcher.run(); + poller.run(); byte[] result2 = stream.toByteArray(); assertEquals(3, result2.length); assertEquals(7, result2[0]); @@ -182,18 +181,18 @@ public class ByteStreamTargetTests { public void testStreamWriteBetweenDispatches() throws IOException { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - dispatcher.setMaxMessagesPerPoll(2); - dispatcher.setReceiveTimeout(0); - dispatcher.subscribe(target); + poller.setMaxMessagesPerPoll(2); + poller.setReceiveTimeout(0); + poller.subscribe(target); 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); - dispatcher.run(); + poller.run(); byte[] result1 = stream.toByteArray(); assertEquals(6, result1.length); stream.write(new byte[] {123}); stream.flush(); - dispatcher.run(); + poller.run(); byte[] result2 = stream.toByteArray(); assertEquals(10, result2.length); assertEquals(1, result2[0]); diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetTests.java index 798096d429..4b5a08d7db 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetTests.java @@ -24,8 +24,7 @@ import org.junit.Before; import org.junit.Test; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.dispatcher.BroadcastingDispatcher; -import org.springframework.integration.dispatcher.PollingDispatcher; +import org.springframework.integration.endpoint.ChannelPoller; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; import org.springframework.integration.scheduling.PollingSchedule; @@ -37,13 +36,13 @@ public class CharacterStreamTargetTests { private QueueChannel channel; - private PollingDispatcher dispatcher; + private ChannelPoller poller; @Before public void initialize() { this.channel = new QueueChannel(10); - this.dispatcher = new PollingDispatcher(channel, new PollingSchedule(0), new BroadcastingDispatcher()); + this.poller = new ChannelPoller(channel, new PollingSchedule(0)); } @@ -59,13 +58,13 @@ public class CharacterStreamTargetTests { public void testTwoStringsAndNoNewLinesByDefault() { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); - dispatcher.subscribe(target); - dispatcher.setMaxMessagesPerPoll(1); + poller.subscribe(target); + poller.setMaxMessagesPerPoll(1); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - dispatcher.run(); + poller.run(); assertEquals("foo", writer.toString()); - dispatcher.run(); + poller.run(); assertEquals("foobar", writer.toString()); } @@ -74,14 +73,14 @@ public class CharacterStreamTargetTests { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); target.setShouldAppendNewLine(true); - dispatcher.subscribe(target); - dispatcher.setMaxMessagesPerPoll(1); + poller.subscribe(target); + poller.setMaxMessagesPerPoll(1); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - dispatcher.run(); + poller.run(); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine, writer.toString()); - dispatcher.run(); + poller.run(); assertEquals("foo" + newLine + "bar" + newLine, writer.toString()); } @@ -89,11 +88,11 @@ public class CharacterStreamTargetTests { public void testMaxMessagesPerTaskSameAsMessageCount() { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); - dispatcher.setMaxMessagesPerPoll(2); - dispatcher.subscribe(target); + poller.setMaxMessagesPerPoll(2); + poller.subscribe(target); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - dispatcher.run(); + poller.run(); assertEquals("foobar", writer.toString()); } @@ -101,13 +100,13 @@ public class CharacterStreamTargetTests { public void testMaxMessagesPerTaskExceedsMessageCountWithAppendedNewLines() { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); - dispatcher.setMaxMessagesPerPoll(10); - dispatcher.setReceiveTimeout(0); - dispatcher.subscribe(target); + poller.setMaxMessagesPerPoll(10); + poller.setReceiveTimeout(0); + poller.subscribe(target); target.setShouldAppendNewLine(true); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); - dispatcher.run(); + poller.run(); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine + "bar" + newLine, writer.toString()); } @@ -116,11 +115,11 @@ public class CharacterStreamTargetTests { public void testSingleNonStringObject() { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); - dispatcher.subscribe(target); - dispatcher.setMaxMessagesPerPoll(1); + poller.subscribe(target); + poller.setMaxMessagesPerPoll(1); TestObject testObject = new TestObject("foo"); channel.send(new GenericMessage(testObject)); - dispatcher.run(); + poller.run(); assertEquals("foo", writer.toString()); } @@ -128,14 +127,14 @@ public class CharacterStreamTargetTests { public void testTwoNonStringObjectWithOutNewLines() { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); - dispatcher.setReceiveTimeout(0); - dispatcher.setMaxMessagesPerPoll(2); - dispatcher.subscribe(target); + poller.setReceiveTimeout(0); + poller.setMaxMessagesPerPoll(2); + poller.subscribe(target); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); channel.send(new GenericMessage(testObject1), 0); channel.send(new GenericMessage(testObject2), 0); - dispatcher.run(); + poller.run(); assertEquals("foobar", writer.toString()); } @@ -144,14 +143,14 @@ public class CharacterStreamTargetTests { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); target.setShouldAppendNewLine(true); - dispatcher.setReceiveTimeout(0); - dispatcher.setMaxMessagesPerPoll(2); - dispatcher.subscribe(target); + poller.setReceiveTimeout(0); + poller.setMaxMessagesPerPoll(2); + poller.subscribe(target); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); channel.send(new GenericMessage(testObject1), 0); channel.send(new GenericMessage(testObject2), 0); - dispatcher.run(); + poller.run(); String newLine = System.getProperty("line.separator"); assertEquals("foo" + newLine + "bar" + newLine, writer.toString()); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java index 146ae34f23..e1653b3db4 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/bus/DefaultMessageBus.java @@ -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 pollingDispatchers = new CopyOnWriteArraySet(); + private final Set pollers = new CopyOnWriteArraySet(); 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) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java index 2d58b02cc0..ffd19390d5 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/DirectChannel.java @@ -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); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java index afcaad3c9b..20c8f2e096 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java @@ -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); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java index 34709c10bd..4b1f09cb53 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/AbstractEndpointParser.java @@ -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 { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java index 2334e23b44..82df1b26e5 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/ChannelAdapterParser.java @@ -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)) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java index 4a9894f2d2..3dcdae7572 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/IntegrationNamespaceUtils.java @@ -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 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"); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/PollingDispatcherFactoryBean.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/PollingDispatcherFactoryBean.java deleted file mode 100644 index d7e4c9694a..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/PollingDispatcherFactoryBean.java +++ /dev/null @@ -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; - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java index 646638279d..1f44446d21 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java @@ -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) 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) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java index 3bef059c4e..221772fbbb 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java @@ -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); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java index 2b44ef4523..ad38e0c18e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java @@ -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) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java index 752044fce8..cd0142b8ab 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java @@ -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); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java deleted file mode 100644 index 9ab4e68330..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java +++ /dev/null @@ -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}. - *

- * 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. - * - *

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; - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java index 8f5bbe6409..bfdfd3ecb6 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/SimpleDispatcher.java @@ -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."); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPoller.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPoller.java new file mode 100644 index 0000000000..2594263707 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPoller.java @@ -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. + * + *

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(); + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ChannelPoller.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ChannelPoller.java new file mode 100644 index 0000000000..e1942cbc4d --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ChannelPoller.java @@ -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); + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePoller.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePoller.java new file mode 100644 index 0000000000..e42cc73c44 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/SourcePoller.java @@ -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}. + *

+ * 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); + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java index 075444defc..dd1aa428df 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/bus/DefaultMessageBusTests.java @@ -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) diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java index 990356c399..840948bccc 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/BroadcastingDispatcherTests.java @@ -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()); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTests.java deleted file mode 100644 index 4982c592c3..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTests.java +++ /dev/null @@ -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); - } - -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java index ddfabc659a..b4bd9ab78b 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/SimpleDispatcherTests.java @@ -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()); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ChannelPollerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ChannelPollerTests.java new file mode 100644 index 0000000000..7ad8878d03 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ChannelPollerTests.java @@ -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); + } + +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/MessagingBridgeTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/MessagingBridgeTests.java index bf170c5055..05049dadab 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/MessagingBridgeTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/MessagingBridgeTests.java @@ -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);