Major refactoring of message bus, scheduling, and dispatching.

spring-integration-core/src/test/java/org/springframework/integration/adapter/AdapterTests.java

@@ -20,6 +20,8 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 
 import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import org.junit.Test;
 
@@ -35,14 +37,13 @@ public class AdapterTests {
 	public void testAdaptersWithBeanDefinitions() throws IOException, InterruptedException {
 		AbstractApplicationContext context = new ClassPathXmlApplicationContext("adapterTests.xml", this.getClass());
 		TestSink sink = (TestSink) context.getBean("sink");
+		CountDownLatch latch = new CountDownLatch(1);
+		sink.setLatch(latch);
 		assertNull(sink.get());
 		context.start();
 		String result = null;
-		int attempts = 0;
-		while (result == null && attempts++ < 100) {
-			Thread.sleep(5);
-			result = sink.get();
-		}
+		latch.await(3000, TimeUnit.MILLISECONDS);
+		result = sink.get();
 		assertNotNull(result);
 		context.close();
 	}
@@ -51,14 +52,13 @@ public class AdapterTests {
 	public void testAdaptersWithNamespace() throws IOException, InterruptedException {
 		AbstractApplicationContext context = new ClassPathXmlApplicationContext("adapterTestsWithNamespace.xml", this.getClass());
 		TestSink sink = (TestSink) context.getBean("sink");
+		CountDownLatch latch = new CountDownLatch(1);
+		sink.setLatch(latch);
 		assertNull(sink.get());
 		context.start();
 		String result = null;
-		int attempts = 0;
-		while (result == null && attempts++ < 100) {
-			Thread.sleep(5);
-			result = sink.get();
-		}
+		latch.await(3000, TimeUnit.MILLISECONDS);
+		result = sink.get();
 		assertNotNull(result);
 		context.close();
 	}

spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java

@@ -44,7 +44,6 @@ public class PollingSourceAdapterTests {
 		adapter.setChannel(channel);
 		adapter.setPeriod(100);
 		adapter.start();
-		adapter.dispatch();
 		Message<?> message = channel.receive();
 		assertNotNull("message should not be null", message);
 		assertEquals("testing.1", message.getPayload());
@@ -56,18 +55,20 @@ public class PollingSourceAdapterTests {
 		SimpleChannel channel = new SimpleChannel(1);
 		PollingSourceAdapter<String> adapter = new PollingSourceAdapter<String>(source);
 		adapter.setChannel(channel);
-		adapter.setPeriod(500);
+		adapter.setInitialDelay(10000);
 		adapter.setSendTimeout(10);
 		adapter.start();
-		adapter.dispatch();
-		adapter.dispatch();
+		adapter.processMessages();
+		adapter.processMessages();
+		adapter.stop();
 		Message<?> message1 = channel.receive();
 		assertNotNull("message should not be null", message1);
 		assertEquals("testing.1", message1.getPayload());
 		Message<?> message2 = channel.receive(0);
 		assertNull("second message should be null", message2);
-		adapter.dispatch();
-		Message<?> message3 = channel.receive(0);
+		adapter.start();
+		adapter.processMessages();
+		Message<?> message3 = channel.receive(100);
 		assertNotNull("third message should not be null", message3);
 		assertEquals("testing.3", message3.getPayload());
 	}
@@ -78,10 +79,10 @@ public class PollingSourceAdapterTests {
 		SimpleChannel channel = new SimpleChannel();
 		PollingSourceAdapter<String> adapter = new PollingSourceAdapter<String>(source);
 		adapter.setChannel(channel);
-		adapter.setPeriod(1000);
+		adapter.setInitialDelay(10000);
 		adapter.setMaxMessagesPerTask(5);
 		adapter.start();
-		adapter.dispatch();
+		adapter.processMessages();
 		Message<?> message1 = channel.receive(0);
 		assertNotNull("message should not be null", message1);
 		assertEquals("testing.1", message1.getPayload());
@@ -104,7 +105,7 @@ public class PollingSourceAdapterTests {
 		adapter.setPeriod(1000);
 		adapter.setMaxMessagesPerTask(2);
 		adapter.start();
-		adapter.dispatch();
+		adapter.processMessages();
 	}
 
 

spring-integration-core/src/test/java/org/springframework/integration/adapter/TestSink.java

@@ -16,6 +16,8 @@
 
 package org.springframework.integration.adapter;
 
+import java.util.concurrent.CountDownLatch;
+
 /**
  * @author Mark Fisher
  */
@@ -23,6 +25,13 @@ public class TestSink {
 
 	private String result;
 
+	private CountDownLatch latch;
+
+
+	public void setLatch(CountDownLatch latch) {
+		this.latch = latch;
+	}
+
 	public void validMethod(String s) {
 	}
 
@@ -34,6 +43,9 @@ public class TestSink {
 	}
 
 	public void store(String s) {
+		if (this.latch != null) {
+			this.latch.countDown();
+		}
 		this.result = s;
 	}
 

spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml

@@ -34,14 +34,9 @@
 
 	<bean id="sink" class="org.springframework.integration.adapter.TestSink"/>
 
-	<bean id="endpoint" class="org.springframework.integration.endpoint.GenericMessageEndpoint">
+	<bean id="endpoint" class="org.springframework.integration.endpoint.DefaultMessageEndpoint">
 		<property name="inputChannelName" value="inputChannel"/>
 		<property name="defaultOutputChannelName" value="outputChannel"/>
-		<property name="consumerPolicy">
-			<bean class="org.springframework.integration.bus.ConsumerPolicy">
-				<property name="receiveTimeout" value="100"/>
-			</bean>
-		</property>
 	</bean>
 
 </beans>

spring-integration-core/src/test/java/org/springframework/integration/adapter/jms/config/JmsSourceAdapterParserTests.java

@@ -39,10 +39,10 @@ public class JmsSourceAdapterParserTests {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
 				"pollingAdapterWithJmsTemplate.xml", this.getClass());
 		context.start();
-		MessageChannel channel = (MessageChannel) context.getBean("channel");
 		JmsPollingSourceAdapter adapter = (JmsPollingSourceAdapter) context.getBean("adapter");
-		adapter.dispatch();
-		Message<?> message = channel.receive(100);
+		adapter.processMessages();
+		MessageChannel channel = (MessageChannel) context.getBean("channel");
+		Message<?> message = channel.receive(500);
 		assertNotNull("message should not be null", message);
 		assertEquals("polling-test", message.getPayload());
 		context.stop();
@@ -53,10 +53,10 @@ public class JmsSourceAdapterParserTests {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
 				"pollingAdapterWithConnectionFactoryAndDestination.xml", this.getClass());
 		context.start();
-		MessageChannel channel = (MessageChannel) context.getBean("channel");
 		JmsPollingSourceAdapter adapter = (JmsPollingSourceAdapter) context.getBean("adapter");
-		adapter.dispatch();
-		Message<?> message = channel.receive(100);
+		adapter.processMessages();
+		MessageChannel channel = (MessageChannel) context.getBean("channel");
+		Message<?> message = channel.receive(500);
 		assertNotNull("message should not be null", message);
 		assertEquals("polling-test", message.getPayload());
 		context.stop();
@@ -67,10 +67,10 @@ public class JmsSourceAdapterParserTests {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
 				"pollingAdapterWithConnectionFactoryAndDestinationName.xml", this.getClass());
 		context.start();
-		MessageChannel channel = (MessageChannel) context.getBean("channel");
 		JmsPollingSourceAdapter adapter = (JmsPollingSourceAdapter) context.getBean("adapter");
-		adapter.dispatch();
-		Message<?> message = channel.receive(100);
+		adapter.processMessages();
+		MessageChannel channel = (MessageChannel) context.getBean("channel");
+		Message<?> message = channel.receive(500);
 		assertNotNull("message should not be null", message);
 		assertEquals("polling-test", message.getPayload());
 		context.stop();
@@ -81,10 +81,10 @@ public class JmsSourceAdapterParserTests {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
 				"messageDrivenAdapterWithConnectionFactoryAndDestination.xml", this.getClass());
 		context.start();
-		MessageChannel channel = (MessageChannel) context.getBean("channel");
 		JmsMessageDrivenSourceAdapter adapter = (JmsMessageDrivenSourceAdapter) context.getBean("adapter");
 		assertEquals(JmsMessageDrivenSourceAdapter.class, adapter.getClass());
-		Message<?> message = channel.receive(100);
+		MessageChannel channel = (MessageChannel) context.getBean("channel");
+		Message<?> message = channel.receive(3000);
 		assertNotNull("message should not be null", message);
 		assertEquals("message-driven-test", message.getPayload());
 		context.stop();
@@ -95,10 +95,10 @@ public class JmsSourceAdapterParserTests {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
 				"messageDrivenAdapterWithConnectionFactoryAndDestinationName.xml", this.getClass());
 		context.start();
-		MessageChannel channel = (MessageChannel) context.getBean("channel");
 		JmsMessageDrivenSourceAdapter adapter = (JmsMessageDrivenSourceAdapter) context.getBean("adapter");
 		assertEquals(JmsMessageDrivenSourceAdapter.class, adapter.getClass());
-		Message<?> message = channel.receive(100);
+		MessageChannel channel = (MessageChannel) context.getBean("channel");
+		Message<?> message = channel.receive(3000);
 		assertNotNull("message should not be null", message);
 		assertEquals("message-driven-test", message.getPayload());
 		context.stop();

spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapterTests.java

@@ -40,7 +40,7 @@ public class ByteStreamSourceAdapterTests {
 		ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream);
 		adapter.setChannel(channel);
 		adapter.start();
-		int count = adapter.dispatch();
+		int count = adapter.processMessages();
 		assertEquals(1, count);
 		Message<?> message1 = channel.receive(0);
 		byte[] payload = (byte[]) message1.getPayload();
@@ -50,7 +50,7 @@ public class ByteStreamSourceAdapterTests {
 		assertEquals(3, payload[2]);
 		Message<?> message2 = channel.receive(0);
 		assertNull(message2);
-		adapter.dispatch();
+		adapter.processMessages();
 		Message<?> message3 = channel.receive(0);
 		assertNull(message3);
 	}
@@ -65,7 +65,7 @@ public class ByteStreamSourceAdapterTests {
 		adapter.setBytesPerMessage(8);
 		adapter.setMaxMessagesPerTask(5);
 		adapter.start();
-		int count = adapter.dispatch();
+		int count = adapter.processMessages();
 		assertEquals(1, count);
 		Message<?> message1 = channel.receive(0);
 		assertEquals(8, ((byte[]) message1.getPayload()).length);
@@ -80,10 +80,11 @@ public class ByteStreamSourceAdapterTests {
 		MessageChannel channel = new SimpleChannel();
 		ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream);
 		adapter.setBytesPerMessage(4);
+		adapter.setInitialDelay(10000);
 		adapter.setMaxMessagesPerTask(1);
 		adapter.setChannel(channel);
 		adapter.start();
-		int count = adapter.dispatch();
+		int count = adapter.processMessages();
 		assertEquals(1, count);
 		Message<?> message1 = channel.receive(0);
 		byte[] bytes1 = (byte[]) message1.getPayload();
@@ -91,7 +92,7 @@ public class ByteStreamSourceAdapterTests {
 		assertEquals(0, bytes1[0]);
 		Message<?> message2 = channel.receive(0);
 		assertNull(message2);
-		adapter.dispatch();
+		adapter.processMessages();
 		Message<?> message3 = channel.receive(0);
 		byte[] bytes3 = (byte[]) message3.getPayload();
 		assertEquals(4, bytes3.length);
@@ -104,11 +105,12 @@ public class ByteStreamSourceAdapterTests {
 		ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
 		MessageChannel channel = new SimpleChannel();
 		ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream);
+		adapter.setInitialDelay(10000);
 		adapter.setChannel(channel);
 		adapter.setBytesPerMessage(4);
 		adapter.setMaxMessagesPerTask(5);
 		adapter.start();
-		int count = adapter.dispatch();
+		int count = adapter.processMessages();
 		assertEquals(2, count);
 		Message<?> message1 = channel.receive(0);
 		byte[] bytes1 = (byte[]) message1.getPayload();
@@ -128,17 +130,18 @@ public class ByteStreamSourceAdapterTests {
 		ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
 		MessageChannel channel = new SimpleChannel();
 		ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream);
+		adapter.setInitialDelay(10000);
 		adapter.setBytesPerMessage(4);
 		adapter.setMaxMessagesPerTask(1);
 		adapter.setChannel(channel);
 		adapter.start();
-		int count = adapter.dispatch();
+		int count = adapter.processMessages();
 		assertEquals(1, count);
 		Message<?> message1 = channel.receive(0);
 		assertEquals(4, ((byte[]) message1.getPayload()).length);
 		Message<?> message2 = channel.receive(0);
 		assertNull(message2);
-		adapter.dispatch();
+		adapter.processMessages();
 		Message<?> message3 = channel.receive(0);
 		assertEquals(2, ((byte[]) message3.getPayload()).length);
 	}
@@ -149,18 +152,19 @@ public class ByteStreamSourceAdapterTests {
 		ByteArrayInputStream stream = new ByteArrayInputStream(bytes);
 		MessageChannel channel = new SimpleChannel();
 		ByteStreamSourceAdapter adapter = new ByteStreamSourceAdapter(stream);
+		adapter.setInitialDelay(10000);
 		adapter.setBytesPerMessage(4);
 		adapter.setShouldTruncate(false);
 		adapter.setMaxMessagesPerTask(1);
 		adapter.setChannel(channel);
 		adapter.start();
-		int count = adapter.dispatch();
+		int count = adapter.processMessages();
 		assertEquals(1, count);
 		Message<?> message1 = channel.receive(0);
 		assertEquals(4, ((byte[]) message1.getPayload()).length);
 		Message<?> message2 = channel.receive(0);
 		assertNull(message2);
-		adapter.dispatch();
+		adapter.processMessages();
 		Message<?> message3 = channel.receive(0);
 		assertEquals(4, ((byte[]) message3.getPayload()).length);
 		assertEquals(0, ((byte[]) message3.getPayload())[3]);

spring-integration-core/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java

@@ -40,13 +40,13 @@ public class CharacterStreamSourceAdapterTests {
 		CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(stream);
 		adapter.setChannel(channel);
 		adapter.start();
-		int count = adapter.dispatch();
+		int count = adapter.processMessages();
 		assertEquals(1, count);
 		Message<?> message1 = channel.receive(0);
 		assertEquals("test", message1.getPayload());
 		Message<?> message2 = channel.receive(0);
 		assertNull(message2);
-		adapter.dispatch();
+		adapter.processMessages();
 		Message<?> message3 = channel.receive(0);
 		assertNull(message3);
 	}
@@ -60,7 +60,7 @@ public class CharacterStreamSourceAdapterTests {
 		adapter.setChannel(channel);
 		adapter.setMaxMessagesPerTask(5);
 		adapter.start();
-		int count = adapter.dispatch();
+		int count = adapter.processMessages();
 		assertEquals(1, count);
 		Message<?> message1 = channel.receive(0);
 		assertEquals("test", message1.getPayload());
@@ -74,16 +74,17 @@ public class CharacterStreamSourceAdapterTests {
 		ByteArrayInputStream stream = new ByteArrayInputStream(s.getBytes());
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(stream);
+		adapter.setInitialDelay(10000);
 		adapter.setMaxMessagesPerTask(1);
 		adapter.setChannel(channel);
 		adapter.start();
-		int count = adapter.dispatch();
+		int count = adapter.processMessages();
 		assertEquals(1, count);
 		Message<?> message1 = channel.receive(0);
 		assertEquals("test1", message1.getPayload());
 		Message<?> message2 = channel.receive(0);
 		assertNull(message2);
-		adapter.dispatch();
+		adapter.processMessages();
 		Message<?> message3 = channel.receive(0);
 		assertEquals("test2", message3.getPayload());
 	}
@@ -97,7 +98,7 @@ public class CharacterStreamSourceAdapterTests {
 		adapter.setChannel(channel);
 		adapter.setMaxMessagesPerTask(5);
 		adapter.start();
-		int count = adapter.dispatch();
+		int count = adapter.processMessages();
 		assertEquals(2, count);
 		Message<?> message1 = channel.receive(0);
 		assertEquals("test1", message1.getPayload());

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

@@ -22,10 +22,10 @@ import java.io.ByteArrayOutputStream;
 
 import org.junit.Test;
 
-import org.springframework.integration.bus.ConsumerPolicy;
-import org.springframework.integration.bus.ChannelPollingMessageDispatcher;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.SimpleChannel;
+import org.springframework.integration.dispatcher.ChannelPollingMessageRetriever;
+import org.springframework.integration.dispatcher.DispatcherTask;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.StringMessage;
 
@@ -40,12 +40,10 @@ public class CharacterStreamTargetAdapterTests {
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
 		adapter.setChannel(channel);
-		ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy();
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
-		dispatcher.addHandler(adapter);
-		dispatcher.start();
+		DispatcherTask dispatcherTask = new DispatcherTask(channel);
+		dispatcherTask.addHandler(adapter);
 		channel.send(new StringMessage("foo"));
-		int count = dispatcher.dispatch();
+		int count = dispatcherTask.dispatch();
 		assertEquals(1, count);
 		String result = new String(stream.toByteArray());
 		assertEquals("foo", result);
@@ -57,16 +55,14 @@ public class CharacterStreamTargetAdapterTests {
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
 		adapter.setChannel(channel);
-		ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy();
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
-		dispatcher.addHandler(adapter);
-		dispatcher.start();
+		DispatcherTask dispatcherTask = new DispatcherTask(channel);
+		dispatcherTask.addHandler(adapter);
 		channel.send(new StringMessage("foo"));
 		channel.send(new StringMessage("bar"));
-		assertEquals(1, dispatcher.dispatch());
+		assertEquals(1, dispatcherTask.dispatch());
 		String result1 = new String(stream.toByteArray());
 		assertEquals("foo", result1);
-		assertEquals(1, dispatcher.dispatch());
+		assertEquals(1, dispatcherTask.dispatch());
 		String result2 = new String(stream.toByteArray());
 		assertEquals("foobar", result2);
 	}
@@ -78,17 +74,15 @@ public class CharacterStreamTargetAdapterTests {
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
 		adapter.setChannel(channel);
 		adapter.setShouldAppendNewLine(true);
-		ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy();
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
-		dispatcher.addHandler(adapter);
-		dispatcher.start();
+		DispatcherTask dispatcherTask = new DispatcherTask(channel);
+		dispatcherTask.addHandler(adapter);
 		channel.send(new StringMessage("foo"));
 		channel.send(new StringMessage("bar"));
-		assertEquals(1, dispatcher.dispatch());
+		assertEquals(1, dispatcherTask.dispatch());
 		String result1 = new String(stream.toByteArray());
 		String newLine = System.getProperty("line.separator");
 		assertEquals("foo" + newLine, result1);
-		assertEquals(1, dispatcher.dispatch());
+		assertEquals(1, dispatcherTask.dispatch());
 		String result2 = new String(stream.toByteArray());
 		assertEquals("foo" + newLine + "bar" + newLine, result2);
 	}
@@ -99,14 +93,13 @@ public class CharacterStreamTargetAdapterTests {
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
 		adapter.setChannel(channel);
-		ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy();
-		policy.setMaxMessagesPerTask(2);
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
-		dispatcher.addHandler(adapter);
-		dispatcher.start();
+		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
+		retriever.setMaxMessagesPerTask(2);
+		DispatcherTask dispatcherTask = new DispatcherTask(retriever);
+		dispatcherTask.addHandler(adapter);
 		channel.send(new StringMessage("foo"));
 		channel.send(new StringMessage("bar"));
-		assertEquals(2, dispatcher.dispatch());
+		assertEquals(2, dispatcherTask.dispatch());
 		String result = new String(stream.toByteArray());
 		assertEquals("foobar", result);
 	}
@@ -118,15 +111,14 @@ public class CharacterStreamTargetAdapterTests {
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
 		adapter.setChannel(channel);
 		adapter.setShouldAppendNewLine(true);
-		ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy();
-		policy.setReceiveTimeout(0);
-		policy.setMaxMessagesPerTask(10);
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
-		dispatcher.addHandler(adapter);
-		dispatcher.start();
+		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
+		retriever.setReceiveTimeout(0);
+		retriever.setMaxMessagesPerTask(10);
+		DispatcherTask dispatcherTask = new DispatcherTask(retriever);
+		dispatcherTask.addHandler(adapter);
 		channel.send(new StringMessage("foo"));
 		channel.send(new StringMessage("bar"));
-		assertEquals(2, dispatcher.dispatch());
+		assertEquals(2, dispatcherTask.dispatch());
 		String result = new String(stream.toByteArray());
 		String newLine = System.getProperty("line.separator");
 		assertEquals("foo" + newLine + "bar" + newLine, result);
@@ -138,13 +130,11 @@ public class CharacterStreamTargetAdapterTests {
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
 		adapter.setChannel(channel);
-		ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy();
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
-		dispatcher.addHandler(adapter);
-		dispatcher.start();
+		DispatcherTask dispatcherTask = new DispatcherTask(channel);
+		dispatcherTask.addHandler(adapter);
 		TestObject testObject = new TestObject("foo");
 		channel.send(new GenericMessage<TestObject>(testObject));
-		int count = dispatcher.dispatch();
+		int count = dispatcherTask.dispatch();
 		assertEquals(1, count);
 		String result = new String(stream.toByteArray());
 		assertEquals("foo", result);
@@ -156,17 +146,16 @@ public class CharacterStreamTargetAdapterTests {
 		MessageChannel channel = new SimpleChannel();
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
 		adapter.setChannel(channel);
-		ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy();
-		policy.setReceiveTimeout(0);
-		policy.setMaxMessagesPerTask(2);
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
-		dispatcher.addHandler(adapter);
-		dispatcher.start();
+		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
+		retriever.setReceiveTimeout(0);
+		retriever.setMaxMessagesPerTask(2);
+		DispatcherTask dispatcherTask = new DispatcherTask(retriever);
+		dispatcherTask.addHandler(adapter);
 		TestObject testObject1 = new TestObject("foo");
 		TestObject testObject2 = new TestObject("bar");
 		channel.send(new GenericMessage<TestObject>(testObject1));
 		channel.send(new GenericMessage<TestObject>(testObject2));
-		assertEquals(2, dispatcher.dispatch());
+		assertEquals(2, dispatcherTask.dispatch());
 		String result = new String(stream.toByteArray());
 		assertEquals("foobar", result);
 	}
@@ -178,17 +167,16 @@ public class CharacterStreamTargetAdapterTests {
 		CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(stream);
 		adapter.setChannel(channel);
 		adapter.setShouldAppendNewLine(true);
-		ConsumerPolicy policy = ConsumerPolicy.newEventDrivenPolicy();
-		policy.setReceiveTimeout(0);
-		policy.setMaxMessagesPerTask(2);
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
-		dispatcher.addHandler(adapter);
-		dispatcher.start();
+		ChannelPollingMessageRetriever retriever = new ChannelPollingMessageRetriever(channel);
+		retriever.setReceiveTimeout(0);
+		retriever.setMaxMessagesPerTask(2);
+		DispatcherTask dispatcherTask = new DispatcherTask(retriever);
+		dispatcherTask.addHandler(adapter);
 		TestObject testObject1 = new TestObject("foo");
 		TestObject testObject2 = new TestObject("bar");
 		channel.send(new GenericMessage<TestObject>(testObject1));
 		channel.send(new GenericMessage<TestObject>(testObject2));
-		assertEquals(2, dispatcher.dispatch());
+		assertEquals(2, dispatcherTask.dispatch());
 		String result = new String(stream.toByteArray());
 		String newLine = System.getProperty("line.separator");
 		assertEquals("foo" + newLine + "bar" + newLine, result);

spring-integration-core/src/test/java/org/springframework/integration/bus/DefaultMessageDispatcherTests.java

@@ -17,6 +17,8 @@
 package org.springframework.integration.bus;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
 
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -26,15 +28,20 @@ import org.junit.Test;
 
 import org.springframework.integration.MessageDeliveryException;
 import org.springframework.integration.channel.SimpleChannel;
-import org.springframework.integration.endpoint.GenericMessageEndpoint;
+import org.springframework.integration.dispatcher.DefaultMessageDispatcher;
+import org.springframework.integration.dispatcher.MessageHandlerRejectedExecutionException;
+import org.springframework.integration.endpoint.DefaultMessageEndpoint;
+import org.springframework.integration.message.ErrorMessage;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.StringMessage;
 import org.springframework.integration.message.selector.PayloadTypeSelector;
+import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
+import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler;
 
 /**
  * @author Mark Fisher
  */
-public class ChannelPollingMessageDispatcherTests {
+public class DefaultMessageDispatcherTests {
 
 	@Test
 	public void testNonBroadcastingDispatcherSendsToExactlyOneEndpoint() throws InterruptedException {
@@ -43,14 +50,15 @@ public class ChannelPollingMessageDispatcherTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		TestEndpoint endpoint1 = new TestEndpoint(counter1, latch);
 		TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);		
-		ConsumerPolicy policy = new ConsumerPolicy();
 		SimpleChannel channel = new SimpleChannel();
 		channel.send(new StringMessage(1, "test"));
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
+		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1));
 		dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1));
+		SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler();
+		scheduler.start();
+		dispatcher.setMessagingTaskScheduler(scheduler);
 		dispatcher.start();
-		dispatcher.dispatch();
 		latch.await(100, TimeUnit.MILLISECONDS);
 		assertEquals("exactly one endpoint should have received message", 1, counter1.get() + counter2.get());
 	}
@@ -62,15 +70,13 @@ public class ChannelPollingMessageDispatcherTests {
 		final CountDownLatch latch = new CountDownLatch(2);
 		TestEndpoint endpoint1 = new TestEndpoint(counter1, latch);
 		TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);		
-		ConsumerPolicy policy = new ConsumerPolicy();
 		SimpleChannel channel = new SimpleChannel();
+		channel.setBroadcaster(true);
 		channel.send(new StringMessage(1, "test"));
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
-		dispatcher.setBroadcast(true);
+		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1));
 		dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1));
 		dispatcher.start();
-		dispatcher.dispatch();
 		latch.await(100, TimeUnit.MILLISECONDS);
 		assertEquals("both endpoints should have received message", 2, counter1.get() + counter2.get());
 	}
@@ -84,10 +90,9 @@ public class ChannelPollingMessageDispatcherTests {
 		TestEndpoint endpoint1 = new TestEndpoint(counter1, latch);
 		TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);
 		TestEndpoint endpoint3 = new TestEndpoint(counter3, latch);
-		ConsumerPolicy policy = new ConsumerPolicy();
 		SimpleChannel channel = new SimpleChannel();
 		channel.send(new StringMessage(1, "test"));
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
+		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1) {
 			@Override
 			public void start() {
@@ -96,7 +101,6 @@ public class ChannelPollingMessageDispatcherTests {
 		dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1));
 		dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1));
 		dispatcher.start();
-		dispatcher.dispatch();
 		latch.await(100, TimeUnit.MILLISECONDS);
 		assertEquals("inactive endpoint should not have received message", 0, counter1.get());
 		assertEquals("exactly one endpoint should have received message", 1, counter2.get() + counter3.get());
@@ -111,11 +115,10 @@ public class ChannelPollingMessageDispatcherTests {
 		TestEndpoint endpoint1 = new TestEndpoint(counter1, latch);
 		TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);
 		TestEndpoint endpoint3 = new TestEndpoint(counter3, latch);
-		ConsumerPolicy policy = new ConsumerPolicy();
 		SimpleChannel channel = new SimpleChannel();
+		channel.setBroadcaster(true);
 		channel.send(new StringMessage(1, "test"));
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
-		dispatcher.setBroadcast(true);
+		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1));
 		dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) {
 			@Override
@@ -124,23 +127,21 @@ public class ChannelPollingMessageDispatcherTests {
 		});
 		dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1));
 		dispatcher.start();
