Removed AbstractPoller and ChannelPoller. Refactored test cases to use PollingConsumerEndpoint.

org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageConsumerTests.java

@@ -20,37 +20,62 @@ import static org.junit.Assert.assertEquals;
 
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.util.Date;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.integration.channel.QueueChannel;
-import org.springframework.integration.endpoint.ChannelPoller;
+import org.springframework.integration.endpoint.PollingConsumerEndpoint;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.StringMessage;
-import org.springframework.integration.scheduling.IntervalTrigger;
+import org.springframework.integration.scheduling.SimpleTaskScheduler;
+import org.springframework.integration.scheduling.Trigger;
 
 /**
  * @author Mark Fisher
  */
 public class ByteStreamWritingMessageConsumerTests {
 
+	private ByteArrayOutputStream stream;
+
+	private ByteStreamWritingMessageConsumer consumer;
+
 	private QueueChannel channel;
 
-	private ChannelPoller poller;
+	private PollingConsumerEndpoint endpoint;
+
+	private TestTrigger trigger = new TestTrigger();
+
+	private SimpleTaskScheduler scheduler; 
 
 
 	@Before
 	public void initialize() {
+		stream = new ByteArrayOutputStream();
+		consumer = new ByteStreamWritingMessageConsumer(stream);
 		this.channel = new QueueChannel(10);
-		this.poller = new ChannelPoller(channel, new IntervalTrigger(0));
+		this.endpoint = new PollingConsumerEndpoint(consumer, channel);
+		scheduler = new SimpleTaskScheduler(new SimpleAsyncTaskExecutor());
+		this.endpoint.setTaskScheduler(scheduler);
+		scheduler.start();
+		trigger.reset();
+		endpoint.setTrigger(trigger);
+	}
+
+	@After
+	public void stop() throws Exception {
+		scheduler.destroy();
 	}
 
 
 	@Test
-	public void testSingleByteArray() {
-		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		ByteStreamWritingMessageConsumer consumer = new ByteStreamWritingMessageConsumer(stream);
+	public void singleByteArray() {
 		consumer.onMessage(new GenericMessage<byte[]>(new byte[] {1,2,3}));
 		byte[] result = stream.toByteArray();
 		assertEquals(3, result.length);
@@ -60,9 +85,7 @@ public class ByteStreamWritingMessageConsumerTests {
 	}
 
 	@Test
-	public void testSingleString() {
-		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		ByteStreamWritingMessageConsumer consumer = new ByteStreamWritingMessageConsumer(stream);
+	public void singleString() {
 		consumer.onMessage(new StringMessage("foo"));
 		byte[] result = stream.toByteArray();
 		assertEquals(3, result.length);
@@ -70,15 +93,14 @@ public class ByteStreamWritingMessageConsumerTests {
 	}
 
 	@Test
-	public void testMaxMessagesPerTaskSameAsMessageCount() {
-		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		ByteStreamWritingMessageConsumer consumer = new ByteStreamWritingMessageConsumer(stream);
-		poller.setMaxMessagesPerPoll(3);
-		poller.setConsumer(consumer);
+	public void maxMessagesPerTaskSameAsMessageCount() {
+		endpoint.setMaxMessagesPerPoll(3);
 		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);
-		poller.run();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		byte[] result = stream.toByteArray();
 		assertEquals(9, result.length);
 		assertEquals(1, result[0]);
@@ -86,31 +108,29 @@ public class ByteStreamWritingMessageConsumerTests {
 	}
 
 	@Test
-	public void testMaxMessagesPerTaskLessThanMessageCount() {
-		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		ByteStreamWritingMessageConsumer consumer = new ByteStreamWritingMessageConsumer(stream);
-		poller.setMaxMessagesPerPoll(2);
-		poller.setConsumer(consumer);
+	public void maxMessagesPerTaskLessThanMessageCount() {
+		endpoint.setMaxMessagesPerPoll(2);
 		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);
-		poller.run();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		byte[] result = stream.toByteArray();
 		assertEquals(6, result.length);
 		assertEquals(1, result[0]);
 	}
 
 	@Test
-	public void testMaxMessagesPerTaskExceedsMessageCount() {
-		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		ByteStreamWritingMessageConsumer consumer = new ByteStreamWritingMessageConsumer(stream);
-		poller.setMaxMessagesPerPoll(5);
-		poller.setReceiveTimeout(0);
-		poller.setConsumer(consumer);
+	public void maxMessagesPerTaskExceedsMessageCount() {
+		endpoint.setMaxMessagesPerPoll(5);
+		endpoint.setReceiveTimeout(0);
 		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);
-		poller.run();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		byte[] result = stream.toByteArray();
 		assertEquals(9, result.length);
 		assertEquals(1, result[0]);
@@ -118,19 +138,21 @@ public class ByteStreamWritingMessageConsumerTests {
 
 	@Test
 	public void testMaxMessagesLessThanMessageCountWithMultipleDispatches() {
-		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		ByteStreamWritingMessageConsumer consumer = new ByteStreamWritingMessageConsumer(stream);
-		poller.setMaxMessagesPerPoll(2);
-		poller.setReceiveTimeout(0);
-		poller.setConsumer(consumer);
+		endpoint.setMaxMessagesPerPoll(2);
+		endpoint.setReceiveTimeout(0);
 		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);
-		poller.run();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		byte[] result1 = stream.toByteArray();
 		assertEquals(6, result1.length);
 		assertEquals(1, result1[0]);
-		poller.run();
+		trigger.reset();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		byte[] result2 = stream.toByteArray();
 		assertEquals(9, result2.length);
 		assertEquals(1, result2[0]);
@@ -139,19 +161,21 @@ public class ByteStreamWritingMessageConsumerTests {
 
 	@Test
 	public void testMaxMessagesExceedsMessageCountWithMultipleDispatches() {
-		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		ByteStreamWritingMessageConsumer consumer = new ByteStreamWritingMessageConsumer(stream);
-		poller.setMaxMessagesPerPoll(5);
-		poller.setReceiveTimeout(0);
-		poller.setConsumer(consumer);
+		endpoint.setMaxMessagesPerPoll(5);
+		endpoint.setReceiveTimeout(0);
 		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);
-		poller.run();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		byte[] result1 = stream.toByteArray();
 		assertEquals(9, result1.length);
 		assertEquals(1, result1[0]);
-		poller.run();
+		trigger.reset();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		byte[] result2 = stream.toByteArray();
 		assertEquals(9, result2.length);	
 		assertEquals(1, result2[0]);
@@ -159,19 +183,21 @@ public class ByteStreamWritingMessageConsumerTests {
 
 	@Test
 	public void testStreamResetBetweenDispatches() {
-		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		ByteStreamWritingMessageConsumer consumer = new ByteStreamWritingMessageConsumer(stream);
-		poller.setMaxMessagesPerPoll(2);
-		poller.setReceiveTimeout(0);
-		poller.setConsumer(consumer);
+		endpoint.setMaxMessagesPerPoll(2);
+		endpoint.setReceiveTimeout(0);
 		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);
-		poller.run();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		byte[] result1 = stream.toByteArray();
 		assertEquals(6, result1.length);
 		stream.reset();
-		poller.run();
+		trigger.reset();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		byte[] result2 = stream.toByteArray();
 		assertEquals(3, result2.length);
 		assertEquals(7, result2[0]);
@@ -179,20 +205,22 @@ public class ByteStreamWritingMessageConsumerTests {
 
 	@Test
 	public void testStreamWriteBetweenDispatches() throws IOException {
-		ByteArrayOutputStream stream = new ByteArrayOutputStream();
-		ByteStreamWritingMessageConsumer consumer = new ByteStreamWritingMessageConsumer(stream);
-		poller.setMaxMessagesPerPoll(2);
-		poller.setReceiveTimeout(0);
-		poller.setConsumer(consumer);
+		endpoint.setMaxMessagesPerPoll(2);
+		endpoint.setReceiveTimeout(0);
 		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);
-		poller.run();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		byte[] result1 = stream.toByteArray();
 		assertEquals(6, result1.length);
 		stream.write(new byte[] {123});
 		stream.flush();
-		poller.run();
+		trigger.reset();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		byte[] result2 = stream.toByteArray();
 		assertEquals(10, result2.length);
 		assertEquals(1, result2[0]);
@@ -200,4 +228,38 @@ public class ByteStreamWritingMessageConsumerTests {
 		assertEquals(7, result2[7]);
 	}
 
+
+	private static class TestTrigger implements Trigger {
+
+		private final AtomicBoolean hasRun = new AtomicBoolean();
+
+		private volatile CountDownLatch latch = new CountDownLatch(1);
+
+
+		public Date getNextRunTime(Date lastScheduledRunTime, Date lastCompleteTime) {
+			if (!hasRun.getAndSet(true)) {
+				return new Date();
+			}
+			this.latch.countDown();
+			return null;
+		}
+
+		public void reset() {
+			this.latch = new CountDownLatch(1);
+			this.hasRun.set(false);
+		}
+
+		public void await() {
+			try {
+				this.latch.await(1000, TimeUnit.MILLISECONDS);
+				if (latch.getCount() != 0) {
+					throw new RuntimeException("test timeout");
+				}
+			}
+			catch (InterruptedException e) {
+				throw new RuntimeException("test latch.await() interrupted");
+			}
+		}
+	}
+
 }

org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageConsumerTests.java

@@ -19,138 +19,163 @@ package org.springframework.integration.stream;
 import static org.junit.Assert.assertEquals;
 
 import java.io.StringWriter;
+import java.util.Date;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
 import org.springframework.integration.channel.QueueChannel;
-import org.springframework.integration.endpoint.ChannelPoller;
+import org.springframework.integration.endpoint.PollingConsumerEndpoint;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.message.StringMessage;
-import org.springframework.integration.scheduling.IntervalTrigger;
+import org.springframework.integration.scheduling.SimpleTaskScheduler;
+import org.springframework.integration.scheduling.Trigger;
 
 /**
  * @author Mark Fisher
  */
 public class CharacterStreamWritingMessageConsumerTests {
 
+	private StringWriter writer;
+
+	private CharacterStreamWritingMessageConsumer consumer;
+
 	private QueueChannel channel;
 
-	private ChannelPoller poller;
+	private PollingConsumerEndpoint endpoint;
+
+	private TestTrigger trigger = new TestTrigger();
+
+	private SimpleTaskScheduler scheduler;
 
 
 	@Before
 	public void initialize() {
+		writer = new StringWriter();
+		consumer = new CharacterStreamWritingMessageConsumer(writer);
 		this.channel = new QueueChannel(10);
-		this.poller = new ChannelPoller(channel, new IntervalTrigger(0));
+		trigger.reset();
+		this.endpoint = new PollingConsumerEndpoint(consumer, channel);
+		scheduler = new SimpleTaskScheduler(new SimpleAsyncTaskExecutor());
+		this.endpoint.setTaskScheduler(scheduler);
+		scheduler.start();
+		trigger.reset();
+		endpoint.setTrigger(trigger);
+	}
+
+	@After
+	public void stop() throws Exception {
+		scheduler.destroy();
 	}
 
 
 	@Test
 	public void singleString() {
-		StringWriter writer = new StringWriter();
-		CharacterStreamWritingMessageConsumer consumer = new CharacterStreamWritingMessageConsumer(writer);
 		consumer.onMessage(new StringMessage("foo"));
 		assertEquals("foo", writer.toString());
 	}
 
 	@Test
 	public void twoStringsAndNoNewLinesByDefault() {
-		StringWriter writer = new StringWriter();
-		CharacterStreamWritingMessageConsumer consumer = new CharacterStreamWritingMessageConsumer(writer);
-		poller.setConsumer(consumer);
-		poller.setMaxMessagesPerPoll(1);
+		endpoint.setMaxMessagesPerPoll(1);
 		channel.send(new StringMessage("foo"), 0);
 		channel.send(new StringMessage("bar"), 0);
-		poller.run();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		assertEquals("foo", writer.toString());
-		poller.run();
+		trigger.reset();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		assertEquals("foobar", writer.toString());
 	}
 
 	@Test
 	public void twoStringsWithNewLines() {
-		StringWriter writer = new StringWriter();
-		CharacterStreamWritingMessageConsumer consumer = new CharacterStreamWritingMessageConsumer(writer);
 		consumer.setShouldAppendNewLine(true);
-		poller.setConsumer(consumer);
-		poller.setMaxMessagesPerPoll(1);
+		endpoint.setMaxMessagesPerPoll(1);
 		channel.send(new StringMessage("foo"), 0);
 		channel.send(new StringMessage("bar"), 0);
-		poller.run();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		String newLine = System.getProperty("line.separator");
 		assertEquals("foo" + newLine, writer.toString());
-		poller.run();
+		trigger.reset();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		assertEquals("foo" + newLine + "bar" + newLine, writer.toString());
 	}
 
 	@Test
 	public void maxMessagesPerTaskSameAsMessageCount() {
-		StringWriter writer = new StringWriter();
-		CharacterStreamWritingMessageConsumer consumer = new CharacterStreamWritingMessageConsumer(writer);
-		poller.setMaxMessagesPerPoll(2);
-		poller.setConsumer(consumer);
+		endpoint.setMaxMessagesPerPoll(2);
 		channel.send(new StringMessage("foo"), 0);
 		channel.send(new StringMessage("bar"), 0);
-		poller.run();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		assertEquals("foobar", writer.toString());
 	}
 
 	@Test
 	public void maxMessagesPerTaskExceedsMessageCountWithAppendedNewLines() {
-		StringWriter writer = new StringWriter();
-		CharacterStreamWritingMessageConsumer consumer = new CharacterStreamWritingMessageConsumer(writer);
-		poller.setMaxMessagesPerPoll(10);
-		poller.setReceiveTimeout(0);
-		poller.setConsumer(consumer);		
+		endpoint.setMaxMessagesPerPoll(10);
+		endpoint.setReceiveTimeout(0);
 		consumer.setShouldAppendNewLine(true);
 		channel.send(new StringMessage("foo"), 0);
 		channel.send(new StringMessage("bar"), 0);
-		poller.run();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		String newLine = System.getProperty("line.separator");
 		assertEquals("foo" + newLine + "bar" + newLine, writer.toString());
 	}
 
 	@Test
 	public void singleNonStringObject() {
-		StringWriter writer = new StringWriter();
-		CharacterStreamWritingMessageConsumer consumer = new CharacterStreamWritingMessageConsumer(writer);
-		poller.setConsumer(consumer);
-		poller.setMaxMessagesPerPoll(1);
+		endpoint.setMaxMessagesPerPoll(1);
 		TestObject testObject = new TestObject("foo");
 		channel.send(new GenericMessage<TestObject>(testObject));
-		poller.run();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		assertEquals("foo", writer.toString());
 	}
 
 	@Test
 	public void twoNonStringObjectWithOutNewLines() {
-		StringWriter writer = new StringWriter();
-		CharacterStreamWritingMessageConsumer consumer = new CharacterStreamWritingMessageConsumer(writer);
-		poller.setReceiveTimeout(0);
-		poller.setMaxMessagesPerPoll(2);
-		poller.setConsumer(consumer);
+		endpoint.setReceiveTimeout(0);
+		endpoint.setMaxMessagesPerPoll(2);
 		TestObject testObject1 = new TestObject("foo");
 		TestObject testObject2 = new TestObject("bar");
 		channel.send(new GenericMessage<TestObject>(testObject1), 0);
 		channel.send(new GenericMessage<TestObject>(testObject2), 0);
-		poller.run();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		assertEquals("foobar", writer.toString());
 	}
 
 	@Test
 	public void twoNonStringObjectWithNewLines() {
-		StringWriter writer = new StringWriter();
-		CharacterStreamWritingMessageConsumer consumer = new CharacterStreamWritingMessageConsumer(writer);
 		consumer.setShouldAppendNewLine(true);
-		poller.setReceiveTimeout(0);
-		poller.setMaxMessagesPerPoll(2);
-		poller.setConsumer(consumer);
+		endpoint.setReceiveTimeout(0);
+		endpoint.setMaxMessagesPerPoll(2);
 		TestObject testObject1 = new TestObject("foo");
 		TestObject testObject2 = new TestObject("bar");
 		channel.send(new GenericMessage<TestObject>(testObject1), 0);
 		channel.send(new GenericMessage<TestObject>(testObject2), 0);
-		poller.run();
+		endpoint.start();
+		trigger.await();
+		endpoint.stop();
 		String newLine = System.getProperty("line.separator");
 		assertEquals("foo" + newLine + "bar" + newLine, writer.toString());
 	}
@@ -169,4 +194,38 @@ public class CharacterStreamWritingMessageConsumerTests {
 		}
 	}
 
+
+	private static class TestTrigger implements Trigger {
+
+		private final AtomicBoolean hasRun = new AtomicBoolean();
+
+		private volatile CountDownLatch latch = new CountDownLatch(1);
+
+
+		public Date getNextRunTime(Date lastScheduledRunTime, Date lastCompleteTime) {
+			if (!hasRun.getAndSet(true)) {
+				return new Date();
+			}
+			this.latch.countDown();
+			return null;
+		}
+
+		public void reset() {
+			this.latch = new CountDownLatch(1);
+			this.hasRun.set(false);
+		}
+
+		public void await() {
+			try {
+				this.latch.await(1000, TimeUnit.MILLISECONDS);
+				if (latch.getCount() != 0) {
+					throw new RuntimeException("test timeout");
+				}
+			}
+			catch (InterruptedException e) {
+				throw new RuntimeException("test latch.await() interrupted");
+			}
+		}
+	}
+
 }

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

@@ -1,157 +0,0 @@
-/*
- * Copyright 2002-2008 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.integration.endpoint;
-
-import org.springframework.beans.factory.InitializingBean;
-import org.springframework.core.task.TaskExecutor;
-import org.springframework.integration.scheduling.Trigger;
-import org.springframework.transaction.PlatformTransactionManager;
-import org.springframework.transaction.TransactionDefinition;
-import org.springframework.transaction.TransactionStatus;
-import org.springframework.transaction.support.DefaultTransactionDefinition;
-import org.springframework.transaction.support.TransactionCallback;
-import org.springframework.transaction.support.TransactionTemplate;
-import org.springframework.util.Assert;
-
-/**
- * Base class for pollers.
- * 
- * @author Mark Fisher
- */
-public abstract class AbstractPoller implements Runnable, InitializingBean {
-
-	public static final int MAX_MESSAGES_UNBOUNDED = -1;
-
-
-	private final Trigger trigger;
-
-	private volatile long maxMessagesPerPoll = MAX_MESSAGES_UNBOUNDED; 
-
-	private volatile TaskExecutor taskExecutor;
-
-	private volatile PlatformTransactionManager transactionManager;
-
-	private volatile TransactionDefinition transactionDefinition;
-
-	private volatile TransactionTemplate transactionTemplate;
-
-	private volatile boolean initialized;
-
-	private final Object initializationMonitor = new Object();
-
-
-	public AbstractPoller(Trigger trigger) {
-		Assert.notNull(trigger, "trigger must not be null");
-		this.trigger = trigger;
-	}
-
-
-	public Trigger getTrigger() {
-		return this.trigger;
-	}
-
-	/**
-	 * Set the maximum number of messages to receive for each poll.
-	 * A non-positive value indicates that polling should repeat as long
-	 * as non-null messages are being received and successfully sent.
-	 * 
-	 * <p>The default is unbounded.
-	 * 
-	 * @see #MAX_MESSAGES_UNBOUNDED
-	 */
-	public void setMaxMessagesPerPoll(int maxMessagesPerPoll) {
-		this.maxMessagesPerPoll = maxMessagesPerPoll;
-	}
-
-	public void setTaskExecutor(TaskExecutor taskExecutor) {
-		this.taskExecutor = taskExecutor;
-	}
-
-	/**
-	 * Specify a transaction manager to use for all exchange operations.
-	 * If none is provided, then the operations will occur without any
-	 * transactional behavior (i.e. there is no default transaction manager).
-	 */
-	public void setTransactionManager(PlatformTransactionManager transactionManager) {
-		this.transactionManager = transactionManager;
-	}
-
-	public void setTransactionDefinition(TransactionDefinition transactionDefinition) {
-		this.transactionDefinition = transactionDefinition;
-	}
-
-	private TransactionTemplate getTransactionTemplate() {
-		if (!this.initialized) {
-			this.afterPropertiesSet();
-		}
-		return this.transactionTemplate;
-	}
-
-	public void afterPropertiesSet() {
-		synchronized (this.initializationMonitor) {
-			if (this.initialized) {
-				return;
-			}
-			if (this.transactionManager != null) {
-				if (this.transactionDefinition == null) {
-					this.transactionDefinition = new DefaultTransactionDefinition();
-				}
-				this.transactionTemplate = new TransactionTemplate(
-						this.transactionManager, this.transactionDefinition);
-			}
-			this.initialized = true;
-		}
-	}
-
-	public void run() {
-		if (this.taskExecutor != null) {
-			this.taskExecutor.execute(new Runnable() {
-				public void run() {
-					poll();
-				}
-			});
-		}
-		else {
-			poll();
-		}
-	}
-
-	private void poll() {
-		int count = 0;
-		while (this.maxMessagesPerPoll < 0 || count < this.maxMessagesPerPoll) {
-			if (!this.innerPoll()) {
-				break;
-			}
-			count++;
-		}
-	}
-
-	private boolean innerPoll() {
-		TransactionTemplate txTemplate = this.getTransactionTemplate();
-		if (txTemplate != null) {
-			return (Boolean) txTemplate.execute(new TransactionCallback() {
-				public Object doInTransaction(TransactionStatus status) {
-					return doPoll();
-				}
-			});
-		}
-		return doPoll();
-	}
-
-	protected abstract boolean doPoll();
-
-}

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

@@ -1,69 +0,0 @@
-/*
- * Copyright 2002-2008 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.integration.endpoint;
-
-import org.springframework.integration.channel.PollableChannel;
-import org.springframework.integration.core.Message;
-import org.springframework.integration.message.MessageConsumer;
-import org.springframework.integration.scheduling.Trigger;
-import org.springframework.util.Assert;
-
-/**
- * @author Mark Fisher
- */
-public class ChannelPoller extends AbstractPoller {
-
-	private final PollableChannel channel;
-
-	private volatile MessageConsumer consumer;
-
-	private volatile long receiveTimeout = 1000;
-
-
-	public ChannelPoller(PollableChannel channel, Trigger trigger) {
-		super(trigger);
-		Assert.notNull(channel, "channel must not be null");
-		this.channel = channel;
-	}
-
-	/**
-	 * Specify the timeout to use when receiving from the channel (in milliseconds).
-	 * A negative value indicates that receive calls should block indefinitely.
-	 * The default value is 1000 (1 second).
-	 */
-	public void setReceiveTimeout(long receiveTimeout) {
-		this.receiveTimeout = receiveTimeout;
-	}
-
-	public void setConsumer(MessageConsumer consumer) {
-		this.consumer = consumer;
-	}
-
-	@Override
-	protected boolean doPoll() {
-		Assert.notNull(this.consumer, "consumer must not be null");
-		Message<?> message = (this.receiveTimeout >= 0)
-				? this.channel.receive(this.receiveTimeout)
-				: this.channel.receive();
-		if (message == null) {
-			return false;
-		}
-		this.consumer.onMessage(message);
-		return true;
-	}
-
-}

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

@@ -1,139 +0,0 @@
-/*
- * Copyright 2002-2008 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.integration.endpoint;
-
-import static org.easymock.EasyMock.createMock;
-import static org.easymock.EasyMock.expect;
-import static org.easymock.EasyMock.expectLastCall;
-import static org.easymock.EasyMock.replay;
-import static org.easymock.EasyMock.reset;
-import static org.easymock.EasyMock.verify;
-
-import org.junit.Before;
-import org.junit.Test;
-
-import org.springframework.integration.channel.PollableChannel;
-import org.springframework.integration.core.Message;
-import org.springframework.integration.message.MessageConsumer;
-import org.springframework.integration.message.MessageRejectedException;
-import org.springframework.integration.scheduling.Trigger;
-
-/**
- * @author Iwein Fuld
- */
-@SuppressWarnings("unchecked")
-public class ChannelPollerTests {
-
-	private ChannelPoller poller;
-	private Trigger triggerMock = createMock(Trigger.class);
-	private PollableChannel channelMock = createMock(PollableChannel.class);
-	private MessageConsumer endpointMock = createMock(MessageConsumer.class);
-	private Message messageMock = createMock(Message.class);
-	private Object[] globalMocks = new Object[] { triggerMock, channelMock, endpointMock, messageMock };
-
-
-	@Before
-	public void init() {
-		poller = new ChannelPoller(channelMock, triggerMock);
-		poller.setConsumer(endpointMock);
-		poller.setReceiveTimeout(-1);
-		reset(globalMocks);
-	}
-
-
-	@Test
-	public void singleMessage() {
-		expect(channelMock.receive()).andReturn(messageMock);
-		endpointMock.onMessage(messageMock);
-		expectLastCall();
-		replay(globalMocks);
-		poller.setMaxMessagesPerPoll(1);
-		poller.run();
-		verify(globalMocks);
-	}
-
-	@Test
-	public void multipleMessages() {
-		expect(channelMock.receive()).andReturn(messageMock).times(5);
-		endpointMock.onMessage(messageMock);
-		expectLastCall().times(5);
-		replay(globalMocks);
-		poller.setMaxMessagesPerPoll(5);
-		poller.run();
-		verify(globalMocks);
-	}
-
-	@Test
-	public void multipleMessages_underrun() {
-		expect(channelMock.receive()).andReturn(messageMock).times(5);
-		expect(channelMock.receive()).andReturn(null);
-		endpointMock.onMessage(messageMock);
-		expectLastCall().times(5);
-		replay(globalMocks);
-		poller.setMaxMessagesPerPoll(6);
-		poller.run();
-		verify(globalMocks);
-	}
-
-	@Test(expected = MessageRejectedException.class)
-	public void rejectedMessage() {
-		expect(channelMock.receive()).andReturn(messageMock);
-		endpointMock.onMessage(messageMock);
-		expectLastCall().andThrow(new MessageRejectedException(messageMock, "intentional test failure"));
-		replay(globalMocks);
-		poller.run();
-		verify(globalMocks);
-	}
-
-	@Test(expected = MessageRejectedException.class)
-	public void droppedMessage_onePerPoll() {
-		expect(channelMock.receive()).andReturn(messageMock).times(1);
-		endpointMock.onMessage(messageMock);
-		expectLastCall().andThrow(new MessageRejectedException(messageMock, "intentional test failure")).anyTimes();
-		replay(globalMocks);
-		poller.setMaxMessagesPerPoll(10);
-		poller.run();
-		verify(globalMocks);
-	}
-
-	@Test
-	public void blockingSourceTimedOut() {
-		poller = new ChannelPoller(channelMock, triggerMock);
-		poller.setConsumer(endpointMock);
-		// we don't need to await the timeout, returning null suffices
-		expect(channelMock.receive(1)).andReturn(null);
-		replay(globalMocks);
-		poller.setReceiveTimeout(1);
-		poller.run();
-		verify(globalMocks);
-	}
-
-	@Test
-	public void blockingSourceNotTimedOut() {
-		poller = new ChannelPoller(channelMock, triggerMock);
-		poller.setConsumer(endpointMock);
-		expect(channelMock.receive(1)).andReturn(messageMock);
-		endpointMock.onMessage(messageMock);
-		expectLastCall();
-		replay(globalMocks);
-		poller.setReceiveTimeout(1);
-		poller.setMaxMessagesPerPoll(1);
-		poller.run();
-		verify(globalMocks);
-	}
-
-}

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

@@ -0,0 +1,234 @@
+/*
+ * Copyright 2002-2008 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.endpoint;
+
+import static org.easymock.EasyMock.createMock;
+import static org.easymock.EasyMock.expect;
+import static org.easymock.EasyMock.expectLastCall;
+import static org.easymock.EasyMock.replay;
+import static org.easymock.EasyMock.reset;
+import static org.easymock.EasyMock.verify;
+
+import java.util.Date;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.integration.channel.PollableChannel;
+import org.springframework.integration.core.Message;
+import org.springframework.integration.message.MessageConsumer;
+import org.springframework.integration.message.MessageRejectedException;
+import org.springframework.integration.message.StringMessage;
+import org.springframework.integration.scheduling.SimpleTaskScheduler;
+import org.springframework.integration.scheduling.Trigger;
+import org.springframework.integration.util.ErrorHandler;
+
+/**
+ * @author Iwein Fuld
+ * @author Mark Fisher
+ */
+@SuppressWarnings("unchecked")
+public class PollingConsumerEndpointTests {
+
+	private PollingConsumerEndpoint endpoint;
+
+	private Trigger trigger = new TestTrigger();
+
+	private TestConsumer consumer = new TestConsumer();
+
+	private Message message = new StringMessage("test");
+
+	private Message badMessage = new StringMessage("bad");
+
+	private TestErrorHandler errorHandler = new TestErrorHandler();
+
+	private PollableChannel channelMock = createMock(PollableChannel.class);
+
+	private SimpleTaskScheduler taskScheduler = new SimpleTaskScheduler(new SimpleAsyncTaskExecutor());
+
+
+	@Before
+	public void init() throws InterruptedException {
+		endpoint = new PollingConsumerEndpoint(consumer, channelMock);
+		endpoint.setTaskScheduler(taskScheduler);
+		taskScheduler.setErrorHandler(errorHandler);
+		taskScheduler.start();
+		endpoint.setTrigger(trigger);
+		endpoint.setReceiveTimeout(-1);
+		reset(channelMock);
+	}
+
+	@After
+	public void stop() {
+		taskScheduler.stop();
+	}
+
+
+	@Test
+	public void singleMessage() throws InterruptedException {
+		expect(channelMock.receive()).andReturn(message);
+		expectLastCall();
+		replay(channelMock);
+		endpoint.setMaxMessagesPerPoll(1);
+		endpoint.start();
+		consumer.await(500);
+		endpoint.stop();
+		verify(channelMock);
+	}
+
+	@Test
+	public void multipleMessages() throws InterruptedException {
+		expect(channelMock.receive()).andReturn(message).times(5);
+		replay(channelMock);
+		endpoint.setMaxMessagesPerPoll(5);
+		endpoint.start();
+		consumer.await(500);
+		endpoint.stop();
+		verify(channelMock);
+	}
+
+	@Test
+	public void multipleMessages_underrun() throws InterruptedException {
+		expect(channelMock.receive()).andReturn(message).times(5);
+		expect(channelMock.receive()).andReturn(null);
+		replay(channelMock);
+		endpoint.setMaxMessagesPerPoll(6);
+		endpoint.start();
+		consumer.await(500);
+		endpoint.stop();
+		verify(channelMock);
+	}
+
+	@Test(expected = MessageRejectedException.class)
+	public void rejectedMessage() throws Throwable {
+		expect(channelMock.receive()).andReturn(badMessage);
+		replay(channelMock);
+		endpoint.start();
+		consumer.await(500);
+		endpoint.stop();
+		verify(channelMock);
+		errorHandler.throwLastErrorIfAvailable();
+	}
+
+	@Test(expected = MessageRejectedException.class)
+	public void droppedMessage_onePerPoll() throws Throwable {
+		expect(channelMock.receive()).andReturn(badMessage).times(1);
+		replay(channelMock);
+		endpoint.setMaxMessagesPerPoll(10);
+		endpoint.start();
+		consumer.await(500);
+		endpoint.stop();
+		verify(channelMock);
+		errorHandler.throwLastErrorIfAvailable();
+	}
+
+	@Test(expected = TestTimeoutException.class)
+	public void blockingSourceTimedOut() throws InterruptedException {
+		// we don't need to await the timeout, returning null suffices
+		expect(channelMock.receive(1)).andReturn(null);
+		replay(channelMock);
+		endpoint.setReceiveTimeout(1);
+		endpoint.start();
+		try {
+			consumer.await(500);
+		}
+		finally {
+			endpoint.stop();
+			verify(channelMock);
+		}
+	}
+
+	@Test
+	public void blockingSourceNotTimedOut() throws InterruptedException {
+		expect(channelMock.receive(1)).andReturn(message);
+		expectLastCall();
+		replay(channelMock);
+		endpoint.setReceiveTimeout(1);
+		endpoint.setMaxMessagesPerPoll(1);
+		endpoint.start();
+		consumer.await(500);
+		endpoint.stop();
+		verify(channelMock);
+	}
+
+
+	private static class TestConsumer implements MessageConsumer {
+
+		private volatile CountDownLatch latch = new CountDownLatch(1);
+
+		public void onMessage(Message<?> message) {
+			try {
+				if ("bad".equals(message.getPayload().toString())) {
+					throw new MessageRejectedException(message, "intentional test failure");
+				}
+			}
+			finally {
+				this.latch.countDown();
+			}
+		}
+
+		public void await(long timeout) throws InterruptedException {
+			this.latch.await(timeout, TimeUnit.MILLISECONDS);
+			if (this.latch.getCount() == 0) {
+				this.latch = new CountDownLatch(1);
+			}
+			else {
+				throw new TestTimeoutException();
+			}
+		}
+	}
+
+
+	private static class TestTrigger implements Trigger {
+
+		private final AtomicBoolean hasRun = new AtomicBoolean();
+
+		public Date getNextRunTime(Date lastScheduledRunTime, Date lastCompleteTime) {
+			if (!hasRun.getAndSet(true)) {
+				return new Date();
+			}
+			return null;
+		}
+	}
+
+
+	private static class TestErrorHandler implements ErrorHandler {
+
+		private volatile Throwable lastError;
+
+		public void handle(Throwable t) {
+			this.lastError = t;
+		}
+
+		public void throwLastErrorIfAvailable() throws Throwable {
+			Throwable t = this.lastError;
+			this.lastError = null;
+			throw t;
+		}
+	}
+
+
+	@SuppressWarnings("serial")
+	private static class TestTimeoutException extends RuntimeException {
+	}
+
+}