Removed EndpointTrigger, EndpointPoller, and EndpointVisitor. MessageBus now schedules PollingDispatchers for endpoints.

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<EndpointTrigger> endpointTriggers = new CopyOnWriteArraySet<EndpointTrigger>();
+	private final Set<PollingDispatcher> pollingDispatchers = new CopyOnWriteArraySet<PollingDispatcher>();
 
 	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) {

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.
+	 * 
+	 * <p>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;
 	}
 
 }

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);
 	}
 

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointPoller.java

@@ -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);
-	}
-
-}

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointTrigger.java

@@ -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<EndpointPoller> {
-
-		public Message<EndpointPoller> receive() {
-			return new TriggerMessage();
-		}
-	}
-
-}

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/EndpointVisitor.java

@@ -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);
-
-}

org.springframework.integration/src/main/java/org/springframework/integration/endpoint/TriggerMessage.java

@@ -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<EndpointPoller> {
-
-	public TriggerMessage() {
-		super(new EndpointPoller());
-	}
-
-}

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);
 	}
+
 }

org.springframework.integration/src/test/java/org/springframework/integration/endpoint/SourceEndpointTests.java

@@ -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<String> {
-
-		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<String> receive() {
-			if (count.get() >= limit) {
-				return null;
-			}
-			return new GenericMessage<String>(message + "." + count.incrementAndGet());
-		}
-	}
-
-}