diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java index 6aaa2bc0e0..6ea0b34d0b 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceTests.java @@ -23,11 +23,7 @@ import java.io.ByteArrayInputStream; import org.junit.Test; -import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.endpoint.SourceEndpoint; -import org.springframework.integration.endpoint.TriggerMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.scheduling.PollingSchedule; /** * @author Mark Fisher @@ -38,69 +34,48 @@ public class ByteStreamSourceTests { public void testEndOfStream() { byte[] bytes = new byte[] {1,2,3}; ByteArrayInputStream stream = new ByteArrayInputStream(bytes); - QueueChannel channel = new QueueChannel(); ByteStreamSource source = new ByteStreamSource(stream); - SourceEndpoint endpoint = new SourceEndpoint(source); - endpoint.setTarget(channel); - endpoint.afterPropertiesSet(); - endpoint.send(new TriggerMessage()); - Message message1 = channel.receive(500); + Message message1 = source.receive(); byte[] payload = (byte[]) message1.getPayload(); assertEquals(3, payload.length); assertEquals(1, payload[0]); assertEquals(2, payload[1]); assertEquals(3, payload[2]); - Message message2 = channel.receive(0); + Message message2 = source.receive(); assertNull(message2); - endpoint.send(new TriggerMessage()); - Message message3 = channel.receive(0); - assertNull(message3); } @Test public void testByteArrayIsTruncated() { byte[] bytes = new byte[] {0,1,2,3,4,5}; ByteArrayInputStream stream = new ByteArrayInputStream(bytes); - QueueChannel channel = new QueueChannel(); ByteStreamSource source = new ByteStreamSource(stream); source.setBytesPerMessage(4); - PollingSchedule schedule = new PollingSchedule(1000); - schedule.setInitialDelay(10000); - SourceEndpoint endpoint = new SourceEndpoint(source); - endpoint.setTarget(channel); - endpoint.afterPropertiesSet(); - endpoint.send(new TriggerMessage()); - Message message1 = channel.receive(0); + Message message1 = source.receive(); assertEquals(4, ((byte[]) message1.getPayload()).length); - Message message2 = channel.receive(0); - assertNull(message2); - endpoint.send(new TriggerMessage()); - Message message3 = channel.receive(0); - assertEquals(2, ((byte[]) message3.getPayload()).length); + Message message2 = source.receive(); + assertEquals(2, ((byte[]) message2.getPayload()).length); + Message message3 = source.receive(); + assertNull(message3); } @Test public void testByteArrayIsNotTruncated() { byte[] bytes = new byte[] {0,1,2,3,4,5}; ByteArrayInputStream stream = new ByteArrayInputStream(bytes); - QueueChannel channel = new QueueChannel(); ByteStreamSource source = new ByteStreamSource(stream); source.setBytesPerMessage(4); source.setShouldTruncate(false); - PollingSchedule schedule = new PollingSchedule(1000); - schedule.setInitialDelay(10000); - SourceEndpoint endpoint = new SourceEndpoint(source); - endpoint.setTarget(channel); - endpoint.afterPropertiesSet(); - endpoint.send(new TriggerMessage()); - Message message1 = channel.receive(0); + Message message1 = source.receive(); assertEquals(4, ((byte[]) message1.getPayload()).length); - Message message2 = channel.receive(0); - assertNull(message2); - endpoint.send(new TriggerMessage()); - Message message3 = channel.receive(0); - assertEquals(4, ((byte[]) message3.getPayload()).length); - assertEquals(0, ((byte[]) message3.getPayload())[3]); + Message message2 = source.receive(); + assertEquals(4, ((byte[]) message2.getPayload()).length); + assertEquals(4, ((byte[]) message2.getPayload())[0]); + assertEquals(5, ((byte[]) message2.getPayload())[1]); + assertEquals(0, ((byte[]) message2.getPayload())[2]); + assertEquals(0, ((byte[]) message2.getPayload())[3]); + Message message3 = source.receive(); + assertNull(message3); } } 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 89e7f39597..ffd1ea558d 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 @@ -44,7 +44,7 @@ public class ByteStreamTargetTests { @Before public void initialize() { this.channel = new QueueChannel(10); - this.dispatcher = new PollingDispatcher(channel, new BroadcastingDispatcher(), new PollingSchedule(0)); + this.dispatcher = new PollingDispatcher(channel, new PollingSchedule(0), new BroadcastingDispatcher()); } @@ -74,8 +74,8 @@ public class ByteStreamTargetTests { public void testMaxMessagesPerTaskSameAsMessageCount() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - dispatcher.setMaxMessagesPerTask(3); - dispatcher.addTarget(target); + dispatcher.setMaxMessagesPerPoll(3); + dispatcher.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); @@ -90,8 +90,8 @@ public class ByteStreamTargetTests { public void testMaxMessagesPerTaskLessThanMessageCount() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - dispatcher.setMaxMessagesPerTask(2); - dispatcher.addTarget(target); + dispatcher.setMaxMessagesPerPoll(2); + dispatcher.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); @@ -105,9 +105,9 @@ public class ByteStreamTargetTests { public void testMaxMessagesPerTaskExceedsMessageCount() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - dispatcher.setMaxMessagesPerTask(5); + dispatcher.setMaxMessagesPerPoll(5); dispatcher.setReceiveTimeout(0); - dispatcher.addTarget(target); + dispatcher.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); @@ -121,9 +121,9 @@ public class ByteStreamTargetTests { public void testMaxMessagesLessThanMessageCountWithMultipleDispatches() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - dispatcher.setMaxMessagesPerTask(2); + dispatcher.setMaxMessagesPerPoll(2); dispatcher.setReceiveTimeout(0); - dispatcher.addTarget(target); + dispatcher.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); @@ -142,9 +142,9 @@ public class ByteStreamTargetTests { public void testMaxMessagesExceedsMessageCountWithMultipleDispatches() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - dispatcher.setMaxMessagesPerTask(5); + dispatcher.setMaxMessagesPerPoll(5); dispatcher.setReceiveTimeout(0); - dispatcher.addTarget(target); + dispatcher.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); @@ -162,9 +162,9 @@ public class ByteStreamTargetTests { public void testStreamResetBetweenDispatches() { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - dispatcher.setMaxMessagesPerTask(2); + dispatcher.setMaxMessagesPerPoll(2); dispatcher.setReceiveTimeout(0); - dispatcher.addTarget(target); + dispatcher.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); @@ -182,9 +182,9 @@ public class ByteStreamTargetTests { public void testStreamWriteBetweenDispatches() throws IOException { ByteArrayOutputStream stream = new ByteArrayOutputStream(); ByteStreamTarget target = new ByteStreamTarget(stream); - dispatcher.setMaxMessagesPerTask(2); + dispatcher.setMaxMessagesPerPoll(2); dispatcher.setReceiveTimeout(0); - dispatcher.addTarget(target); + dispatcher.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); diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java index 1e4bbec690..4288df0530 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceTests.java @@ -23,11 +23,7 @@ import java.io.StringReader; import org.junit.Test; -import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.endpoint.SourceEndpoint; -import org.springframework.integration.endpoint.TriggerMessage; import org.springframework.integration.message.Message; -import org.springframework.integration.scheduling.PollingSchedule; /** * @author Mark Fisher @@ -37,21 +33,11 @@ public class CharacterStreamSourceTests { @Test public void testEndOfStream() { StringReader reader = new StringReader("test"); - QueueChannel channel = new QueueChannel(); CharacterStreamSource source = new CharacterStreamSource(reader); - PollingSchedule schedule = new PollingSchedule(1000); - schedule.setInitialDelay(10000); - SourceEndpoint endpoint = new SourceEndpoint(source); - endpoint.setTarget(channel); - endpoint.afterPropertiesSet(); - endpoint.send(new TriggerMessage()); - Message message1 = channel.receive(0); + Message message1 = source.receive(); assertEquals("test", message1.getPayload()); - Message message2 = channel.receive(0); + Message message2 = source.receive(); assertNull(message2); - endpoint.send(new TriggerMessage()); - Message message3 = channel.receive(0); - assertNull(message3); } } 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 348d029298..798096d429 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 @@ -43,7 +43,7 @@ public class CharacterStreamTargetTests { @Before public void initialize() { this.channel = new QueueChannel(10); - this.dispatcher = new PollingDispatcher(channel, new BroadcastingDispatcher(), new PollingSchedule(0)); + this.dispatcher = new PollingDispatcher(channel, new PollingSchedule(0), new BroadcastingDispatcher()); } @@ -59,7 +59,8 @@ public class CharacterStreamTargetTests { public void testTwoStringsAndNoNewLinesByDefault() { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); - dispatcher.addTarget(target); + dispatcher.subscribe(target); + dispatcher.setMaxMessagesPerPoll(1); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); dispatcher.run(); @@ -73,7 +74,8 @@ public class CharacterStreamTargetTests { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); target.setShouldAppendNewLine(true); - dispatcher.addTarget(target); + dispatcher.subscribe(target); + dispatcher.setMaxMessagesPerPoll(1); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); dispatcher.run(); @@ -87,8 +89,8 @@ public class CharacterStreamTargetTests { public void testMaxMessagesPerTaskSameAsMessageCount() { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); - dispatcher.setMaxMessagesPerTask(2); - dispatcher.addTarget(target); + dispatcher.setMaxMessagesPerPoll(2); + dispatcher.subscribe(target); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); dispatcher.run(); @@ -99,9 +101,9 @@ public class CharacterStreamTargetTests { public void testMaxMessagesPerTaskExceedsMessageCountWithAppendedNewLines() { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); - dispatcher.setMaxMessagesPerTask(10); + dispatcher.setMaxMessagesPerPoll(10); dispatcher.setReceiveTimeout(0); - dispatcher.addTarget(target); + dispatcher.subscribe(target); target.setShouldAppendNewLine(true); channel.send(new StringMessage("foo"), 0); channel.send(new StringMessage("bar"), 0); @@ -114,7 +116,8 @@ public class CharacterStreamTargetTests { public void testSingleNonStringObject() { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); - dispatcher.addTarget(target); + dispatcher.subscribe(target); + dispatcher.setMaxMessagesPerPoll(1); TestObject testObject = new TestObject("foo"); channel.send(new GenericMessage(testObject)); dispatcher.run(); @@ -126,8 +129,8 @@ public class CharacterStreamTargetTests { StringWriter writer = new StringWriter(); CharacterStreamTarget target = new CharacterStreamTarget(writer); dispatcher.setReceiveTimeout(0); - dispatcher.setMaxMessagesPerTask(2); - dispatcher.addTarget(target); + dispatcher.setMaxMessagesPerPoll(2); + dispatcher.subscribe(target); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); channel.send(new GenericMessage(testObject1), 0); @@ -142,8 +145,8 @@ public class CharacterStreamTargetTests { CharacterStreamTarget target = new CharacterStreamTarget(writer); target.setShouldAppendNewLine(true); dispatcher.setReceiveTimeout(0); - dispatcher.setMaxMessagesPerTask(2); - dispatcher.addTarget(target); + dispatcher.setMaxMessagesPerPoll(2); + dispatcher.subscribe(target); TestObject testObject1 = new TestObject("foo"); TestObject testObject2 = new TestObject("bar"); channel.send(new GenericMessage(testObject1), 0); 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 bbe5985ae4..0c421ae3c6 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 @@ -26,6 +26,7 @@ import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; + import org.springframework.beans.BeansException; import org.springframework.beans.factory.DisposableBean; import org.springframework.context.ApplicationContext; @@ -44,10 +45,10 @@ import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.factory.ChannelFactory; import org.springframework.integration.channel.factory.QueueChannelFactory; +import org.springframework.integration.dispatcher.PollingDispatcher; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.DefaultEndpointRegistry; import org.springframework.integration.endpoint.EndpointRegistry; -import org.springframework.integration.endpoint.EndpointTrigger; import org.springframework.integration.endpoint.HandlerEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; import org.springframework.integration.endpoint.MessagingGateway; @@ -55,6 +56,7 @@ import org.springframework.integration.endpoint.TargetEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.MessageSource; import org.springframework.integration.message.MessageTarget; +import org.springframework.integration.message.PollableSource; import org.springframework.integration.message.SubscribableSource; import org.springframework.integration.scheduling.MessagePublishingErrorHandler; import org.springframework.integration.scheduling.PollingSchedule; @@ -84,7 +86,7 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A private final EndpointRegistry endpointRegistry = new DefaultEndpointRegistry(); - private final Set endpointTriggers = new CopyOnWriteArraySet(); + private final Set pollingDispatchers = new CopyOnWriteArraySet(); private volatile Schedule defaultPollerSchedule = new PollingSchedule(0); @@ -348,11 +350,13 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A } return; } - Schedule schedule = endpoint.getSchedule(); - EndpointTrigger trigger = new EndpointTrigger(schedule != null ? schedule : this.defaultPollerSchedule); - trigger.addTarget(endpoint); - if (this.endpointTriggers.add(trigger)) { - this.taskScheduler.schedule(trigger); + if (source != null && source instanceof PollableSource) { + Schedule schedule = endpoint.getSchedule(); + schedule = schedule != null ? schedule : this.defaultPollerSchedule; + PollingDispatcher poller = new PollingDispatcher((PollableSource) source, schedule); + poller.subscribe(endpoint); + this.pollingDispatchers.add(poller); + this.taskScheduler.schedule(poller); } } @@ -389,10 +393,10 @@ public class DefaultMessageBus implements MessageBus, ApplicationContextAware, A public void deactivateEndpoint(MessageEndpoint endpoint) { Assert.notNull(endpoint, "'endpoint' must not be null"); - for (EndpointTrigger trigger : this.endpointTriggers) { - boolean removed = trigger.removeTarget(endpoint); + for (PollingDispatcher poller : this.pollingDispatchers) { + boolean removed = poller.unsubscribe(endpoint); if (removed && this.logger.isInfoEnabled()) { - logger.info("removed endpoint '" + endpoint + "' from dispatcher"); + logger.info("removed endpoint '" + endpoint + "' from dispatcher '" + poller + "'"); } } if (endpoint instanceof Lifecycle) { 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 index 8a22c28384..4d47dda07d 100644 --- 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 @@ -16,11 +16,15 @@ package org.springframework.integration.dispatcher; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.integration.message.BlockingSource; import org.springframework.integration.message.BlockingTarget; -import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageExchangeTemplate; import org.springframework.integration.message.MessageTarget; 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; @@ -28,7 +32,14 @@ import org.springframework.util.Assert; /** * @author Mark Fisher */ -public class PollingDispatcher implements SchedulableTask { +public class PollingDispatcher implements SchedulableTask, SubscribableSource { + + public final static int MAX_MESSAGES_UNBOUNDED = -1; + + public final static long DEFAULT_RECEIVE_TIMEOUT = 5000; + + + private final Log logger = LogFactory.getLog(this.getClass()); private final PollableSource source; @@ -36,21 +47,31 @@ public class PollingDispatcher implements SchedulableTask { private final Schedule schedule; - private volatile long receiveTimeout = 5000; + private final MessageExchangeTemplate messageExchangeTemplate; - private volatile int maxMessagesPerTask = 1; + 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, MessageDispatcher dispatcher, Schedule 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"); - Assert.notNull(dispatcher, "dispatcher must not be null"); this.source = source; - this.dispatcher = dispatcher; this.schedule = schedule; + this.dispatcher = (dispatcher != null) + ? dispatcher : new SimpleDispatcher(); + this.messageExchangeTemplate = (messageExchangeTemplate != null) + ? messageExchangeTemplate : createDefaultTemplate(); } @@ -63,7 +84,7 @@ public class PollingDispatcher implements SchedulableTask { * The default value is 5000 (5 seconds). */ public void setReceiveTimeout(long receiveTimeout) { - this.receiveTimeout = receiveTimeout; + this.messageExchangeTemplate.setReceiveTimeout(receiveTimeout); } /** @@ -79,16 +100,20 @@ public class PollingDispatcher implements SchedulableTask { * 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 setMaxMessagesPerTask(int maxMessagesPerTask) { - this.maxMessagesPerTask = maxMessagesPerTask; + public void setMaxMessagesPerPoll(int maxMessagesPerPoll) { + this.maxMessagesPerPoll = maxMessagesPerPoll; } - public boolean addTarget(MessageTarget target) { + public boolean subscribe(MessageTarget target) { return this.dispatcher.addTarget(target); } - public boolean removeTarget(MessageTarget target) { + public boolean unsubscribe(MessageTarget target) { return this.dispatcher.removeTarget(target); } @@ -98,22 +123,29 @@ public class PollingDispatcher implements SchedulableTask { public void run() { int count = 0; - while (this.maxMessagesPerTask <= 0 || count < this.maxMessagesPerTask) { - if (!this.dispatch()) { - return; + 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; } - private boolean dispatch() { - final Message message = (this.source instanceof BlockingSource && this.receiveTimeout >= 0) - ? ((BlockingSource) this.source).receive(this.receiveTimeout) - : this.source.receive(); - if (message == null) { - return false; - } - return this.dispatcher.send(message); + 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/endpoint/AbstractEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java index fdefae56b2..103e132810 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractEndpoint.java @@ -208,10 +208,6 @@ public abstract class AbstractEndpoint implements MessageEndpoint, ChannelRegist if (logger.isDebugEnabled()) { logger.debug("endpoint '" + this + "' handling message: " + message); } - if (message.getPayload() instanceof EndpointVisitor) { - ((EndpointVisitor) message.getPayload()).visitEndpoint(this); - return true; - } return this.send(message, 0); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointPoller.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointPoller.java deleted file mode 100644 index 83822471f4..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointPoller.java +++ /dev/null @@ -1,41 +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.endpoint; - -import org.springframework.integration.ConfigurationException; -import org.springframework.integration.message.MessageSource; -import org.springframework.integration.message.PollableSource; - -/** - * @author Mark Fisher - */ -public class EndpointPoller implements EndpointVisitor { - - public void visitEndpoint(MessageEndpoint endpoint) { - MessageSource source = endpoint.getSource(); - if (source == null) { - throw new ConfigurationException("unable to poll for endpoint '" - + endpoint + "', source is null"); - } - if (!(source instanceof PollableSource)) { - throw new ConfigurationException("unable to poll for endpoint '" - + endpoint + ", source is not a PollableSource"); - } - endpoint.getMessageExchangeTemplate().receiveAndForward((PollableSource) source, endpoint); - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointTrigger.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointTrigger.java deleted file mode 100644 index fdb6009f1c..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointTrigger.java +++ /dev/null @@ -1,65 +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.endpoint; - -import org.springframework.integration.dispatcher.BroadcastingDispatcher; -import org.springframework.integration.dispatcher.PollingDispatcher; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.PollableSource; -import org.springframework.integration.scheduling.PollingSchedule; -import org.springframework.integration.scheduling.Schedule; - -/** - * A {@link PollingDispatcher} implementation that sends a message - * to trigger endpoint polling. - * - * @author Mark Fisher - */ -public class EndpointTrigger extends PollingDispatcher { - - /** - * Create an endpoint trigger with the specified {@link Schedule}. - */ - public EndpointTrigger(Schedule schedule) { - super(new TriggerSource(), new BroadcastingDispatcher(), schedule); - } - - /** - * Create an endpoint trigger. A {@link PollingSchedule} will be - * created with the specified interval. - */ - public EndpointTrigger(long interval) { - this(new PollingSchedule(interval)); - } - - /** - * Create an endpoint trigger that will run one time only when submitted to - * a {@link org.springframework.integration.scheduling.TaskScheduler}. - */ - public EndpointTrigger() { - this(null); - } - - - private static class TriggerSource implements PollableSource { - - public Message receive() { - return new TriggerMessage(); - } - } - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointVisitor.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointVisitor.java deleted file mode 100644 index 259a59c202..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointVisitor.java +++ /dev/null @@ -1,26 +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.endpoint; - -/** - * @author Mark Fisher - */ -public interface EndpointVisitor { - - void visitEndpoint(MessageEndpoint endpoint); - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TriggerMessage.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TriggerMessage.java deleted file mode 100644 index 8bca61650d..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TriggerMessage.java +++ /dev/null @@ -1,33 +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.endpoint; - -import org.springframework.integration.message.GenericMessage; - -/** - * A convenience Message implementation for sending a polling trigger - * to an endpoint. - * - * @author Mark Fisher - */ -public class TriggerMessage extends GenericMessage { - - public TriggerMessage() { - super(new EndpointPoller()); - } - -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTest.java b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTests.java similarity index 71% rename from org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTest.java rename to org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTests.java index aa29dc5540..4dd067372e 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTest.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/dispatcher/PollingDispatcherTests.java @@ -1,3 +1,19 @@ +/* + * 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; @@ -8,39 +24,39 @@ import static org.easymock.EasyMock.verify; import org.junit.Before; import org.junit.Test; + 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 PollingDispatcherTest { +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 }; + private Object[] globalMocks = new Object[] { scheduleMock, dispatcherMock, sourceMock, messageMock }; + @Before public void init() { - pollingDispatcher = new PollingDispatcher(sourceMock, dispatcherMock, - scheduleMock); + 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); } @@ -50,7 +66,7 @@ public class PollingDispatcherTest { expect(sourceMock.receive()).andReturn(messageMock).times(5); expect(dispatcherMock.send(messageMock)).andReturn(true).times(5); replay(globalMocks); - pollingDispatcher.setMaxMessagesPerTask(5); + pollingDispatcher.setMaxMessagesPerPoll(5); pollingDispatcher.run(); verify(globalMocks); } @@ -61,7 +77,7 @@ public class PollingDispatcherTest { expect(sourceMock.receive()).andReturn(null); expect(dispatcherMock.send(messageMock)).andReturn(true).times(5); replay(globalMocks); - pollingDispatcher.setMaxMessagesPerTask(6); + pollingDispatcher.setMaxMessagesPerPoll(6); pollingDispatcher.run(); verify(globalMocks); } @@ -80,15 +96,14 @@ public class PollingDispatcherTest { expect(sourceMock.receive()).andReturn(messageMock).times(1); expect(dispatcherMock.send(messageMock)).andReturn(false).anyTimes(); replay(globalMocks); - pollingDispatcher.setMaxMessagesPerTask(10); + pollingDispatcher.setMaxMessagesPerPoll(10); pollingDispatcher.run(); verify(globalMocks); } @Test public void blockingSourceTimedOut() { - pollingDispatcher = new PollingDispatcher(sourceMock, dispatcherMock, - scheduleMock); + 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); @@ -99,8 +114,7 @@ public class PollingDispatcherTest { @Test public void blockingSourceNotTimedOut() { - pollingDispatcher = new PollingDispatcher(sourceMock, dispatcherMock, - scheduleMock); + pollingDispatcher = new PollingDispatcher(sourceMock, scheduleMock, dispatcherMock); expect(sourceMock.receive(1)).andReturn(messageMock); expect(dispatcherMock.send(messageMock)).andReturn(false); replay(globalMocks); @@ -108,4 +122,5 @@ public class PollingDispatcherTest { pollingDispatcher.run(); verify(globalMocks); } + } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java deleted file mode 100644 index f54bda1033..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java +++ /dev/null @@ -1,75 +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.endpoint; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; - -import java.util.concurrent.atomic.AtomicInteger; - -import org.junit.Test; - -import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.message.GenericMessage; -import org.springframework.integration.message.Message; -import org.springframework.integration.message.PollableSource; - -/** - * @author Mark Fisher - */ -public class SourceEndpointTests { - - @Test - public void testPolledSourceSendsToChannel() { - TestSource source = new TestSource("testing", 1); - QueueChannel channel = new QueueChannel(); - SourceEndpoint endpoint = new SourceEndpoint(source); - endpoint.setTarget(channel); - endpoint.afterPropertiesSet(); - endpoint.send(new TriggerMessage()); - Message message = channel.receive(1000); - assertNotNull("message should not be null", message); - assertEquals("testing.1", message.getPayload()); - } - - - private static class TestSource implements PollableSource { - - private String message; - - private int limit; - - private AtomicInteger count = new AtomicInteger(); - - public TestSource(String message, int limit) { - this.message = message; - this.limit = limit; - } - - public void resetCounter() { - this.count.set(0); - } - - public Message receive() { - if (count.get() >= limit) { - return null; - } - return new GenericMessage(message + "." + count.incrementAndGet()); - } - } - -}