diff --git a/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageConsumerTests.java b/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageConsumerTests.java index ecbd1247e1..97c3d40611 100644 --- a/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageConsumerTests.java +++ b/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(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(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - 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(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - 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(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - 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(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - 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(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - 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(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - 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(new byte[] {1,2,3}), 0); channel.send(new GenericMessage(new byte[] {4,5,6}), 0); channel.send(new GenericMessage(new byte[] {7,8,9}), 0); - 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"); + } + } + } + } diff --git a/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageConsumerTests.java b/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageConsumerTests.java index 65aef64d99..8fcd83d068 100644 --- a/org.springframework.integration.stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageConsumerTests.java +++ b/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)); - 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(testObject1), 0); channel.send(new GenericMessage(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(testObject1), 0); channel.send(new GenericMessage(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"); + } + } + } + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPoller.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPoller.java deleted file mode 100644 index 55ce4befeb..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/AbstractPoller.java +++ /dev/null @@ -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. - * - *

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(); - -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ChannelPoller.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ChannelPoller.java deleted file mode 100644 index 9f0979e149..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/ChannelPoller.java +++ /dev/null @@ -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; - } - -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ChannelPollerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ChannelPollerTests.java deleted file mode 100644 index ef67ed7caf..0000000000 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/ChannelPollerTests.java +++ /dev/null @@ -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); - } - -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/PollingConsumerEndpointTests.java new file mode 100644 index 0000000000..69a0383293 --- /dev/null +++ b/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 { + } + +}