-		dispatcher.dispatch();
 		latch.await(100, TimeUnit.MILLISECONDS);
 		assertEquals("inactive endpoint should not have received message", 0, counter2.get());
 		assertEquals("both active endpoints should have received message", 2, counter1.get() + counter3.get());
 	}
 
 	@Test
-	public void testDispatcherWithNoExecutors() {
-		ConsumerPolicy policy = new ConsumerPolicy();
+	public void testDispatcherWithNoExecutorsDoesNotFail() {
 		SimpleChannel channel = new SimpleChannel();
 		channel.send(new StringMessage(1, "test"));
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
-		assertEquals(0, dispatcher.dispatch());
+		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
+		dispatcher.start();
 	}
 
-	@Test(expected=MessageDeliveryException.class)
-	public void testBroadcastingDispatcherReachesRejectionLimitAndShouldFail() {
+	@Test
+	public void testBroadcastingDispatcherReachesRejectionLimitAndShouldFail() throws InterruptedException {
 		final AtomicInteger counter1 = new AtomicInteger();
 		final AtomicInteger counter2 = new AtomicInteger();
 		final AtomicInteger counter3 = new AtomicInteger();
@@ -148,23 +149,30 @@ public class ChannelPollingMessageDispatcherTests {
 		TestEndpoint endpoint1 = new TestEndpoint(counter1, latch);
 		TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);
 		TestEndpoint endpoint3 = new TestEndpoint(counter3, latch);
-		ConsumerPolicy policy = new ConsumerPolicy();
 		SimpleChannel channel = new SimpleChannel();
+		channel.setBroadcaster(true);
 		channel.send(new StringMessage(1, "test"));
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
-		dispatcher.setBroadcast(true);
+		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		dispatcher.setRejectionLimit(2);
 		dispatcher.setRetryInterval(3);
 		dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1));
 		dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) {
 			@Override
-			public Message handle(Message<?> message) {
-				throw new MessageHandlerRejectedExecutionException(null);
+			public Message<?> handle(Message<?> message) {
+				throw new MessageHandlerRejectedExecutionException();
 			}
 		});
 		dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1));
