Added 'taskAdviceChain' and 'dispatchAdviceChain' to PollingSourceEndpoint.

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java

@@ -25,7 +25,8 @@ import org.junit.Test;
 
 import org.springframework.integration.channel.DispatcherPolicy;
 import org.springframework.integration.channel.QueueChannel;
-import org.springframework.integration.dispatcher.PollingDispatcher;
+import org.springframework.integration.dispatcher.DefaultPollingDispatcher;
+import org.springframework.integration.dispatcher.PollingDispatcherTask;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.StringMessage;
 
@@ -63,12 +64,12 @@ public class ByteStreamTargetAdapterTests {
 		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
 		dispatcherPolicy.setMaxMessagesPerTask(3);
 		QueueChannel channel = new QueueChannel(5, dispatcherPolicy);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
-		dispatcher.subscribe(adapter);
+		PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null);
+		task.getDispatcher().subscribe(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
-		dispatcher.run();
+		task.run();
 		byte[] result = stream.toByteArray();
 		assertEquals(9, result.length);
 		assertEquals(1, result[0]);
@@ -82,12 +83,12 @@ public class ByteStreamTargetAdapterTests {
 		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
 		dispatcherPolicy.setMaxMessagesPerTask(2);
 		QueueChannel channel = new QueueChannel(5, dispatcherPolicy);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
-		dispatcher.subscribe(adapter);
+		PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null);
+		task.getDispatcher().subscribe(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
-		dispatcher.run();
+		task.run();
 		byte[] result = stream.toByteArray();
 		assertEquals(6, result.length);
 		assertEquals(1, result[0]);
@@ -101,12 +102,12 @@ public class ByteStreamTargetAdapterTests {
 		dispatcherPolicy.setMaxMessagesPerTask(5);
 		dispatcherPolicy.setReceiveTimeout(0);
 		QueueChannel channel = new QueueChannel(5, dispatcherPolicy);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
-		dispatcher.subscribe(adapter);
+		PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null);
+		task.getDispatcher().subscribe(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
-		dispatcher.run();
+		task.run();
 		byte[] result = stream.toByteArray();
 		assertEquals(9, result.length);
 		assertEquals(1, result[0]);
@@ -120,16 +121,16 @@ public class ByteStreamTargetAdapterTests {
 		dispatcherPolicy.setMaxMessagesPerTask(2);
 		dispatcherPolicy.setReceiveTimeout(0);
 		QueueChannel channel = new QueueChannel(5, dispatcherPolicy);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
-		dispatcher.subscribe(adapter);
+		PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null);
+		task.getDispatcher().subscribe(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
-		dispatcher.run();
+		task.run();
 		byte[] result1 = stream.toByteArray();
 		assertEquals(6, result1.length);
 		assertEquals(1, result1[0]);
-		dispatcher.run();
+		task.run();
 		byte[] result2 = stream.toByteArray();
 		assertEquals(9, result2.length);
 		assertEquals(1, result2[0]);
@@ -144,16 +145,16 @@ public class ByteStreamTargetAdapterTests {
 		dispatcherPolicy.setMaxMessagesPerTask(5);
 		dispatcherPolicy.setReceiveTimeout(0);
 		QueueChannel channel = new QueueChannel(5, dispatcherPolicy);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
-		dispatcher.subscribe(adapter);
+		PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null);
+		task.getDispatcher().subscribe(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
-		dispatcher.run();
+		task.run();
 		byte[] result1 = stream.toByteArray();
 		assertEquals(9, result1.length);
 		assertEquals(1, result1[0]);
-		dispatcher.run();
+		task.run();
 		byte[] result2 = stream.toByteArray();
 		assertEquals(9, result2.length);	
 		assertEquals(1, result2[0]);
@@ -167,16 +168,16 @@ public class ByteStreamTargetAdapterTests {
 		dispatcherPolicy.setMaxMessagesPerTask(2);
 		dispatcherPolicy.setReceiveTimeout(0);
 		QueueChannel channel = new QueueChannel(5, dispatcherPolicy);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
-		dispatcher.subscribe(adapter);
+		PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null);
+		task.getDispatcher().subscribe(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
-		dispatcher.run();
+		task.run();
 		byte[] result1 = stream.toByteArray();
 		assertEquals(6, result1.length);
 		stream.reset();
-		dispatcher.run();
+		task.run();
 		byte[] result2 = stream.toByteArray();
 		assertEquals(3, result2.length);
 		assertEquals(7, result2[0]);
@@ -190,17 +191,17 @@ public class ByteStreamTargetAdapterTests {
 		dispatcherPolicy.setMaxMessagesPerTask(2);
 		dispatcherPolicy.setReceiveTimeout(0);
 		QueueChannel channel = new QueueChannel(5, dispatcherPolicy);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
-		dispatcher.subscribe(adapter);
+		PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null);
+		task.getDispatcher().subscribe(adapter);
 		channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {4,5,6}), 0);
 		channel.send(new GenericMessage<byte[]>(new byte[] {7,8,9}), 0);
-		dispatcher.run();
+		task.run();
 		byte[] result1 = stream.toByteArray();
 		assertEquals(6, result1.length);
 		stream.write(new byte[] {123});
 		stream.flush();
-		dispatcher.run();
+		task.run();
 		byte[] result2 = stream.toByteArray();
 		assertEquals(10, result2.length);
 		assertEquals(1, result2[0]);

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java

@@ -25,7 +25,8 @@ import org.junit.Test;
 import org.springframework.integration.channel.DispatcherPolicy;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.QueueChannel;
-import org.springframework.integration.dispatcher.PollingDispatcher;
+import org.springframework.integration.dispatcher.DefaultPollingDispatcher;
+import org.springframework.integration.dispatcher.PollingDispatcherTask;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.StringMessage;
 
@@ -47,13 +48,13 @@ public class CharacterStreamTargetAdapterTests {
 		MessageChannel channel = new QueueChannel();
 		StringWriter writer = new StringWriter();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
-		dispatcher.subscribe(adapter);
+		PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null);
+		task.getDispatcher().subscribe(adapter);
 		channel.send(new StringMessage("foo"), 0);
 		channel.send(new StringMessage("bar"), 0);
-		dispatcher.run();
+		task.run();
 		assertEquals("foo", writer.toString());
-		dispatcher.run();
+		task.run();
 		assertEquals("foobar", writer.toString());
 	}
 
@@ -63,14 +64,14 @@ public class CharacterStreamTargetAdapterTests {
 		StringWriter writer = new StringWriter();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer);
 		adapter.setShouldAppendNewLine(true);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
-		dispatcher.subscribe(adapter);
+		PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null);
+		task.getDispatcher().subscribe(adapter);
 		channel.send(new StringMessage("foo"), 0);
 		channel.send(new StringMessage("bar"), 0);
-		dispatcher.run();
+		task.run();
 		String newLine = System.getProperty("line.separator");
 		assertEquals("foo" + newLine, writer.toString());
-		dispatcher.run();
+		task.run();
 		assertEquals("foo" + newLine + "bar" + newLine, writer.toString());
 	}
 
@@ -81,11 +82,11 @@ public class CharacterStreamTargetAdapterTests {
 		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
 		dispatcherPolicy.setMaxMessagesPerTask(2);
 		QueueChannel channel = new QueueChannel(5, dispatcherPolicy);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
-		dispatcher.subscribe(adapter);
+		PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null);
+		task.getDispatcher().subscribe(adapter);
 		channel.send(new StringMessage("foo"), 0);
 		channel.send(new StringMessage("bar"), 0);
-		dispatcher.run();
+		task.run();
 		assertEquals("foobar", writer.toString());
 	}
 
@@ -97,12 +98,12 @@ public class CharacterStreamTargetAdapterTests {
 		dispatcherPolicy.setMaxMessagesPerTask(10);
 		dispatcherPolicy.setReceiveTimeout(0);
 		QueueChannel channel = new QueueChannel(5, dispatcherPolicy);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
+		PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null);
+		task.getDispatcher().subscribe(adapter);		
 		adapter.setShouldAppendNewLine(true);
-		dispatcher.subscribe(adapter);
 		channel.send(new StringMessage("foo"), 0);
 		channel.send(new StringMessage("bar"), 0);
-		dispatcher.run();
+		task.run();
 		String newLine = System.getProperty("line.separator");
 		assertEquals("foo" + newLine + "bar" + newLine, writer.toString());
 	}
@@ -112,11 +113,11 @@ public class CharacterStreamTargetAdapterTests {
 		MessageChannel channel = new QueueChannel();
 		StringWriter writer = new StringWriter();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
-		dispatcher.subscribe(adapter);
+		PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null);
+		task.getDispatcher().subscribe(adapter);
 		TestObject testObject = new TestObject("foo");
 		channel.send(new GenericMessage<TestObject>(testObject));
-		dispatcher.run();
+		task.run();
 		assertEquals("foo", writer.toString());
 	}
 