+		SimpleChannel errorChannel = new SimpleChannel();
+		SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler();
+		scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel));
+		dispatcher.setMessagingTaskScheduler(scheduler);
 		dispatcher.start();
-		dispatcher.dispatch();
+		latch.await(500, TimeUnit.MILLISECONDS);
+		Message<?> errorMessage = errorChannel.receive(100);
+		assertNotNull(errorMessage);
+		assertTrue(errorMessage instanceof ErrorMessage);
+		assertEquals(MessageDeliveryException.class, ((ErrorMessage) errorMessage).getPayload().getClass());
 	}
 
 	@Test
@@ -176,90 +184,95 @@ public class ChannelPollingMessageDispatcherTests {
 		TestEndpoint endpoint1 = new TestEndpoint(counter1, latch);
 		TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);
 		TestEndpoint endpoint3 = new TestEndpoint(counter3, latch);
-		ConsumerPolicy policy = new ConsumerPolicy();
 		SimpleChannel channel = new SimpleChannel();
+		channel.setBroadcaster(true);
 		channel.send(new StringMessage(1, "test"));
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
-		dispatcher.setBroadcast(true);
+		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		dispatcher.setRejectionLimit(2);
 		dispatcher.setRetryInterval(3);
 		dispatcher.setShouldFailOnRejectionLimit(false);
 		dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1));
 		dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) {
 			@Override
-			public Message handle(Message<?> message) {
-				throw new MessageHandlerRejectedExecutionException(null);
+			public Message<?> handle(Message<?> message) {
+				throw new MessageHandlerRejectedExecutionException();
 			}
 		});
 		dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1));
 		dispatcher.start();