@@ -128,13 +129,13 @@ public class CharacterStreamTargetAdapterTests {
 		dispatcherPolicy.setReceiveTimeout(0);
 		dispatcherPolicy.setMaxMessagesPerTask(2);
 		QueueChannel channel = new QueueChannel(5, dispatcherPolicy);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
-		dispatcher.subscribe(adapter);
+		PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null);
+		task.getDispatcher().subscribe(adapter);
 		TestObject testObject1 = new TestObject("foo");
 		TestObject testObject2 = new TestObject("bar");
 		channel.send(new GenericMessage<TestObject>(testObject1), 0);
 		channel.send(new GenericMessage<TestObject>(testObject2), 0);
-		dispatcher.run();
+		task.run();
 		assertEquals("foobar", writer.toString());
 	}
 
@@ -147,13 +148,13 @@ public class CharacterStreamTargetAdapterTests {
 		dispatcherPolicy.setMaxMessagesPerTask(2);
 		QueueChannel channel = new QueueChannel(5, dispatcherPolicy);
 		adapter.setShouldAppendNewLine(true);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
-		dispatcher.subscribe(adapter);
+		PollingDispatcherTask task = new PollingDispatcherTask(new DefaultPollingDispatcher(channel), null);
+		task.getDispatcher().subscribe(adapter);
 		TestObject testObject1 = new TestObject("foo");
 		TestObject testObject2 = new TestObject("bar");
 		channel.send(new GenericMessage<TestObject>(testObject1), 0);
 		channel.send(new GenericMessage<TestObject>(testObject2), 0);
-		dispatcher.run();
+		task.run();
 		String newLine = System.getProperty("line.separator");
 		assertEquals("foo" + newLine + "bar" + newLine, writer.toString());
 	}

spring-integration-core/ivy.xml

@@ -21,8 +21,8 @@
 
 	<dependencies>
 		<dependency org="org.junit" name="junit" rev="4.4" conf="test->default"/>
-		<dependency org="org.springframework" name="spring-context" rev="2.5.2" conf="compile->default"/>
 		<dependency org="org.springframework" name="spring-aop" rev="2.5.2" conf="compile->default"/>
+		<dependency org="org.springframework" name="spring-context" rev="2.5.2" conf="compile->default"/>
 	</dependencies>
 
 </ivy-module>

spring-integration-core/src/main/java/org/springframework/integration/bus/SubscriptionManager.java

@@ -28,7 +28,8 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.context.Lifecycle;
 import org.springframework.integration.ConfigurationException;
 import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.dispatcher.PollingDispatcher;
+import org.springframework.integration.dispatcher.DefaultPollingDispatcher;
+import org.springframework.integration.dispatcher.PollingDispatcherTask;
 import org.springframework.integration.dispatcher.SynchronousChannel;
 import org.springframework.integration.endpoint.TargetEndpoint;
 import org.springframework.integration.message.MessagingException;
@@ -55,7 +56,7 @@ public class SubscriptionManager {
 
 	private volatile Schedule defaultSchedule = new PollingSchedule(5);
 
-	private final ConcurrentMap<Schedule, PollingDispatcher> dispatchers = new ConcurrentHashMap<Schedule, PollingDispatcher>();
+	private final ConcurrentMap<Schedule, PollingDispatcherTask> dispatcherTasks = new ConcurrentHashMap<Schedule, PollingDispatcherTask>();
 
 	private final List<Lifecycle> lifecycleTargets = new CopyOnWriteArrayList<Lifecycle>();
 
@@ -119,21 +120,22 @@ public class SubscriptionManager {
 			}
 			return;
 		}