-		dispatcher.dispatch();
 		latch.await(100, TimeUnit.MILLISECONDS);
 		assertEquals("rejecting endpoint should not have received message", 0, counter2.get());
 		assertEquals("both non-rejecting endpoints should have received message", 2, counter1.get() + counter3.get());
 	}
 
-	@Test(expected=MessageDeliveryException.class)
+	@Test
 	public void testNonBroadcastingDispatcherReachesRejectionLimitAndShouldFail() {
 		final AtomicInteger counter1 = new AtomicInteger();
 		final AtomicInteger counter2 = new AtomicInteger();
 		final CountDownLatch latch = new CountDownLatch(1);
 		TestEndpoint endpoint1 = new TestEndpoint(counter1, latch);
 		TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);
-		ConsumerPolicy policy = new ConsumerPolicy();
 		SimpleChannel channel = new SimpleChannel();
 		channel.send(new StringMessage(1, "test"));
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
+		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		dispatcher.setRejectionLimit(2);
 		dispatcher.setRetryInterval(3);
 		dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1) {
 			@Override
-			public Message handle(Message<?> message) {
-				throw new MessageHandlerRejectedExecutionException(null);
+			public Message<?> handle(Message<?> message) {
+				throw new MessageHandlerRejectedExecutionException();
 			}
 		});
 		dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) {
 			@Override
-			public Message handle(Message<?> message) {
-				throw new MessageHandlerRejectedExecutionException(null);
+			public Message<?> handle(Message<?> message) {
+				throw new MessageHandlerRejectedExecutionException();
 			}
 		});
+		SimpleChannel errorChannel = new SimpleChannel();
+		SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler();
+		scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel));
+		dispatcher.setMessagingTaskScheduler(scheduler);
 		dispatcher.start();
-		dispatcher.dispatch();
+		Message<?> errorMessage = errorChannel.receive(100);
+		assertNotNull(errorMessage);
+		assertTrue(errorMessage instanceof ErrorMessage);
+		assertEquals(MessageDeliveryException.class, ((ErrorMessage) errorMessage).getPayload().getClass());
 	}
 
 	@Test
-	public void testNonBroadcastingDispatcherReachesRejectionLimitButShouldNotFail() {
+	public void testNonBroadcastingDispatcherReachesRejectionLimitButShouldNotFail() throws InterruptedException {
 		final AtomicInteger counter1 = new AtomicInteger();
 		final AtomicInteger counter2 = new AtomicInteger();
 		final AtomicInteger rejectedCounter1 = new AtomicInteger();
 		final AtomicInteger rejectedCounter2 = new AtomicInteger();
-		final CountDownLatch latch = new CountDownLatch(1);
+		final CountDownLatch latch = new CountDownLatch(4);
 		TestEndpoint endpoint1 = new TestEndpoint(counter1, latch);
 		TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);
-		ConsumerPolicy policy = new ConsumerPolicy();
 		SimpleChannel channel = new SimpleChannel();
 		channel.send(new StringMessage(1, "test"));
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
+		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		dispatcher.setRejectionLimit(2);
 		dispatcher.setRetryInterval(3);
 		dispatcher.setShouldFailOnRejectionLimit(false);
 		dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1) {
 			@Override
-			public Message handle(Message<?> message) {
+			public Message<?> handle(Message<?> message) {
 				rejectedCounter1.incrementAndGet();
-				throw new MessageHandlerRejectedExecutionException(null);
+				latch.countDown();
+				throw new MessageHandlerRejectedExecutionException();
 			}
 		});
 		dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) {
 			@Override
-			public Message handle(Message<?> message) {
+			public Message<?> handle(Message<?> message) {
 				rejectedCounter2.incrementAndGet();
-				throw new MessageHandlerRejectedExecutionException(null);
+				latch.countDown();
+				throw new MessageHandlerRejectedExecutionException();
 			}
 		});
 		dispatcher.start();
-		dispatcher.dispatch();
+		latch.await(100, TimeUnit.MILLISECONDS);
 		assertEquals("rejecting endpoints should not have received message", 0, counter1.get() + counter2.get());
 		assertEquals("endpoint1 should have rejected two times", 2, rejectedCounter1.get());
 		assertEquals("endpoint2 should have rejected two times", 2, rejectedCounter2.get());
@@ -277,39 +290,37 @@ public class ChannelPollingMessageDispatcherTests {
 		TestEndpoint endpoint1 = new TestEndpoint(counter1, latch);
 		TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);
 		TestEndpoint endpoint3 = new TestEndpoint(counter3, latch);