-		PollingDispatcher dispatcher = this.dispatchers.get(schedule);
-		if (dispatcher == null) {
-			dispatcher = this.dispatchers.putIfAbsent(schedule, new PollingDispatcher(this.channel, schedule));
+		PollingDispatcherTask dispatcherTask = this.dispatcherTasks.get(schedule);
+		if (dispatcherTask == null) {
+			DefaultPollingDispatcher dispatcher = new DefaultPollingDispatcher(this.channel);
+			dispatcherTask = this.dispatcherTasks.putIfAbsent(schedule, new PollingDispatcherTask(dispatcher, schedule));
 		}
-		this.dispatchers.get(schedule).subscribe(target);
-		if (dispatcher == null && this.isRunning()) {
+		this.dispatcherTasks.get(schedule).getDispatcher().subscribe(target);
+		if (dispatcherTask == null && this.isRunning()) {
 			this.scheduleDispatcherTask(schedule);
 		}
 	}
 
 	public boolean removeTarget(Target target) {
 		boolean removed = false;
-		Collection<PollingDispatcher> dispatcherValues = this.dispatchers.values();
-		for (PollingDispatcher dispatcher : dispatcherValues) {
-			removed = (removed || dispatcher.unsubscribe(target));
+		Collection<PollingDispatcherTask> dispatcherTaskValues = this.dispatcherTasks.values();
+		for (PollingDispatcherTask dispatcherTask : dispatcherTaskValues) {
+			removed = (removed || dispatcherTask.getDispatcher().unsubscribe(target));
 		}
 		return removed;
 	}
@@ -156,7 +158,7 @@ public class SubscriptionManager {
 			for (Lifecycle target : lifecycleTargets) {
 				target.start();
 			}
-			for (Schedule schedule : this.dispatchers.keySet()) {
+			for (Schedule schedule : this.dispatcherTasks.keySet()) {
 				this.scheduleDispatcherTask(schedule);
 			}
 			this.running = true;
@@ -164,9 +166,9 @@ public class SubscriptionManager {
 	}
 
 	private void scheduleDispatcherTask(Schedule schedule) {
-		PollingDispatcher dispatcher = this.dispatchers.get(schedule);
-		if (dispatcher != null) {
-			this.scheduler.schedule(dispatcher);
+		PollingDispatcherTask dispatcherTask = this.dispatcherTasks.get(schedule);
+		if (dispatcherTask != null) {
+			this.scheduler.schedule(dispatcherTask);
 		}
 	}
 

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/DefaultPollingDispatcher.java

@@ -0,0 +1,97 @@
+/*
+ * 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 java.util.LinkedList;
+import java.util.List;
+
+import org.springframework.integration.channel.DispatcherPolicy;
+import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.message.BlockingSource;
+import org.springframework.integration.message.Message;
+import org.springframework.integration.message.MessageDeliveryAware;
+import org.springframework.integration.message.MessageDeliveryException;
+import org.springframework.integration.message.PollableSource;
+import org.springframework.util.Assert;
+
+/**
+ * A subclass of {@link SimpleDispatcher} that adds message retrieval
+ * capabilities by polling a {@link PollableSource} for {@link Message Messages}.
+ * The number of messages retrieved per poll is limited by the '<em>maxMessagesPerTask</em>'
+ * property of the provided {@link DispatcherPolicy}, and the timeout for each
+ * receive call is determined by the policy's '<em>receiveTimeout</em>'
+ * property. In general, it is recommended to use a value of 1 (the default) for
+ * 'maxMessagesPerTask' whenever a significant timeout is provided. Otherwise
+ * the poller may be holding on to available messages while waiting for
+ * additional messages. Note that the 'timeout' value is only relevant if the
+ * specified source is an implementation of {@link BlockingSource}. The default
+ * timeout value is 0 indicating that the method should return immediately
+ * rather than waiting for a {@link Message} to become available.
+ * 
+ * @author Mark Fisher
+ */
+public class DefaultPollingDispatcher extends SimpleDispatcher implements PollingDispatcher {
+
+	private final PollableSource<?> source;
+
+
+	public DefaultPollingDispatcher(MessageChannel channel) {
+		this(channel, channel.getDispatcherPolicy());
+	}
+
+	public DefaultPollingDispatcher(PollableSource<?> source, DispatcherPolicy dispatcherPolicy) {
+		super(dispatcherPolicy);
+		Assert.notNull(source, "source must not be null");
+		this.source = source;
+		this.dispatcherPolicy.setReceiveTimeout(0);
+	}
+
+	@Override
+	public boolean dispatch(Message<?> message) {
+		boolean sent = super.dispatch(message);
+		if (this.source instanceof MessageDeliveryAware) {
+			if (sent) {
+				((MessageDeliveryAware) this.source).onSend(message);
+			}
+			else {
+				((MessageDeliveryAware) this.source).onFailure(new MessageDeliveryException(message, "failed to send message"));
+			}
+		}
+		return sent;
+	}
+
+	public List<Message<?>> poll() {
+		List<Message<?>> messages = new LinkedList<Message<?>>();
+		int limit = this.dispatcherPolicy.getMaxMessagesPerTask();
+		while (messages.size() < limit) {
+			Message<?> message = null;
+			long timeout = this.dispatcherPolicy.getReceiveTimeout();
+			if (this.source instanceof BlockingSource && timeout >= 0) {
+				message = ((BlockingSource<?>) this.source).receive(timeout);
+			}
+			else {
+				message = this.source.receive();
+			}
+			if (message == null) {
+				return messages;
+			}
+			messages.add(message);
+		}
+		return messages;
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java

@@ -27,4 +27,6 @@ public interface MessageDispatcher {
 
 	boolean dispatch(Message<?> message);
 
+	void setSendTimeout(long timeout);
+
 }

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcher.java

@@ -16,94 +16,12 @@
 
 package org.springframework.integration.dispatcher;
 
-import java.util.LinkedList;
-import java.util.List;
-
-import org.springframework.integration.channel.DispatcherPolicy;
-import org.springframework.integration.channel.MessageChannel;
-import org.springframework.integration.message.BlockingSource;
-import org.springframework.integration.message.Message;
-import org.springframework.integration.message.MessageDeliveryAware;
-import org.springframework.integration.message.MessageDeliveryException;
-import org.springframework.integration.message.PollableSource;
-import org.springframework.integration.scheduling.MessagingTask;
-import org.springframework.integration.scheduling.Schedule;
-import org.springframework.util.Assert;
+import org.springframework.integration.message.Poller;
+import org.springframework.integration.message.SubscribableSource;
 
 /**
- * A subclass of {@link SimpleDispatcher} that adds message retrieval
- * capabilities and may be scheduled to run as a task. It polls a source for
- * {@link Message Messages}. The number of messages retrieved per poll is
- * limited by the '<em>maxMessagesPerTask</em>' property of the provided
- * {@link DispatcherPolicy}, and the timeout for each receive call is
- * determined by the policy's '<em>receiveTimeout</em>' property. In
- * general, it is recommended to use a value of 1 (the default) for
- * 'maxMessagesPerTask' whenever a significant timeout is provided. Otherwise
- * the poller may be holding on to available messages while waiting for
- * additional messages. Note that the 'timeout' value is only relevant if the
- * specified source is an implementation of {@link BlockingSource}. The default
- * timeout value is 0 indicating that the method should return immediately
- * rather than waiting for a {@link Message} to become available.
- * 
  * @author Mark Fisher
  */
-public class PollingDispatcher extends SimpleDispatcher implements MessagingTask {
-
-	private final PollableSource<?> source;
-
-	private final Schedule schedule;
-
-
-	public PollingDispatcher(MessageChannel channel, Schedule schedule) {
-		this(channel, channel.getDispatcherPolicy(), schedule);
-	}
-
-	public PollingDispatcher(PollableSource<?> source, DispatcherPolicy dispatcherPolicy, Schedule schedule) {
-		super(dispatcherPolicy);
-		Assert.notNull(source, "source must not be null");
-		this.source = source;
-		this.schedule = schedule;
-		this.dispatcherPolicy.setReceiveTimeout(0);
-	}
-
-
-	public Schedule getSchedule() {
-		return this.schedule;
-	}
-
-	public List<Message<?>> poll() {
-		List<Message<?>> messages = new LinkedList<Message<?>>();
-		int limit = this.dispatcherPolicy.getMaxMessagesPerTask();
-		while (messages.size() < limit) {
-			Message<?> message = null;
-			long timeout = this.dispatcherPolicy.getReceiveTimeout();
-			if (this.source instanceof BlockingSource && timeout >= 0) {
-				message = ((BlockingSource<?>) this.source).receive(timeout);
-			}
-			else {
-				message = this.source.receive();
-			}
-			if (message == null) {
-				return messages;
-			}
-			messages.add(message);
-		}
-		return messages;
-	}
-
-	public void run() {
-		List<Message<?>> messages = this.poll();
-		for (Message<?> message : messages) {
-			boolean sent = this.dispatch(message);
-			if (this.source instanceof MessageDeliveryAware) {
-				if (sent) {
-					((MessageDeliveryAware) this.source).onSend(message);
-				}
-				else {
-					((MessageDeliveryAware) this.source).onFailure(new MessageDeliveryException(message, "failed to send message"));
-				}
-			}
-		}
-	}
+public interface PollingDispatcher extends Poller, MessageDispatcher, SubscribableSource {
 
 }

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/PollingDispatcherTask.java

@@ -0,0 +1,60 @@
+/*
+ * 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 java.util.List;
+
+import org.springframework.integration.message.Message;
+import org.springframework.integration.scheduling.MessagingTask;
+import org.springframework.integration.scheduling.Schedule;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link MessagingTask} that combines polling and dispatching.
+ * 
+ * @author Mark Fisher
+ */
+public class PollingDispatcherTask implements MessagingTask {
+
+	private final PollingDispatcher dispatcher;
+
+	private final Schedule schedule;
+
+
+	public PollingDispatcherTask(PollingDispatcher dispatcher, Schedule schedule) {
+		Assert.notNull(dispatcher, "dispatcher must not be null");
+		this.dispatcher = dispatcher;
+		this.schedule = schedule;
+	}
+
+
+	public PollingDispatcher getDispatcher() {
+		return this.dispatcher;
+	}
+
+	public Schedule getSchedule() {
+		return this.schedule;
+	}
+
+	public void run() {
+		List<Message<?>> messages = this.dispatcher.poll();
+		for (Message<?> message : messages) {
+			this.dispatcher.dispatch(message);
+		}
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/endpoint/PollingSourceEndpoint.java

@@ -16,9 +16,19 @@
 
 package org.springframework.integration.endpoint;
 
+import java.lang.reflect.Method;
+import java.util.List;
+
+import org.aopalliance.aop.Advice;
+
+import org.springframework.aop.framework.ProxyFactory;
+import org.springframework.aop.support.StaticMethodMatcherPointcutAdvisor;
+import org.springframework.beans.factory.InitializingBean;
 import org.springframework.integration.channel.DispatcherPolicy;
 import org.springframework.integration.channel.MessageChannel;
+import org.springframework.integration.dispatcher.DefaultPollingDispatcher;
 import org.springframework.integration.dispatcher.PollingDispatcher;
+import org.springframework.integration.dispatcher.PollingDispatcherTask;
 import org.springframework.integration.message.PollableSource;
 import org.springframework.integration.scheduling.MessagingTask;
 import org.springframework.integration.scheduling.PollingSchedule;
@@ -31,21 +41,60 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class PollingSourceEndpoint extends AbstractSourceEndpoint implements MessagingTask {
+public class PollingSourceEndpoint extends AbstractSourceEndpoint implements MessagingTask, InitializingBean {
 
 	private final DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
 
-	private final PollingDispatcher dispatcher;
+	private volatile PollingDispatcher dispatcher;
+
+	private volatile MessagingTask task;
+
+	private volatile List<Advice> taskAdviceChain;
+
+	private volatile List<Advice> dispatchAdviceChain;
 
 
 	public PollingSourceEndpoint(PollableSource<?> source, MessageChannel channel, PollingSchedule schedule) {
 		super(source, channel);
 		Assert.notNull(schedule, "schedule must not be null");
-		this.dispatcher = new PollingDispatcher(source, this.dispatcherPolicy, schedule);
+		this.dispatcher = new DefaultPollingDispatcher(source, this.dispatcherPolicy);
 		this.dispatcher.subscribe(this.getChannel());
+		this.task = new PollingDispatcherTask(this.dispatcher, schedule);
 	}
 
 
+	public void setTaskAdviceChain(List<Advice> taskAdviceChain) {
+		this.taskAdviceChain = taskAdviceChain;
+	}
+
+	public void setDispatchAdviceChain(List<Advice> dispatchAdviceChain) {
+		this.dispatchAdviceChain = dispatchAdviceChain;
+	}
+
+	public void afterPropertiesSet() {
+		this.initializeProxies();
+	}
+
+	public void initializeProxies() {
+		if (this.dispatchAdviceChain != null && this.dispatchAdviceChain.size() > 0) {
+			ProxyFactory proxyFactory = new ProxyFactory(this.dispatcher);
+			proxyFactory.setInterfaces(new Class[] { PollingDispatcher.class });
+			for (Advice advice : this.dispatchAdviceChain) {
+				proxyFactory.addAdvisor(new DispatchMethodAdvisor(advice));
+			}
+			this.dispatcher = (PollingDispatcher) proxyFactory.getProxy();
+			this.task = new PollingDispatcherTask(this.dispatcher, this.task.getSchedule());
+		}
+		if (this.taskAdviceChain != null && this.taskAdviceChain.size() > 0) {
+			ProxyFactory proxyFactory = new ProxyFactory(this.task);
+			proxyFactory.setInterfaces(new Class[] { MessagingTask.class });
+			for (Advice advice : this.taskAdviceChain) {
+				proxyFactory.addAdvice(advice);
+			}
+			this.task = (MessagingTask) proxyFactory.getProxy();
+		}
+	}
+
 	public void setMaxMessagesPerTask(int maxMessagesPerTask) {
 		this.dispatcherPolicy.setMaxMessagesPerTask(maxMessagesPerTask);
 	}
@@ -55,11 +104,25 @@ public class PollingSourceEndpoint extends AbstractSourceEndpoint implements Mes
 	}
 
 	public Schedule getSchedule() {
-		return this.dispatcher.getSchedule();
+		return this.task.getSchedule();
 	}
 
 	public void run() {
-		this.dispatcher.run();
+		this.task.run();
+	}
+
+
+	@SuppressWarnings("serial")
+	private static class DispatchMethodAdvisor extends StaticMethodMatcherPointcutAdvisor {
+
+		DispatchMethodAdvisor(Advice advice) {
+			super(advice);
+		}
+
+		@SuppressWarnings("unchecked")
+		public boolean matches(Method method, Class targetClass) {
+			return method.getName().equals("dispatch");
+		}
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/message/Poller.java

@@ -0,0 +1,28 @@
+/*
+ * 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.message;
+
+import java.util.List;
+
+/**
+ * @author Mark Fisher
+ */
+public interface Poller {
+
+	 List<Message<?>> poll();
+
+}

spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DefaultPollingDispatcherTests.java

@@ -33,14 +33,14 @@ import org.springframework.integration.message.StringMessage;
 /**
  * @author Mark Fisher
  */
-public class PollingDispatcherTests {
+public class DefaultPollingDispatcherTests {
 
 	@Test
 	public void testSingleMessagePerRetrieval() {
 		DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
 		dispatcherPolicy.setReceiveTimeout(0);
 		MessageChannel channel = new QueueChannel(5, dispatcherPolicy);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
+		DefaultPollingDispatcher dispatcher = new DefaultPollingDispatcher(channel);
 		Collection<Message<?>> results = dispatcher.poll();
 		assertTrue(results.isEmpty());
 		channel.send(new StringMessage("test1"), 0);
@@ -59,7 +59,7 @@ public class PollingDispatcherTests {
 		dispatcherPolicy.setReceiveTimeout(0);
 		dispatcherPolicy.setMaxMessagesPerTask(2);
 		MessageChannel channel = new QueueChannel(5, dispatcherPolicy);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
+		DefaultPollingDispatcher dispatcher = new DefaultPollingDispatcher(channel);
 		Collection<Message<?>> results = dispatcher.poll();
 		assertTrue(results.isEmpty());
 		channel.send(new StringMessage("test1"), 0);
@@ -81,7 +81,7 @@ public class PollingDispatcherTests {
 		dispatcherPolicy.setReceiveTimeout(0);
 		dispatcherPolicy.setMaxMessagesPerTask(1);
 		MessageChannel channel = new QueueChannel(5, dispatcherPolicy);
-		PollingDispatcher dispatcher = new PollingDispatcher(channel, null);
+		DefaultPollingDispatcher dispatcher = new DefaultPollingDispatcher(channel);
 		Collection<Message<?>> results = dispatcher.poll();
 		assertTrue(results.isEmpty());
 		channel.send(new StringMessage("test1"), 0);

spring-integration-core/src/test/java/org/springframework/integration/endpoint/PollingSourceEndpointTests.java

@@ -20,10 +20,18 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.aopalliance.aop.Advice;
+import org.aopalliance.intercept.MethodInterceptor;
+import org.aopalliance.intercept.MethodInvocation;
+
 import org.junit.Test;
 
+import org.springframework.aop.MethodBeforeAdvice;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
@@ -90,6 +98,118 @@ public class PollingSourceEndpointTests {
 		assertNull("message should be null", message4);
 	}
 
+	@Test
+	public void testTaskAdviceChain() {
+		TestSource source = new TestSource("testing", 3);
+		QueueChannel channel = new QueueChannel();
+		PollingSchedule schedule = new PollingSchedule(1000);
+		schedule.setInitialDelay(10000);
+		PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
+		final StringBuffer buffer = new StringBuffer();
+		List<Advice> taskAdviceChain = new ArrayList<Advice>();
+		taskAdviceChain.add(new MethodBeforeAdvice() {
+			public void before(Method method, Object[] args, Object target) throws Throwable {
+				buffer.append(1);
+			}
+		});
+		taskAdviceChain.add(new MethodInterceptor() {
+			public Object invoke(MethodInvocation invocation) throws Throwable {
+				buffer.append(2);
+				Object retval = invocation.proceed();
+				buffer.append(4);
+				return retval;
+			}
+		});
+		taskAdviceChain.add(new MethodBeforeAdvice() {
+			public void before(Method method, Object[] args, Object target) throws Throwable {
+				buffer.append(3);
+			}
+		});
+		endpoint.setTaskAdviceChain(taskAdviceChain);
+		endpoint.afterPropertiesSet();
+		endpoint.setMaxMessagesPerTask(5);
+		endpoint.run();
+		assertEquals("1234", buffer.toString());
+	}
+
+	@Test
+	public void testDispatchAdviceChain() {
+		TestSource source = new TestSource("testing", 2);
+		QueueChannel channel = new QueueChannel();
+		PollingSchedule schedule = new PollingSchedule(1000);
+		schedule.setInitialDelay(10000);
+		PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
+		final StringBuffer buffer = new StringBuffer();
+		List<Advice> dispatchAdviceChain = new ArrayList<Advice>();
+		dispatchAdviceChain.add(new MethodBeforeAdvice() {
+			public void before(Method method, Object[] args, Object target) throws Throwable {
+				buffer.append("a");
+			}
+		});
+		dispatchAdviceChain.add(new MethodInterceptor() {
+			public Object invoke(MethodInvocation invocation) throws Throwable {
+				buffer.append("b");
+				Object retval = invocation.proceed();
+				buffer.append("d");
+				return retval;
+			}
+		});
+		dispatchAdviceChain.add(new MethodBeforeAdvice() {
+			public void before(Method method, Object[] args, Object target) throws Throwable {
+				buffer.append("c");
+			}
+		});
+		endpoint.setDispatchAdviceChain(dispatchAdviceChain);
+		endpoint.afterPropertiesSet();
+		endpoint.setMaxMessagesPerTask(5);
+		endpoint.run();
+		assertEquals("abcdabcd", buffer.toString());
+	}
+
+	@Test
+	public void testTaskAndDispatchAdviceChains() {
+		TestSource source = new TestSource("testing", 3);
+		QueueChannel channel = new QueueChannel();
+		PollingSchedule schedule = new PollingSchedule(1000);
+		schedule.setInitialDelay(10000);
+		PollingSourceEndpoint endpoint = new PollingSourceEndpoint(source, channel, schedule);
+		List<Advice> dispatchAdviceChain = new ArrayList<Advice>();
+		List<Advice> taskAdviceChain = new ArrayList<Advice>();
+		final StringBuffer buffer = new StringBuffer();
+		dispatchAdviceChain.add(new MethodBeforeAdvice() {
+			public void before(Method method, Object[] args, Object target) throws Throwable {
+				buffer.append("a");
+			}
+		});
+		dispatchAdviceChain.add(new MethodInterceptor() {
+			public Object invoke(MethodInvocation invocation) throws Throwable {
+				buffer.append("b");
+				Object retval = invocation.proceed();
+				buffer.append("c");
+				return retval;
+			}
+		});
+		taskAdviceChain.add(new MethodInterceptor() {
+			public Object invoke(MethodInvocation invocation) throws Throwable {
+				buffer.append(1);
+				Object retval = invocation.proceed();
+				buffer.append(3);
+				return retval;
+			}
+		});
+		taskAdviceChain.add(new MethodBeforeAdvice() {
+			public void before(Method method, Object[] args, Object target) throws Throwable {
+				buffer.append(2);
+			}
+		});
+		endpoint.setTaskAdviceChain(taskAdviceChain);
+		endpoint.setDispatchAdviceChain(dispatchAdviceChain);
+		endpoint.afterPropertiesSet();
+		endpoint.setMaxMessagesPerTask(5);
+		endpoint.run();
+		assertEquals("12abcabcabc3", buffer.toString());
+	}
+
 
 	private static class TestSource implements PollableSource<String> {