-		ConsumerPolicy policy = new ConsumerPolicy();
 		SimpleChannel channel = new SimpleChannel();
 		channel.send(new StringMessage(1, "test"));
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
+		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		dispatcher.setRejectionLimit(2);
 		dispatcher.setRetryInterval(3);
 		dispatcher.setShouldFailOnRejectionLimit(false);
 		dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1) {
 			@Override
-			public Message handle(Message<?> message) {
+			public Message<?> handle(Message<?> message) {
 				rejectedCounter1.incrementAndGet();
-				throw new MessageHandlerRejectedExecutionException(null);
+				throw new MessageHandlerRejectedExecutionException();
 			}
 		});
 		dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) {
 			@Override
-			public Message handle(Message<?> message) {
+			public Message<?> handle(Message<?> message) {
 				if (rejectedCounter2.get() == 1) {
 					return super.handle(message);
 				}
 				rejectedCounter2.incrementAndGet();
-				throw new MessageHandlerRejectedExecutionException(null);
+				throw new MessageHandlerRejectedExecutionException();
 			}
 		});
 		dispatcher.addHandler(new PooledMessageHandler(endpoint3, 1, 1) {
 			@Override
-			public Message handle(Message<?> message) {
+			public Message<?> handle(Message<?> message) {
 				rejectedCounter3.incrementAndGet();
-				throw new MessageHandlerRejectedExecutionException(null);
+				throw new MessageHandlerRejectedExecutionException();
 			}
 		});
 		dispatcher.start();
-		dispatcher.dispatch();
 		latch.await(100, TimeUnit.MILLISECONDS);
 		assertEquals("endpoint1 should not have received message", 0, counter1.get());
 		assertEquals("endpoint2 should have received message the second time", 1, counter2.get());
@@ -328,36 +339,34 @@ public class ChannelPollingMessageDispatcherTests {
 		final CountDownLatch latch = new CountDownLatch(2);
 		TestEndpoint endpoint1 = new TestEndpoint(counter1, latch);
 		TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);
-		ConsumerPolicy policy = new ConsumerPolicy();
 		SimpleChannel channel = new SimpleChannel();
+		channel.setBroadcaster(true);
 		channel.send(new StringMessage(1, "test"));
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
-		dispatcher.setBroadcast(true);
+		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		dispatcher.setRejectionLimit(5);
 		dispatcher.setRetryInterval(3);
 		dispatcher.setShouldFailOnRejectionLimit(false);
 		dispatcher.addHandler(new PooledMessageHandler(endpoint1, 1, 1) {
 			@Override
-			public Message handle(Message<?> message) {
+			public Message<?> handle(Message<?> message) {
 				if (rejectedCounter1.get() == 2) {
 					return super.handle(message);
 				}
 				rejectedCounter1.incrementAndGet();
-				throw new MessageHandlerRejectedExecutionException(null);
+				throw new MessageHandlerRejectedExecutionException();
 			}
 		});
 		dispatcher.addHandler(new PooledMessageHandler(endpoint2, 1, 1) {
 			@Override
-			public Message handle(Message<?> message) {
+			public Message<?> handle(Message<?> message) {
 				if (rejectedCounter2.get() == 4) {
 					return super.handle(message);
 				}
 				rejectedCounter2.incrementAndGet();
-				throw new MessageHandlerRejectedExecutionException(null);
+				throw new MessageHandlerRejectedExecutionException();
 			}
 		});
 		dispatcher.start();
-		dispatcher.dispatch();
 		latch.await(100, TimeUnit.MILLISECONDS);
 		assertEquals("endpoint1 should have received one message", 1, counter1.get());
 		assertEquals("endpoint2 should have received one message", 1, counter2.get());
@@ -372,10 +381,9 @@ public class ChannelPollingMessageDispatcherTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		TestEndpoint endpoint1 = new TestEndpoint(counter1, latch);
 		TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);
-		ConsumerPolicy policy = new ConsumerPolicy();
 		SimpleChannel channel = new SimpleChannel();
 		channel.send(new StringMessage(1, "test"));
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
+		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		PooledMessageHandler executor1 = new PooledMessageHandler(endpoint1, 1, 1);
 		PooledMessageHandler executor2 = new PooledMessageHandler(endpoint2, 1, 1);
 		executor1.addMessageSelector(new PayloadTypeSelector(Integer.class));
@@ -383,7 +391,6 @@ public class ChannelPollingMessageDispatcherTests {
 		dispatcher.addHandler(executor1);
 		dispatcher.addHandler(executor2);
 		dispatcher.start();
-		dispatcher.dispatch();
 		latch.await(100, TimeUnit.MILLISECONDS);
 		assertEquals("endpoint1 should not have accepted the message", 0, counter1.get());
 		assertEquals("endpoint2 should have accepted the message", 1, counter2.get());
@@ -399,13 +406,12 @@ public class ChannelPollingMessageDispatcherTests {
 		final CountDownLatch endpointLatch = new CountDownLatch(1);
 		TestEndpoint endpoint1 = new TestEndpoint(counter1, endpointLatch);
 		TestEndpoint endpoint2 = new TestEndpoint(counter2, endpointLatch);
-		ConsumerPolicy policy = new ConsumerPolicy();
 		SimpleChannel channel = new SimpleChannel();
 		channel.send(new StringMessage(1, "test"));
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
+		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		PooledMessageHandler executor1 = new PooledMessageHandler(endpoint1, 1, 1) {
 			@Override
-			public Message handle(Message<?> message) {
+			public Message<?> handle(Message<?> message) {
 				attemptedCounter1.incrementAndGet();
 				attemptedLatch.countDown();
 				return super.handle(message);
@@ -413,7 +419,7 @@ public class ChannelPollingMessageDispatcherTests {
 		};
 		PooledMessageHandler executor2 = new PooledMessageHandler(endpoint2, 1, 1) {
 			@Override
-			public Message handle(Message<?> message) {
+			public Message<?> handle(Message<?> message) {
 				attemptedCounter2.incrementAndGet();
 				attemptedLatch.countDown();
 				return super.handle(message);
@@ -424,7 +430,6 @@ public class ChannelPollingMessageDispatcherTests {
 		dispatcher.addHandler(executor1);
 		dispatcher.addHandler(executor2);
 		dispatcher.start();
-		dispatcher.dispatch();
 		attemptedLatch.await(100, TimeUnit.MILLISECONDS);
 		assertEquals("endpoint1 should not have accepted the message", 0, counter1.get());
 		assertEquals("endpoint2 should not have accepted the message", 0, counter2.get());
@@ -441,11 +446,10 @@ public class ChannelPollingMessageDispatcherTests {
 		final CountDownLatch latch = new CountDownLatch(1);
 		TestEndpoint endpoint1 = new TestEndpoint(counter1, latch);
 		TestEndpoint endpoint2 = new TestEndpoint(counter2, latch);
-		ConsumerPolicy policy = new ConsumerPolicy();
 		SimpleChannel channel = new SimpleChannel();
+		channel.setBroadcaster(true);
 		channel.send(new StringMessage(1, "test"));
-		ChannelPollingMessageDispatcher dispatcher = new ChannelPollingMessageDispatcher(channel, policy);
-		dispatcher.setBroadcast(true);
+		DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(channel);
 		PooledMessageHandler executor1 = new PooledMessageHandler(endpoint1, 1, 1);
 		PooledMessageHandler executor2 = new PooledMessageHandler(endpoint2, 1, 1);
 		executor1.addMessageSelector(new PayloadTypeSelector(Integer.class));
@@ -453,14 +457,12 @@ public class ChannelPollingMessageDispatcherTests {
 		dispatcher.addHandler(executor1);
 		dispatcher.addHandler(executor2);
 		dispatcher.start();
-		dispatcher.dispatch();
 		latch.await(100, TimeUnit.MILLISECONDS);
 		assertEquals("endpoint1 should not have accepted the message", 0, counter1.get());
 		assertEquals("endpoint2 should have accepted the message", 1, counter2.get());
 	}
 
-
-	private static class TestEndpoint extends GenericMessageEndpoint {
+	private static class TestEndpoint extends DefaultMessageEndpoint {
 
 		private AtomicInteger counter;
 
@@ -473,7 +475,7 @@ public class ChannelPollingMessageDispatcherTests {
 		}
 
 		@Override
-		public Message handle(Message message) {
+		public Message<?> handle(Message<?> message) {
 			counter.incrementAndGet();
 			latch.countDown();
 			return null;

spring-integration-core/src/test/java/org/springframework/integration/bus/FixedDelayConsumerTests.java

@@ -24,24 +24,27 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
+
 import org.springframework.integration.channel.SimpleChannel;
-import org.springframework.integration.endpoint.GenericMessageEndpoint;
+import org.springframework.integration.endpoint.ConcurrencyPolicy;
+import org.springframework.integration.endpoint.DefaultMessageEndpoint;
 import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.scheduling.PollingSchedule;
 
 /**
  * @author Mark Fisher
  */
 public class FixedDelayConsumerTests {
 
-	//@Test
+	@Test
 	public void testAllSentMessagesAreReceivedWithinTimeLimit() throws Exception {
 		int messagesToSend = 20;
 		final AtomicInteger counter = new AtomicInteger(0);
 		final CountDownLatch latch = new CountDownLatch(messagesToSend);
 		SimpleChannel channel = new SimpleChannel();
-		MessageEndpoint endpoint = new GenericMessageEndpoint() {
+		MessageEndpoint endpoint = new DefaultMessageEndpoint() {
 			@Override
 			public Message<?> handle(Message<?> message) {
 				counter.incrementAndGet();
@@ -50,18 +53,19 @@ public class FixedDelayConsumerTests {
 			}
 		};
 		MessageBus bus = new MessageBus();
+		bus.initialize();
 		bus.registerChannel("testChannel", channel);
 		bus.registerEndpoint("testEndpoint", endpoint);
-		ConsumerPolicy policy = new ConsumerPolicy();
-		policy.setConcurrency(1);
-		policy.setMaxConcurrency(1);
-		policy.setMaxMessagesPerTask(1);
-		policy.setFixedRate(true);
-		policy.setPeriod(10);
+		PollingSchedule schedule = new PollingSchedule(10);
+		schedule.setFixedRate(false);
+		ConcurrencyPolicy concurrencyPolicy = new ConcurrencyPolicy();
+		concurrencyPolicy.setCoreConcurrency(1);
+		concurrencyPolicy.setMaxConcurrency(1);
 		Subscription subscription = new Subscription();
+		subscription.setSchedule(schedule);
+		subscription.setConcurrencyPolicy(concurrencyPolicy);
 		subscription.setChannel("testChannel");
-		subscription.setReceiver("testEndpoint");
-		subscription.setPolicy(policy);
+		subscription.setHandler("testEndpoint");
 		bus.activateSubscription(subscription);
 		bus.start();
 		for (int i = 0; i < messagesToSend; i++) {
@@ -77,7 +81,7 @@ public class FixedDelayConsumerTests {
 		final AtomicInteger counter = new AtomicInteger(0);
 		final CountDownLatch latch = new CountDownLatch(messagesToSend);
 		SimpleChannel channel = new SimpleChannel();
-		MessageEndpoint endpoint = new GenericMessageEndpoint() {
+		MessageEndpoint endpoint = new DefaultMessageEndpoint() {
 			@Override
 			public Message<?> handle(Message<?> message) {
 				counter.incrementAndGet();
@@ -86,18 +90,15 @@ public class FixedDelayConsumerTests {
 			}
 		};
 		MessageBus bus = new MessageBus();
+		bus.initialize();
 		bus.registerChannel("testChannel", channel);
 		bus.registerEndpoint("testEndpoint", endpoint);
-		ConsumerPolicy policy = new ConsumerPolicy();
-		policy.setConcurrency(1);
-		policy.setMaxConcurrency(1);
-		policy.setMaxMessagesPerTask(1);
-		policy.setFixedRate(true);
-		policy.setPeriod(10);
+		PollingSchedule schedule = new PollingSchedule(10);
+		schedule.setFixedRate(false);
 		Subscription subscription = new Subscription();
 		subscription.setChannel("testChannel");
-		subscription.setReceiver("testEndpoint");
-		subscription.setPolicy(policy);
+		subscription.setHandler("testEndpoint");
+		subscription.setSchedule(schedule);
 		bus.activateSubscription(subscription);
 		for (int i = 0; i < messagesToSend; i++) {
 			channel.send(new GenericMessage<String>(1, "test " + (i+1)));

spring-integration-core/src/test/java/org/springframework/integration/bus/FixedRateConsumerTests.java

@@ -26,10 +26,12 @@ import java.util.concurrent.atomic.AtomicInteger;
 import org.junit.Test;
 
 import org.springframework.integration.channel.SimpleChannel;
-import org.springframework.integration.endpoint.GenericMessageEndpoint;
+import org.springframework.integration.endpoint.ConcurrencyPolicy;
+import org.springframework.integration.endpoint.DefaultMessageEndpoint;
 import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.GenericMessage;
+import org.springframework.integration.scheduling.PollingSchedule;
 
 /**
  * @author Mark Fisher
@@ -42,7 +44,7 @@ public class FixedRateConsumerTests {
 		final AtomicInteger counter = new AtomicInteger(0);
 		final CountDownLatch latch = new CountDownLatch(messagesToSend);
 		SimpleChannel channel = new SimpleChannel();
-		MessageEndpoint endpoint = new GenericMessageEndpoint() {
+		MessageEndpoint endpoint = new DefaultMessageEndpoint() {
 			@Override
 			public Message<?> handle(Message<?> message) {
 				counter.incrementAndGet();
@@ -51,15 +53,15 @@ public class FixedRateConsumerTests {
 			}
 		};
 		MessageBus bus = new MessageBus();
+		bus.initialize();
 		bus.registerChannel("testChannel", channel);
 		bus.registerEndpoint("testEndpoint", endpoint);
-		ConsumerPolicy policy = new ConsumerPolicy();
-		policy.setFixedRate(true);
-		policy.setPeriod(10);
+		PollingSchedule schedule = new PollingSchedule(10);
+		schedule.setFixedRate(true);
 		Subscription subscription = new Subscription();
 		subscription.setChannel("testChannel");
-		subscription.setReceiver("testEndpoint");
-		subscription.setPolicy(policy);
+		subscription.setHandler("testEndpoint");
+		subscription.setSchedule(schedule);
 		bus.activateSubscription(subscription);
 		bus.start();
 		for (int i = 0; i < messagesToSend; i++) {
@@ -75,7 +77,7 @@ public class FixedRateConsumerTests {
 		final AtomicInteger counter = new AtomicInteger(0);
 		final CountDownLatch latch = new CountDownLatch(messagesToSend);
 		SimpleChannel channel = new SimpleChannel();
-		MessageEndpoint endpoint = new GenericMessageEndpoint() {
+		MessageEndpoint endpoint = new DefaultMessageEndpoint() {
 			@Override
 			public Message<?> handle(Message<?> message) {
 				counter.incrementAndGet();
@@ -84,18 +86,19 @@ public class FixedRateConsumerTests {
 			}
 		};
 		MessageBus bus = new MessageBus();
+		bus.initialize();
 		bus.registerChannel("testChannel", channel);
 		bus.registerEndpoint("testEndpoint", endpoint);
-		ConsumerPolicy policy = new ConsumerPolicy();
-		policy.setConcurrency(1);
-		policy.setMaxConcurrency(1);
-		policy.setMaxMessagesPerTask(1);
-		policy.setFixedRate(true);
-		policy.setPeriod(20);
+		PollingSchedule schedule = new PollingSchedule(5);
+		schedule.setFixedRate(true);
+		ConcurrencyPolicy  concurrencyPolicy = new ConcurrencyPolicy();
+		concurrencyPolicy.setCoreConcurrency(1);
+		concurrencyPolicy.setMaxConcurrency(1);
 		Subscription subscription = new Subscription();
 		subscription.setChannel("testChannel");
-		subscription.setReceiver("testEndpoint");
-		subscription.setPolicy(policy);
+		subscription.setHandler("testEndpoint");
+		subscription.setSchedule(schedule);
+		subscription.setConcurrencyPolicy(concurrencyPolicy);
 		bus.activateSubscription(subscription);
 		bus.start();
 		for (int i = 0; i < messagesToSend; i++) {
@@ -103,8 +106,8 @@ public class FixedRateConsumerTests {
 		}
 		latch.await(80, TimeUnit.MILLISECONDS);
 		int count = counter.get();
-		assertTrue("received " + count + ", but expected less than 7", count < 7);
-		assertTrue("received " + count + ", but expected more than 3", count > 3);
+		assertTrue("received " + count + ", but expected less than 20", count < 20);
+		assertTrue("received " + count + ", but expected more than 5", count > 5);
 	}
 
 }

spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java

@@ -21,13 +21,19 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
 import org.junit.Test;
 
 import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.adapter.PollableSource;
+import org.springframework.integration.adapter.PollingSourceAdapter;
 import org.springframework.integration.adapter.SourceAdapter;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.SimpleChannel;
-import org.springframework.integration.endpoint.GenericMessageEndpoint;
+import org.springframework.integration.endpoint.DefaultMessageEndpoint;
 import org.springframework.integration.message.ErrorMessage;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.Message;
@@ -46,7 +52,7 @@ public class MessageBusTests {
 		bus.registerChannel("sourceChannel", sourceChannel);
 		sourceChannel.send(new StringMessage("123", "test"));
 		bus.registerChannel("targetChannel", targetChannel);
-		GenericMessageEndpoint endpoint = new GenericMessageEndpoint();
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
 		endpoint.setInputChannelName("sourceChannel");
 		endpoint.setDefaultOutputChannelName("targetChannel");
 		bus.registerEndpoint("endpoint", endpoint);
@@ -78,11 +84,9 @@ public class MessageBusTests {
 		sourceChannel.send(new GenericMessage<String>("123", "test"));		
 		MessageChannel targetChannel = (MessageChannel) context.getBean("targetChannel");
 		MessageBus bus = (MessageBus) context.getBean("bus");
-		ConsumerPolicy policy = new ConsumerPolicy();
 		Subscription subscription = new Subscription();
 		subscription.setChannel("sourceChannel");
-		subscription.setReceiver("endpoint");
-		subscription.setPolicy(policy);
+		subscription.setHandler("endpoint");
 		bus.activateSubscription(subscription);
 		Message<?> result = targetChannel.receive(100);
 		assertEquals("test", result.getPayload());
@@ -93,10 +97,10 @@ public class MessageBusTests {
 		SimpleChannel inputChannel = new SimpleChannel();
 		SimpleChannel outputChannel1 = new SimpleChannel();
 		SimpleChannel outputChannel2 = new SimpleChannel();
-		GenericMessageEndpoint endpoint1 = new GenericMessageEndpoint();
+		DefaultMessageEndpoint endpoint1 = new DefaultMessageEndpoint();
 		endpoint1.setDefaultOutputChannelName("output1");
 		endpoint1.setInputChannelName("input");
-		GenericMessageEndpoint endpoint2 = new GenericMessageEndpoint();
+		DefaultMessageEndpoint endpoint2 = new DefaultMessageEndpoint();
 		endpoint2.setDefaultOutputChannelName("output2");
 		endpoint2.setInputChannelName("input");
 		MessageBus bus = new MessageBus();
@@ -114,11 +118,14 @@ public class MessageBusTests {
 	}
 
 	@Test
-	public void testInvalidMessageChannelWithFailedDispatch() {
+	public void testInvalidMessageChannelWithFailedDispatch() throws InterruptedException {
 		MessageBus bus = new MessageBus();
-		SourceAdapter sourceAdapter = new FailingSourceAdapter();
+		CountDownLatch latch = new CountDownLatch(1);
+		SourceAdapter sourceAdapter = new PollingSourceAdapter<Object>(new FailingSource(latch));
+		sourceAdapter.setChannel(new SimpleChannel());
 		bus.registerSourceAdapter("testAdapter", sourceAdapter);
 		bus.start();
+		latch.await(1000, TimeUnit.MILLISECONDS);
 		Message<?> message = bus.getInvalidMessageChannel().receive(100);
 		assertNotNull("message should not be null", message);
 		assertTrue(message instanceof ErrorMessage);
@@ -127,27 +134,18 @@ public class MessageBusTests {
 	}
 
 
-	private static class FailingSourceAdapter implements SourceAdapter, MessageDispatcher {
+	private static class FailingSource implements PollableSource<Object> {
 
-		public void setChannel(MessageChannel channel) {
+		private CountDownLatch latch;
+
+		public FailingSource(CountDownLatch latch) {
+			this.latch = latch;
 		}
 
-		public int dispatch() {
+		public Collection<Object> poll(int limit) {
+			latch.countDown();
 			throw new RuntimeException("intentional test failure");
 		}
-
-		public ConsumerPolicy getConsumerPolicy() {
-			return ConsumerPolicy.newPollingPolicy(1000);
-		}
-
-		public boolean isRunning() {
-			return true;
-		}
-
-		public void start() {
-		}
-
-		public void stop() {
-		}
 	}
+
 }

spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml

@@ -10,7 +10,7 @@
 
 	<bean id="targetChannel" class="org.springframework.integration.channel.SimpleChannel"/>
 
-	<bean id="endpoint" class="org.springframework.integration.endpoint.GenericMessageEndpoint">
+	<bean id="endpoint" class="org.springframework.integration.endpoint.DefaultMessageEndpoint">
 		<property name="inputChannelName" value="sourceChannel"/>
 		<property name="defaultOutputChannelName" value="targetChannel"/>
 	</bean>

spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java

@@ -35,7 +35,7 @@ public class EndpointParserTests {
 	@Test
 	public void testSimpleEndpoint() throws InterruptedException {
 		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
-				"genericEndpointTests.xml", this.getClass());
+				"simpleEndpointTests.xml", this.getClass());
 		context.start();
 		MessageChannel channel = (MessageChannel) context.getBean("testChannel");
 		TestHandler handler = (TestHandler) context.getBean("testHandler");

spring-integration-core/src/test/java/org/springframework/integration/config/handlerAdapterEndpointTests.xml

@@ -12,7 +12,7 @@
 	<channel id="testChannel" capacity="50"/>
 
 	<endpoint input-channel="testChannel" handler-ref="testBean" handler-method="store">
-		<consumer period="100"/>
+		<schedule period="100"/>
 	</endpoint>
 
 	<beans:bean id="testBean" class="org.springframework.integration.config.TestBean">

spring-integration-core/src/test/java/org/springframework/integration/config/simpleEndpointTests.xml

@@ -12,7 +12,7 @@
 	<channel id="testChannel" capacity="50"/>
 
 	<endpoint input-channel="testChannel" handler-ref="testHandler">
-		<consumer period="100"/>
+		<schedule period="100"/>
 	</endpoint>
 	
 	<beans:bean id="testHandler" class="org.springframework.integration.config.TestHandler">

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

@@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 
 import org.junit.Test;
+
 import org.springframework.integration.bus.MessageBus;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.SimpleChannel;
@@ -30,18 +31,18 @@ import org.springframework.integration.message.StringMessage;
 /**
  * @author Mark Fisher
  */
-public class GenericMessageEndpointTests {
+public class DefaultMessageEndpointTests {
 
 	@Test
 	public void testDefaultReplyChannel() throws Exception {
 		MessageChannel channel = new SimpleChannel();
 		MessageChannel replyChannel = new SimpleChannel();
 		MessageHandler handler = new MessageHandler() {
-			public Message<String> handle(Message message) {
+			public Message<String> handle(Message<?> message) {
 				return new StringMessage("123", "hello " + message.getPayload());
 			}
 		};
-		GenericMessageEndpoint endpoint = new GenericMessageEndpoint();
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
 		endpoint.setInputChannelName("testChannel");
 		endpoint.setHandler(handler);
 		endpoint.setDefaultOutputChannelName("replyChannel");
@@ -62,11 +63,11 @@ public class GenericMessageEndpointTests {
 		MessageChannel channel = new SimpleChannel();
 		final MessageChannel replyChannel = new SimpleChannel();
 		MessageHandler handler = new MessageHandler() {
-			public Message handle(Message message) {
+			public Message<?> handle(Message<?> message) {
 				return new StringMessage("123", "hello " + message.getPayload());
 			}
 		};
-		GenericMessageEndpoint endpoint = new GenericMessageEndpoint();
+		DefaultMessageEndpoint endpoint = new DefaultMessageEndpoint();
 		endpoint.setInputChannelName("testChannel");
 		endpoint.setHandler(handler);
 		MessageBus bus = new MessageBus();
@@ -77,7 +78,7 @@ public class GenericMessageEndpointTests {
 		StringMessage testMessage = new StringMessage(1, "test");
 		testMessage.getHeader().setReplyChannelName("replyChannel");
 		channel.send(testMessage);
-		Message reply = replyChannel.receive(50);
+		Message<?> reply = replyChannel.receive(50);
 		assertNotNull(reply);
 		assertEquals("hello test", reply.getPayload());
 	}