diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapterTests.java index e5b7267ab3..00019ecc64 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapterTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapterTests.java @@ -32,7 +32,7 @@ import org.springframework.context.event.ContextStartedEvent; import org.springframework.context.event.ContextStoppedEvent; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; /** @@ -42,7 +42,7 @@ public class ApplicationEventSourceAdapterTests { @Test public void testAnyApplicationEventSentByDefault() { - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); ApplicationEventSourceAdapter adapter = new ApplicationEventSourceAdapter(channel); Message message1 = channel.receive(0); assertNull(message1); @@ -58,7 +58,7 @@ public class ApplicationEventSourceAdapterTests { @Test public void testOnlyConfiguredEventTypesAreSent() { - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); ApplicationEventSourceAdapter adapter = new ApplicationEventSourceAdapter(channel); List> eventTypes = new ArrayList>(); eventTypes.add(TestApplicationEvent1.class); diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapterTests.java index edcd2855af..48e22e7176 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapterTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapterTests.java @@ -27,7 +27,7 @@ import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationEventPublisher; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; import org.springframework.integration.scheduling.Subscription; @@ -45,7 +45,7 @@ public class ApplicationEventTargetAdapterTests { latch.countDown(); } }; - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); ApplicationEventTargetAdapter adapter = new ApplicationEventTargetAdapter(); adapter.setApplicationEventPublisher(publisher); MessageBus bus = new MessageBus(); diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/applicationEventSourceAdapterTests.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/applicationEventSourceAdapterTests.xml index 59161a80f9..43d88b7b01 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/applicationEventSourceAdapterTests.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/applicationEventSourceAdapterTests.xml @@ -6,7 +6,7 @@ - + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerSourceAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerSourceAdapterTests.java index c4df56366e..84b6d4c8b2 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerSourceAdapterTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerSourceAdapterTests.java @@ -30,7 +30,7 @@ import java.util.concurrent.Executors; import org.junit.Test; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; import org.springframework.mock.web.MockHttpServletRequest; @@ -45,7 +45,7 @@ public class HttpInvokerSourceAdapterTests { @Test public void testRequestOnly() throws Exception { - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); HttpInvokerSourceAdapter adapter = new HttpInvokerSourceAdapter(channel); adapter.setExpectReply(false); adapter.afterPropertiesSet(); @@ -60,7 +60,7 @@ public class HttpInvokerSourceAdapterTests { @Test public void testRequestExpectingReply() throws Exception { - final MessageChannel channel = new SimpleChannel(); + final MessageChannel channel = new QueueChannel(); Executors.newSingleThreadExecutor().execute(new Runnable() { public void run() { Message message = channel.receive(); diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParserTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParserTests.java index 22b40230c1..7cd7aa27f0 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParserTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParserTests.java @@ -26,7 +26,7 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.adapter.rmi.RmiSourceAdapter; import org.springframework.integration.adapter.rmi.RmiTargetAdapter; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.HandlerEndpoint; import org.springframework.integration.message.StringMessage; @@ -35,7 +35,7 @@ import org.springframework.integration.message.StringMessage; */ public class RmiTargetAdapterParserTests { - private final SimpleChannel testChannel = new SimpleChannel(); + private final QueueChannel testChannel = new QueueChannel(); @Before diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapterTests.java index 417298b166..f9fa14f2d2 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapterTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapterTests.java @@ -25,7 +25,7 @@ import org.junit.Test; import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; import org.springframework.integration.scheduling.PollingSchedule; @@ -38,7 +38,7 @@ public class ByteStreamSourceAdapterTests { public void testEndOfStream() { byte[] bytes = new byte[] {1,2,3}; ByteArrayInputStream stream = new ByteArrayInputStream(bytes); - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); ByteStreamSource source = new ByteStreamSource(stream); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); @@ -61,7 +61,7 @@ public class ByteStreamSourceAdapterTests { public void testEndOfStreamWithMaxMessagesPerTask() throws Exception { byte[] bytes = new byte[] {0,1,2,3,4,5,6,7}; ByteArrayInputStream stream = new ByteArrayInputStream(bytes); - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); ByteStreamSource source = new ByteStreamSource(stream); source.setBytesPerMessage(8); PollingSchedule schedule = new PollingSchedule(1000); @@ -79,7 +79,7 @@ public class ByteStreamSourceAdapterTests { public void testMultipleMessagesWithSingleMessagePerTask() { byte[] bytes = new byte[] {0,1,2,3,4,5,6,7}; ByteArrayInputStream stream = new ByteArrayInputStream(bytes); - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); ByteStreamSource source = new ByteStreamSource(stream); source.setBytesPerMessage(4); PollingSchedule schedule = new PollingSchedule(1000); @@ -104,7 +104,7 @@ public class ByteStreamSourceAdapterTests { public void testLessThanMaxMessagesAvailable() { byte[] bytes = new byte[] {0,1,2,3,4,5,6,7}; ByteArrayInputStream stream = new ByteArrayInputStream(bytes); - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); ByteStreamSource source = new ByteStreamSource(stream); source.setBytesPerMessage(4); PollingSchedule schedule = new PollingSchedule(1000); @@ -128,7 +128,7 @@ public class ByteStreamSourceAdapterTests { public void testByteArrayIsTruncated() { byte[] bytes = new byte[] {0,1,2,3,4,5}; ByteArrayInputStream stream = new ByteArrayInputStream(bytes); - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); ByteStreamSource source = new ByteStreamSource(stream); source.setBytesPerMessage(4); PollingSchedule schedule = new PollingSchedule(1000); @@ -149,7 +149,7 @@ public class ByteStreamSourceAdapterTests { public void testByteArrayIsNotTruncated() { byte[] bytes = new byte[] {0,1,2,3,4,5}; ByteArrayInputStream stream = new ByteArrayInputStream(bytes); - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); ByteStreamSource source = new ByteStreamSource(stream); source.setBytesPerMessage(4); source.setShouldTruncate(false); diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java index 1a90dc0108..23fbd3356a 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamTargetAdapterTests.java @@ -24,7 +24,7 @@ import java.io.IOException; import org.junit.Test; import org.springframework.integration.channel.DispatcherPolicy; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.dispatcher.PollingDispatcher; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; @@ -62,7 +62,7 @@ public class ByteStreamTargetAdapterTests { ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream); DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(3); - SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); + QueueChannel channel = new QueueChannel(5, dispatcherPolicy); PollingDispatcher dispatcher = new PollingDispatcher(channel, null); dispatcher.subscribe(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); @@ -81,7 +81,7 @@ public class ByteStreamTargetAdapterTests { ByteStreamTargetAdapter adapter = new ByteStreamTargetAdapter(stream); DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(2); - SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); + QueueChannel channel = new QueueChannel(5, dispatcherPolicy); PollingDispatcher dispatcher = new PollingDispatcher(channel, null); dispatcher.subscribe(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); @@ -100,7 +100,7 @@ public class ByteStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(5); dispatcherPolicy.setReceiveTimeout(0); - SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); + QueueChannel channel = new QueueChannel(5, dispatcherPolicy); PollingDispatcher dispatcher = new PollingDispatcher(channel, null); dispatcher.subscribe(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); @@ -119,7 +119,7 @@ public class ByteStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(2); dispatcherPolicy.setReceiveTimeout(0); - SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); + QueueChannel channel = new QueueChannel(5, dispatcherPolicy); PollingDispatcher dispatcher = new PollingDispatcher(channel, null); dispatcher.subscribe(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); @@ -143,7 +143,7 @@ public class ByteStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(5); dispatcherPolicy.setReceiveTimeout(0); - SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); + QueueChannel channel = new QueueChannel(5, dispatcherPolicy); PollingDispatcher dispatcher = new PollingDispatcher(channel, null); dispatcher.subscribe(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); @@ -166,7 +166,7 @@ public class ByteStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(2); dispatcherPolicy.setReceiveTimeout(0); - SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); + QueueChannel channel = new QueueChannel(5, dispatcherPolicy); PollingDispatcher dispatcher = new PollingDispatcher(channel, null); dispatcher.subscribe(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); @@ -189,7 +189,7 @@ public class ByteStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(2); dispatcherPolicy.setReceiveTimeout(0); - SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); + QueueChannel channel = new QueueChannel(5, dispatcherPolicy); PollingDispatcher dispatcher = new PollingDispatcher(channel, null); dispatcher.subscribe(adapter); channel.send(new GenericMessage(new byte[] {1,2,3}), 0); diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java index 2e61778a01..ca72cecbc7 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java @@ -25,7 +25,7 @@ import org.junit.Test; import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; import org.springframework.integration.scheduling.PollingSchedule; @@ -37,7 +37,7 @@ public class CharacterStreamSourceAdapterTests { @Test public void testEndOfStream() { StringReader reader = new StringReader("test"); - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); CharacterStreamSource source = new CharacterStreamSource(reader); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); @@ -55,7 +55,7 @@ public class CharacterStreamSourceAdapterTests { @Test public void testEndOfStreamWithMaxMessagesPerTask() { StringReader reader = new StringReader("test"); - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); CharacterStreamSource source = new CharacterStreamSource(reader); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); @@ -72,7 +72,7 @@ public class CharacterStreamSourceAdapterTests { public void testMultipleLinesWithSingleMessagePerTask() { String s = "test1" + System.getProperty("line.separator") + "test2"; StringReader reader = new StringReader(s); - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); CharacterStreamSource source = new CharacterStreamSource(reader); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); @@ -92,7 +92,7 @@ public class CharacterStreamSourceAdapterTests { public void testLessThanMaxMessagesAvailable() { String s = "test1" + System.getProperty("line.separator") + "test2"; StringReader reader = new StringReader(s); - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); CharacterStreamSource source = new CharacterStreamSource(reader); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(5000); diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java index 89e928154b..3976de3af9 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamTargetAdapterTests.java @@ -24,7 +24,7 @@ import org.junit.Test; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.dispatcher.PollingDispatcher; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.StringMessage; @@ -44,7 +44,7 @@ public class CharacterStreamTargetAdapterTests { @Test public void testTwoStringsAndNoNewLinesByDefault() { - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); StringWriter writer = new StringWriter(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer); PollingDispatcher dispatcher = new PollingDispatcher(channel, null); @@ -59,7 +59,7 @@ public class CharacterStreamTargetAdapterTests { @Test public void testTwoStringsWithNewLines() { - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); StringWriter writer = new StringWriter(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer); adapter.setShouldAppendNewLine(true); @@ -80,7 +80,7 @@ public class CharacterStreamTargetAdapterTests { CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer); DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(2); - SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); + QueueChannel channel = new QueueChannel(5, dispatcherPolicy); PollingDispatcher dispatcher = new PollingDispatcher(channel, null); dispatcher.subscribe(adapter); channel.send(new StringMessage("foo"), 0); @@ -96,7 +96,7 @@ public class CharacterStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setMaxMessagesPerTask(10); dispatcherPolicy.setReceiveTimeout(0); - SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); + QueueChannel channel = new QueueChannel(5, dispatcherPolicy); PollingDispatcher dispatcher = new PollingDispatcher(channel, null); adapter.setShouldAppendNewLine(true); dispatcher.subscribe(adapter); @@ -109,7 +109,7 @@ public class CharacterStreamTargetAdapterTests { @Test public void testSingleNonStringObject() { - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); StringWriter writer = new StringWriter(); CharacterStreamTargetAdapter adapter = new CharacterStreamTargetAdapter(writer); PollingDispatcher dispatcher = new PollingDispatcher(channel, null); @@ -127,7 +127,7 @@ public class CharacterStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(2); - SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); + QueueChannel channel = new QueueChannel(5, dispatcherPolicy); PollingDispatcher dispatcher = new PollingDispatcher(channel, null); dispatcher.subscribe(adapter); TestObject testObject1 = new TestObject("foo"); @@ -145,7 +145,7 @@ public class CharacterStreamTargetAdapterTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(2); - SimpleChannel channel = new SimpleChannel(5, dispatcherPolicy); + QueueChannel channel = new QueueChannel(5, dispatcherPolicy); adapter.setShouldAppendNewLine(true); PollingDispatcher dispatcher = new PollingDispatcher(channel, null); dispatcher.subscribe(adapter); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/DefaultErrorChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/DefaultErrorChannel.java index 9f2da472ba..f22b159b15 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/DefaultErrorChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/DefaultErrorChannel.java @@ -20,7 +20,7 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.RendezvousChannel; import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter; import org.springframework.integration.message.Message; @@ -29,7 +29,7 @@ import org.springframework.integration.message.Message; * * @author Mark Fisher */ -public class DefaultErrorChannel extends SimpleChannel { +public class DefaultErrorChannel extends RendezvousChannel { private final Log logger = LogFactory.getLog(this.getClass()); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index b7735f4202..207f6f35c9 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -41,7 +41,7 @@ import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.dispatcher.SynchronousChannel; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.DefaultEndpointRegistry; @@ -334,7 +334,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio if (this.logger.isInfoEnabled()) { logger.info("auto-creating channel '" + channelName + "'"); } - channel = new SimpleChannel(); + channel = new QueueChannel(); this.registerChannel(channelName, channel); } } @@ -347,7 +347,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio "' configured as 'default-output' for endpoint '" + endpoint + "'. Consider enabling the 'autoCreateChannels' option for the message bus."); } - this.registerChannel(outputChannelName, new SimpleChannel()); + this.registerChannel(outputChannelName, new QueueChannel()); } } if (endpoint instanceof TargetEndpoint) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java index aea26c5c2a..5b5d8fd442 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java @@ -30,9 +30,6 @@ import org.springframework.integration.message.selector.MessageSelector; */ public interface MessageChannel extends BlockingSource, BlockingTarget { - static final int DEFAULT_CAPACITY = 100; - - /** * Return the name of this channel. */ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java index c4c78fd52b..bf9091a42b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java @@ -31,7 +31,7 @@ import org.springframework.integration.message.MessagePriority; * * @author Mark Fisher */ -public class PriorityChannel extends BaseBlockingQueueChannel { +public class PriorityChannel extends QueueChannel { private final Semaphore semaphore; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/BaseBlockingQueueChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java similarity index 66% rename from spring-integration-core/src/main/java/org/springframework/integration/channel/BaseBlockingQueueChannel.java rename to spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java index 690838a397..49a225e25e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/BaseBlockingQueueChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java @@ -19,6 +19,7 @@ package org.springframework.integration.channel; import java.util.ArrayList; import java.util.List; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import org.springframework.integration.message.Message; @@ -26,11 +27,18 @@ import org.springframework.integration.message.selector.MessageSelector; import org.springframework.util.Assert; /** - * Base implementation for message channels that delegate to a {@link BlockingQueue}. + * Simple implementation of a message channel. Each {@link Message} is placed in + * a {@link BlockingQueue} whose capacity may be specified upon construction. + * The capacity must be a positive integer value. For a zero-capacity version + * based upon a {@link java.util.concurrent.SynchronousQueue}, consider the + * {@link RendezvousChannel}. * * @author Mark Fisher */ -public class BaseBlockingQueueChannel extends AbstractMessageChannel { +public class QueueChannel extends AbstractMessageChannel { + + public static final int DEFAULT_CAPACITY = 100; + private final BlockingQueue> queue; @@ -38,12 +46,37 @@ public class BaseBlockingQueueChannel extends AbstractMessageChannel { /** * Create a channel with the specified queue and dispatcher policy. */ - public BaseBlockingQueueChannel(BlockingQueue> queue, DispatcherPolicy dispatcherPolicy) { + public QueueChannel(BlockingQueue> queue, DispatcherPolicy dispatcherPolicy) { super((dispatcherPolicy != null) ? dispatcherPolicy : new DispatcherPolicy()); Assert.notNull(queue, "'queue' must not be null"); this.queue = queue; } + /** + * Create a channel with the specified queue capacity and dispatcher policy. + */ + public QueueChannel(int capacity, DispatcherPolicy dispatcherPolicy) { + super((dispatcherPolicy != null) ? dispatcherPolicy : new DispatcherPolicy()); + Assert.isTrue(capacity > 0, "The capacity must be a positive integer. " + + "For a zero-capacity alternative, consider '" + RendezvousChannel.class + "'"); + this.queue = new LinkedBlockingQueue>(capacity); + } + + /** + * Create a channel with the specified queue capacity. + */ + public QueueChannel(int capacity) { + this(capacity, null); + } + + /** + * Create a channel with the default queue capacity. + * @see #DEFAULT_CAPACITY + */ + public QueueChannel() { + this(DEFAULT_CAPACITY, null); + } + protected boolean doSend(Message message, long timeout) { Assert.notNull(message, "'message' must not be null"); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/RendezvousChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/RendezvousChannel.java new file mode 100644 index 0000000000..4831cd3512 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/RendezvousChannel.java @@ -0,0 +1,40 @@ +/* + * 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.channel; + +import java.util.concurrent.SynchronousQueue; + +import org.springframework.integration.message.Message; + +/** + * A zero-capacity version of {@link QueueChannel} that delegates to a + * {@link SynchronousQueue} internally. This accommodates "handoff" scenarios + * (i.e. blocking while waiting for another party to send or receive). + * + * @author Mark Fisher + */ +public class RendezvousChannel extends QueueChannel { + + public RendezvousChannel(DispatcherPolicy dispatcherPolicy) { + super(new SynchronousQueue>(), dispatcherPolicy); + } + + public RendezvousChannel() { + this(null); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java index 79b463d1bd..1be20c09d7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java @@ -96,7 +96,7 @@ public class RequestReplyTemplate { * @return the reply message or null */ public Message request(Message requestMessage, long sendTimeout, long receiveTimeout) { - SimpleChannel replyChannel = new SimpleChannel(0); + RendezvousChannel replyChannel = new RendezvousChannel(); requestMessage.getHeader().setReturnAddress(replyChannel); this.requestChannel.send(requestMessage, sendTimeout); return replyChannel.receive(receiveTimeout); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java deleted file mode 100644 index 77db771dbe..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java +++ /dev/null @@ -1,55 +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.channel; - -import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.SynchronousQueue; - -import org.springframework.integration.message.Message; - -/** - * Simple implementation of a message channel. Each {@link Message} is - * placed in a queue whose capacity may be specified upon construction. If no - * capacity is specified, the {@link #DEFAULT_CAPACITY} will be used. - * - * @author Mark Fisher - */ -public class SimpleChannel extends BaseBlockingQueueChannel { - - /** - * Create a channel with the specified queue capacity and dispatcher policy. - */ - public SimpleChannel(int capacity, DispatcherPolicy dispatcherPolicy) { - super((capacity > 0) ? new LinkedBlockingQueue>(capacity) : - new SynchronousQueue>(true), dispatcherPolicy); - } - - /** - * Create a channel with the specified queue capacity. - */ - public SimpleChannel(int capacity) { - this(capacity, null); - } - - /** - * Create a channel with the default queue capacity. - */ - public SimpleChannel() { - this(DEFAULT_CAPACITY, null); - } - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java index 66cdeb32cb..2138dd36ed 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -23,16 +23,14 @@ import org.w3c.dom.NodeList; import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.config.RuntimeBeanReference; import org.springframework.beans.factory.parsing.BeanComponentDefinition; -import org.springframework.beans.factory.support.GenericBeanDefinition; import org.springframework.beans.factory.support.ManagedList; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.beans.factory.xml.BeanDefinitionParser; import org.springframework.beans.factory.xml.ParserContext; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.PriorityChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.channel.interceptor.MessageSelectingInterceptor; -import org.springframework.integration.message.selector.MessageSelector; import org.springframework.integration.message.selector.PayloadTypeSelector; import org.springframework.util.StringUtils; @@ -62,7 +60,7 @@ public class ChannelParser implements BeanDefinitionParser { public BeanDefinition parse(Element element, ParserContext parserContext) { boolean isPriorityChannel = (element.getLocalName().equals("priority-channel")); - Class channelClass = (isPriorityChannel) ? PriorityChannel.class : SimpleChannel.class; + Class channelClass = (isPriorityChannel) ? PriorityChannel.class : QueueChannel.class; RootBeanDefinition channelDef = new RootBeanDefinition(channelClass); channelDef.setSource(parserContext.extractSource(element)); boolean isPublishSubscribe = "true".equals(element.getAttribute(PUBLISH_SUBSCRIBE_ATTRIBUTE)); @@ -83,7 +81,7 @@ public class ChannelParser implements BeanDefinitionParser { } } String capAttr = element.getAttribute(CAPACITY_ATTRIBUTE); - int capacity = (StringUtils.hasText(capAttr)) ? Integer.parseInt(capAttr) : SimpleChannel.DEFAULT_CAPACITY; + int capacity = (StringUtils.hasText(capAttr)) ? Integer.parseInt(capAttr) : QueueChannel.DEFAULT_CAPACITY; channelDef.getConstructorArgumentValues().addIndexedArgumentValue(0, capacity); channelDef.getConstructorArgumentValues().addIndexedArgumentValue(1, dispatcherPolicy); if (isPriorityChannel) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingTargetTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingTargetTests.java index 556bca288c..8061bacb95 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingTargetTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingTargetTests.java @@ -28,7 +28,7 @@ import java.util.concurrent.TimeUnit; import org.junit.Test; import org.springframework.integration.bus.MessageBus; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessagingException; @@ -86,7 +86,7 @@ public class MethodInvokingTargetTests { target.setObject(testBean); target.setMethod("foo"); target.afterPropertiesSet(); - SimpleChannel channel = new SimpleChannel(); + QueueChannel channel = new QueueChannel(); Subscription subscription = new Subscription(channel); Message message = new GenericMessage("123", "testing"); channel.send(message); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java index 470fb416bf..46a528ac14 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java @@ -24,7 +24,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.PollableSource; @@ -38,7 +38,7 @@ public class PollingSourceAdapterTests { @Test public void testPolledSourceSendsToChannel() { TestSource source = new TestSource("testing", 1); - SimpleChannel channel = new SimpleChannel(); + QueueChannel channel = new QueueChannel(); PollingSchedule schedule = new PollingSchedule(100); PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); adapter.run(); @@ -50,7 +50,7 @@ public class PollingSourceAdapterTests { @Test public void testSendTimeout() { TestSource source = new TestSource("testing", 1); - SimpleChannel channel = new SimpleChannel(1); + QueueChannel channel = new QueueChannel(1); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); @@ -71,7 +71,7 @@ public class PollingSourceAdapterTests { @Test public void testMultipleMessagesPerPoll() { TestSource source = new TestSource("testing", 3); - SimpleChannel channel = new SimpleChannel(); + QueueChannel channel = new QueueChannel(); PollingSchedule schedule = new PollingSchedule(1000); schedule.setInitialDelay(10000); PollingSourceAdapter adapter = new PollingSourceAdapter(source, channel, schedule); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml index b348789c29..565d60a5e5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/adapterTests.xml @@ -8,7 +8,7 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aop/MessagePublishingInterceptorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aop/MessagePublishingInterceptorTests.java index 6de614ca06..66a27bfe2b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aop/MessagePublishingInterceptorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aop/MessagePublishingInterceptorTests.java @@ -24,7 +24,7 @@ import org.junit.Test; import org.springframework.aop.framework.ProxyFactory; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; /** @@ -34,36 +34,34 @@ public class MessagePublishingInterceptorTests { @Test public void testNonNullReturnValuePublishedWithDefaultChannel() { - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); MessagePublishingInterceptor interceptor = new MessagePublishingInterceptor(); interceptor.setDefaultChannel(channel); TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), interceptor); proxy.messageTest(); - Message message = channel.receive(0); + Message message = channel.receive(0); assertNotNull(message); assertEquals("hello world", message.getPayload()); } @Test public void testNullReturnValueNotPublished() { - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); MessagePublishingInterceptor interceptor = new MessagePublishingInterceptor(); interceptor.setDefaultChannel(channel); TestService proxy = (TestService) this.createProxy(new TestServiceImpl(null), interceptor); proxy.messageTest(); - Message message = channel.receive(0); - assertNull(message); + assertNull(channel.receive(0)); } @Test public void testVoidReturnValueNotPublished() { - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); MessagePublishingInterceptor interceptor = new MessagePublishingInterceptor(); interceptor.setDefaultChannel(channel); TestService proxy = (TestService) this.createProxy(new TestServiceImpl(null), interceptor); proxy.voidTest(); - Message message = channel.receive(0); - assertNull(message); + assertNull(channel.receive(0)); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java index f340f28ba0..3f47290dc4 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java @@ -25,9 +25,9 @@ import org.junit.Test; import org.springframework.aop.framework.ProxyFactory; import org.springframework.integration.annotation.Publisher; import org.springframework.integration.channel.ChannelRegistry; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; import org.springframework.integration.channel.DefaultChannelRegistry; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; /** @@ -37,7 +37,7 @@ public class PublisherAnnotationAdvisorTests { @Test public void testPublisherAnnotation() { - final MessageChannel channel = new SimpleChannel(); + final MessageChannel channel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("testChannel", channel); PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelRegistry); @@ -50,7 +50,7 @@ public class PublisherAnnotationAdvisorTests { @Test public void testNoPublisherAnnotation() { - final MessageChannel channel = new SimpleChannel(); + final MessageChannel channel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("testChannel", channel); PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelRegistry); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml index b4f9cc371a..3703ba1f64 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml @@ -6,7 +6,7 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java index 809ce44c9b..9543bedf36 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -33,7 +33,8 @@ import org.springframework.integration.adapter.PollingSourceAdapter; import org.springframework.integration.adapter.SourceAdapter; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.channel.RendezvousChannel; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.ErrorMessage; @@ -53,8 +54,8 @@ public class MessageBusTests { @Test public void testOutputChannel() { MessageBus bus = new MessageBus(); - MessageChannel sourceChannel = new SimpleChannel(); - MessageChannel targetChannel = new SimpleChannel(); + MessageChannel sourceChannel = new QueueChannel(); + MessageChannel targetChannel = new QueueChannel(); bus.registerChannel("sourceChannel", sourceChannel); StringMessage message = new StringMessage("test"); message.getHeader().setReturnAddress("targetChannel"); @@ -76,9 +77,9 @@ public class MessageBusTests { @Test public void testChannelsWithoutHandlers() { MessageBus bus = new MessageBus(); - MessageChannel sourceChannel = new SimpleChannel(); + MessageChannel sourceChannel = new QueueChannel(); sourceChannel.send(new StringMessage("123", "test")); - MessageChannel targetChannel = new SimpleChannel(); + MessageChannel targetChannel = new QueueChannel(); bus.registerChannel("sourceChannel", sourceChannel); bus.registerChannel("targetChannel", targetChannel); bus.start(); @@ -102,9 +103,9 @@ public class MessageBusTests { @Test public void testExactlyOneHandlerReceivesPointToPointMessage() { - SimpleChannel inputChannel = new SimpleChannel(); - SimpleChannel outputChannel1 = new SimpleChannel(); - SimpleChannel outputChannel2 = new SimpleChannel(); + QueueChannel inputChannel = new QueueChannel(); + QueueChannel outputChannel1 = new QueueChannel(); + QueueChannel outputChannel2 = new QueueChannel(); MessageHandler handler1 = new MessageHandler() { public Message handle(Message message) { message.getHeader().setReturnAddress("output1"); @@ -134,9 +135,9 @@ public class MessageBusTests { @Test public void testBothHandlersReceivePublishSubscribeMessage() throws InterruptedException { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(true); - SimpleChannel inputChannel = new SimpleChannel(10, dispatcherPolicy); - SimpleChannel outputChannel1 = new SimpleChannel(); - SimpleChannel outputChannel2 = new SimpleChannel(); + QueueChannel inputChannel = new QueueChannel(10, dispatcherPolicy); + QueueChannel outputChannel1 = new QueueChannel(); + QueueChannel outputChannel2 = new QueueChannel(); final CountDownLatch latch = new CountDownLatch(2); MessageHandler handler1 = new MessageHandler() { public Message handle(Message message) { @@ -173,7 +174,7 @@ public class MessageBusTests { public void testErrorChannelWithFailedDispatch() throws InterruptedException { MessageBus bus = new MessageBus(); CountDownLatch latch = new CountDownLatch(1); - SourceAdapter sourceAdapter = new PollingSourceAdapter(new FailingSource(latch), new SimpleChannel(), new PollingSchedule(1000)); + SourceAdapter sourceAdapter = new PollingSourceAdapter(new FailingSource(latch), new QueueChannel(), new PollingSchedule(1000)); bus.registerSourceAdapter("testAdapter", sourceAdapter); bus.start(); latch.await(1000, TimeUnit.MILLISECONDS); @@ -204,7 +205,7 @@ public class MessageBusTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setRejectionLimit(1); dispatcherPolicy.setRetryInterval(0); - SimpleChannel testChannel = new SimpleChannel(0, dispatcherPolicy); + RendezvousChannel testChannel = new RendezvousChannel(dispatcherPolicy); bus.registerChannel("testChannel", testChannel); bus.registerHandler("testHandler", testHandler, new Subscription(testChannel)); bus.start(); @@ -234,7 +235,7 @@ public class MessageBusTests { @Test public void testErrorChannelRegistration() { - MessageChannel errorChannel = new SimpleChannel(); + MessageChannel errorChannel = new QueueChannel(); MessageBus bus = new MessageBus(); bus.setErrorChannel(errorChannel); assertEquals(errorChannel, bus.getErrorChannel()); @@ -242,7 +243,7 @@ public class MessageBusTests { @Test public void testHandlerSubscribedToErrorChannel() throws InterruptedException { - MessageChannel errorChannel = new SimpleChannel(); + MessageChannel errorChannel = new QueueChannel(); MessageBus bus = new MessageBus(); bus.setErrorChannel(errorChannel); final CountDownLatch latch = new CountDownLatch(1); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java index a6c2ea6ebd..22a8bdfa20 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java @@ -29,7 +29,7 @@ import org.junit.Test; import org.springframework.integration.bus.SubscriptionManager; import org.springframework.integration.channel.DispatcherPolicy; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.HandlerEndpoint; import org.springframework.integration.endpoint.MessageEndpoint; @@ -60,7 +60,7 @@ public class SubscriptionManagerTests { final CountDownLatch latch = new CountDownLatch(1); MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - SimpleChannel channel = new SimpleChannel(); + QueueChannel channel = new QueueChannel(); channel.send(new StringMessage(1, "test")); SubscriptionManager manager = new SubscriptionManager(channel, scheduler); manager.addTarget(createEndpoint(handler1, true)); @@ -78,7 +78,7 @@ public class SubscriptionManagerTests { final CountDownLatch latch = new CountDownLatch(2); MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true)); + QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true)); channel.send(new StringMessage(1, "test")); SubscriptionManager manager = new SubscriptionManager(channel, scheduler); manager.addTarget(createEndpoint(handler1, true)); @@ -98,7 +98,7 @@ public class SubscriptionManagerTests { MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); - SimpleChannel channel = new SimpleChannel(); + QueueChannel channel = new QueueChannel(); SubscriptionManager manager = new SubscriptionManager(channel, scheduler); MessageEndpoint inactiveEndpoint = createEndpoint(handler1, true); manager.addTarget(inactiveEndpoint); @@ -122,7 +122,7 @@ public class SubscriptionManagerTests { MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); - SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true)); + QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true)); SubscriptionManager manager = new SubscriptionManager(channel, scheduler); MessageEndpoint inactiveEndpoint = createEndpoint(handler2, true); manager.addTarget(createEndpoint(handler1, true)); @@ -139,7 +139,7 @@ public class SubscriptionManagerTests { @Test public void testDispatcherWithNoExecutorsDoesNotFail() { - SimpleChannel channel = new SimpleChannel(); + QueueChannel channel = new QueueChannel(); channel.send(new StringMessage(1, "test")); SubscriptionManager manager = new SubscriptionManager(channel, scheduler); manager.start(); @@ -152,7 +152,7 @@ public class SubscriptionManagerTests { final CountDownLatch latch = new CountDownLatch(2); MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler3 = TestHandlers.countingCountDownHandler(counter3, latch); - SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true)); + QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true)); channel.getDispatcherPolicy().setRejectionLimit(2); channel.getDispatcherPolicy().setRetryInterval(3); channel.send(new StringMessage(1, "test")); @@ -164,7 +164,7 @@ public class SubscriptionManagerTests { } }); manager.addTarget(createEndpoint(handler3, true)); - SimpleChannel errorChannel = new SimpleChannel(); + QueueChannel errorChannel = new QueueChannel(); scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel)); manager.start(); latch.await(2000, TimeUnit.MILLISECONDS); @@ -182,7 +182,7 @@ public class SubscriptionManagerTests { final CountDownLatch latch = new CountDownLatch(3); MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true)); + QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true)); channel.getDispatcherPolicy().setRejectionLimit(2); channel.getDispatcherPolicy().setRetryInterval(3); channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false); @@ -207,14 +207,14 @@ public class SubscriptionManagerTests { final CountDownLatch latch = new CountDownLatch(4); MessageHandler handler1 = TestHandlers.rejectingCountDownHandler(latch); MessageHandler handler2 = TestHandlers.rejectingCountDownHandler(latch); - SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(false)); + QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(false)); channel.send(new StringMessage(1, "test")); SubscriptionManager manager = new SubscriptionManager(channel, scheduler); channel.getDispatcherPolicy().setRejectionLimit(2); channel.getDispatcherPolicy().setRetryInterval(3); manager.addTarget(createEndpoint(handler1, false)); manager.addTarget(createEndpoint(handler2, false)); - SimpleChannel errorChannel = new SimpleChannel(); + QueueChannel errorChannel = new QueueChannel(); scheduler.setErrorHandler(new MessagePublishingErrorHandler(errorChannel)); manager.start(); latch.await(2000, TimeUnit.MILLISECONDS); @@ -232,7 +232,7 @@ public class SubscriptionManagerTests { final AtomicInteger rejectedCounter1 = new AtomicInteger(); final AtomicInteger rejectedCounter2 = new AtomicInteger(); final CountDownLatch latch = new CountDownLatch(4); - SimpleChannel channel = new SimpleChannel(); + QueueChannel channel = new QueueChannel(); channel.getDispatcherPolicy().setRejectionLimit(2); channel.getDispatcherPolicy().setRetryInterval(3); channel.getDispatcherPolicy().setShouldFailOnRejectionLimit(false); @@ -275,7 +275,7 @@ public class SubscriptionManagerTests { dispatcherPolicy.setRejectionLimit(2); dispatcherPolicy.setRetryInterval(3); dispatcherPolicy.setShouldFailOnRejectionLimit(false); - SimpleChannel channel = new SimpleChannel(25, dispatcherPolicy); + QueueChannel channel = new QueueChannel(25, dispatcherPolicy); channel.send(new StringMessage(1, "test")); SubscriptionManager manager = new SubscriptionManager(channel, scheduler); manager.addTarget(createEndpoint(new MessageHandler() { @@ -326,7 +326,7 @@ public class SubscriptionManagerTests { dispatcherPolicy.setRejectionLimit(5); dispatcherPolicy.setRetryInterval(3); dispatcherPolicy.setShouldFailOnRejectionLimit(false); - SimpleChannel channel = new SimpleChannel(25, dispatcherPolicy); + QueueChannel channel = new QueueChannel(25, dispatcherPolicy); channel.send(new StringMessage(1, "test")); SubscriptionManager manager = new SubscriptionManager(channel, scheduler); manager.addTarget(createEndpoint(new MessageHandler() { @@ -365,7 +365,7 @@ public class SubscriptionManagerTests { final CountDownLatch latch = new CountDownLatch(1); MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - SimpleChannel channel = new SimpleChannel(); + QueueChannel channel = new QueueChannel(); channel.send(new StringMessage(1, "test")); SubscriptionManager manager = new SubscriptionManager(channel, scheduler); HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1); @@ -391,7 +391,7 @@ public class SubscriptionManagerTests { final CountDownLatch handlerLatch = new CountDownLatch(1); MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, handlerLatch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, handlerLatch); - SimpleChannel channel = new SimpleChannel(); + QueueChannel channel = new QueueChannel(); channel.send(new StringMessage(1, "test")); SubscriptionManager manager = new SubscriptionManager(channel, scheduler); final HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1); @@ -431,7 +431,7 @@ public class SubscriptionManagerTests { final CountDownLatch latch = new CountDownLatch(1); MessageHandler handler1 = TestHandlers.countingCountDownHandler(counter1, latch); MessageHandler handler2 = TestHandlers.countingCountDownHandler(counter2, latch); - SimpleChannel channel = new SimpleChannel(5, new DispatcherPolicy(true)); + QueueChannel channel = new QueueChannel(5, new DispatcherPolicy(true)); channel.send(new StringMessage(1, "test")); SubscriptionManager manager = new SubscriptionManager(channel, scheduler); HandlerEndpoint endpoint1 = new HandlerEndpoint(handler1); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/SynchronousChannelSubscriptionTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/SynchronousChannelSubscriptionTests.java index 22c14e1904..e1a6f12018 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/SynchronousChannelSubscriptionTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/SynchronousChannelSubscriptionTests.java @@ -24,7 +24,7 @@ import org.junit.Test; import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.MessageEndpointAnnotationPostProcessor; import org.springframework.integration.dispatcher.SynchronousChannel; import org.springframework.integration.endpoint.HandlerEndpoint; @@ -81,7 +81,7 @@ public class SynchronousChannelSubscriptionTests { @Test(expected=MessagingException.class) public void testExceptionThrownFromRegisteredEndpoint() { - SimpleChannel errorChannel = new SimpleChannel(); + QueueChannel errorChannel = new QueueChannel(); bus.setErrorChannel(errorChannel); HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() { public Message handle(Message message) { @@ -97,7 +97,7 @@ public class SynchronousChannelSubscriptionTests { @Test(expected=MessagingException.class) public void testExceptionThrownFromAnnotatedEndpoint() { - SimpleChannel errorChannel = new SimpleChannel(); + QueueChannel errorChannel = new QueueChannel(); bus.setErrorChannel(errorChannel); MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor(bus); postProcessor.afterPropertiesSet(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml index 4e3bf0900b..ed8744c81c 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/messageBusTests.xml @@ -6,9 +6,9 @@ - + - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelPurgerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelPurgerTests.java index fa7c8f4596..19194069c1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelPurgerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelPurgerTests.java @@ -35,7 +35,7 @@ public class ChannelPurgerTests { @Test public void testPurgeAllWithoutSelector() { - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); channel.send(new StringMessage("test1")); channel.send(new StringMessage("test2")); channel.send(new StringMessage("test3")); @@ -47,7 +47,7 @@ public class ChannelPurgerTests { @Test public void testPurgeAllWithSelector() { - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); channel.send(new StringMessage("test1")); channel.send(new StringMessage("test2")); channel.send(new StringMessage("test3")); @@ -63,7 +63,7 @@ public class ChannelPurgerTests { @Test public void testPurgeNoneWithSelector() { - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); channel.send(new StringMessage("test1")); channel.send(new StringMessage("test2")); channel.send(new StringMessage("test3")); @@ -81,7 +81,7 @@ public class ChannelPurgerTests { @Test public void testPurgeSubsetWithSelector() { - MessageChannel channel = new SimpleChannel(); + MessageChannel channel = new QueueChannel(); channel.send(new StringMessage("test1")); channel.send(new StringMessage("test2")); channel.send(new StringMessage("test3")); @@ -100,8 +100,8 @@ public class ChannelPurgerTests { @Test public void testMultipleChannelsWithNoSelector() { - MessageChannel channel1 = new SimpleChannel(); - MessageChannel channel2 = new SimpleChannel(); + MessageChannel channel1 = new QueueChannel(); + MessageChannel channel2 = new QueueChannel(); channel1.send(new StringMessage("test1")); channel1.send(new StringMessage("test2")); channel2.send(new StringMessage("test1")); @@ -115,8 +115,8 @@ public class ChannelPurgerTests { @Test public void testMultipleChannelsWithSelector() { - MessageChannel channel1 = new SimpleChannel(); - MessageChannel channel2 = new SimpleChannel(); + MessageChannel channel1 = new QueueChannel(); + MessageChannel channel2 = new QueueChannel(); channel1.send(new StringMessage("test1")); channel1.send(new StringMessage("test2")); channel1.send(new StringMessage("test3")); @@ -142,8 +142,8 @@ public class ChannelPurgerTests { @Test public void testPurgeNoneWithSelectorAndMultipleChannels() { - MessageChannel channel1 = new SimpleChannel(); - MessageChannel channel2 = new SimpleChannel(); + MessageChannel channel1 = new QueueChannel(); + MessageChannel channel2 = new QueueChannel(); channel1.send(new StringMessage("test1")); channel1.send(new StringMessage("test2")); channel2.send(new StringMessage("test1")); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/DefaultChannelRegistryTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/DefaultChannelRegistryTests.java index 581a8a1227..db6f940391 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/DefaultChannelRegistryTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/DefaultChannelRegistryTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -29,7 +29,7 @@ public class DefaultChannelRegistryTests { @Test public void testLookupRegisteredChannel() { - SimpleChannel testChannel = new SimpleChannel(); + QueueChannel testChannel = new QueueChannel(); DefaultChannelRegistry registry = new DefaultChannelRegistry(); registry.registerChannel("testChannel", testChannel); MessageChannel lookedUpChannel = registry.lookupChannel("testChannel"); @@ -46,7 +46,7 @@ public class DefaultChannelRegistryTests { @Test public void testLookupUnregisteredChannel() { - SimpleChannel testChannel = new SimpleChannel(); + QueueChannel testChannel = new QueueChannel(); DefaultChannelRegistry registry = new DefaultChannelRegistry(); registry.registerChannel("testChannel", testChannel); MessageChannel lookedUpChannel1 = registry.lookupChannel("testChannel"); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/SimpleChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java similarity index 93% rename from spring-integration-core/src/test/java/org/springframework/integration/channel/SimpleChannelTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java index 75eb291586..68d919e72e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/SimpleChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java @@ -39,13 +39,13 @@ import org.springframework.integration.message.selector.UnexpiredMessageSelector /** * @author Mark Fisher */ -public class SimpleChannelTests { +public class QueueChannelTests { @Test public void testSimpleSendAndReceive() throws Exception { final AtomicBoolean messageReceived = new AtomicBoolean(false); final CountDownLatch latch = new CountDownLatch(1); - final SimpleChannel channel = new SimpleChannel(); + final QueueChannel channel = new QueueChannel(); new Thread(new Runnable() { public void run() { Message message = channel.receive(); @@ -64,7 +64,7 @@ public class SimpleChannelTests { @Test public void testImmediateReceive() throws Exception { final AtomicBoolean messageReceived = new AtomicBoolean(false); - final SimpleChannel channel = new SimpleChannel(); + final QueueChannel channel = new QueueChannel(); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); Executor singleThreadExecutor = Executors.newSingleThreadExecutor(); @@ -102,7 +102,7 @@ public class SimpleChannelTests { @Test public void testBlockingReceiveWithNoTimeout() throws Exception{ - final SimpleChannel channel = new SimpleChannel(); + final QueueChannel channel = new QueueChannel(); final AtomicBoolean receiveInterrupted = new AtomicBoolean(false); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { @@ -122,7 +122,7 @@ public class SimpleChannelTests { @Test public void testBlockingReceiveWithTimeout() throws Exception{ - final SimpleChannel channel = new SimpleChannel(); + final QueueChannel channel = new QueueChannel(); final AtomicBoolean receiveInterrupted = new AtomicBoolean(false); final CountDownLatch latch = new CountDownLatch(1); Thread t = new Thread(new Runnable() { @@ -142,7 +142,7 @@ public class SimpleChannelTests { @Test public void testImmediateSend() { - SimpleChannel channel = new SimpleChannel(3); + QueueChannel channel = new QueueChannel(3); boolean result1 = channel.send(new GenericMessage(1, "test-1")); assertTrue(result1); boolean result2 = channel.send(new GenericMessage(2, "test-2"), 100); @@ -155,7 +155,7 @@ public class SimpleChannelTests { @Test public void testBlockingSendWithNoTimeout() throws Exception{ - final SimpleChannel channel = new SimpleChannel(1); + final QueueChannel channel = new QueueChannel(1); boolean result1 = channel.send(new GenericMessage(1, "test-1")); assertTrue(result1); final AtomicBoolean sendInterrupted = new AtomicBoolean(false); @@ -176,7 +176,7 @@ public class SimpleChannelTests { @Test public void testBlockingSendWithTimeout() throws Exception{ - final SimpleChannel channel = new SimpleChannel(1); + final QueueChannel channel = new QueueChannel(1); boolean result1 = channel.send(new GenericMessage(1, "test-1")); assertTrue(result1); final AtomicBoolean sendInterrupted = new AtomicBoolean(false); @@ -197,7 +197,7 @@ public class SimpleChannelTests { @Test public void testClear() { - SimpleChannel channel = new SimpleChannel(2); + QueueChannel channel = new QueueChannel(2); StringMessage message1 = new StringMessage("test1"); StringMessage message2 = new StringMessage("test2"); StringMessage message3 = new StringMessage("test3"); @@ -212,7 +212,7 @@ public class SimpleChannelTests { @Test public void testClearEmptyChannel() { - SimpleChannel channel = new SimpleChannel(); + QueueChannel channel = new QueueChannel(); List> clearedMessages = channel.clear(); assertNotNull(clearedMessages); assertEquals(0, clearedMessages.size()); @@ -220,7 +220,7 @@ public class SimpleChannelTests { @Test public void testPurge() { - SimpleChannel channel = new SimpleChannel(2); + QueueChannel channel = new QueueChannel(2); long minute = 60 * 1000; long time = System.currentTimeMillis(); Date past = new Date(time - minute); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/RequestReplyTemplateTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/RequestReplyTemplateTests.java index 9d7fd8d016..d100a73f20 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/RequestReplyTemplateTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/RequestReplyTemplateTests.java @@ -39,7 +39,7 @@ import org.springframework.integration.scheduling.Subscription; */ public class RequestReplyTemplateTests { - private final SimpleChannel requestChannel = new SimpleChannel(); + private final QueueChannel requestChannel = new QueueChannel(); public RequestReplyTemplateTests() { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java index 31723e5123..16f045ed78 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java @@ -28,7 +28,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; @@ -37,7 +37,7 @@ import org.springframework.integration.message.StringMessage; */ public class ChannelInterceptorTests { - private final SimpleChannel channel = new SimpleChannel(); + private final QueueChannel channel = new QueueChannel(); @Test @@ -81,7 +81,7 @@ public class ChannelInterceptorTests { public void testPostSendInterceptorWithUnsentMessage() { final AtomicInteger invokedCounter = new AtomicInteger(0); final AtomicInteger sentCounter = new AtomicInteger(0); - final SimpleChannel singleItemChannel = new SimpleChannel(1); + final QueueChannel singleItemChannel = new QueueChannel(1); singleItemChannel.addInterceptor(new ChannelInterceptorAdapter() { @Override public void postSend(Message message, MessageChannel channel, boolean sent) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/MessageSelectingInterceptorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/MessageSelectingInterceptorTests.java index b34fdccbc5..f52c459250 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/MessageSelectingInterceptorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/MessageSelectingInterceptorTests.java @@ -23,7 +23,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.StringMessage; @@ -39,7 +39,7 @@ public class MessageSelectingInterceptorTests { final AtomicInteger counter = new AtomicInteger(); MessageSelector selector = new TestMessageSelector(true, counter); MessageSelectingInterceptor interceptor = new MessageSelectingInterceptor(selector); - SimpleChannel channel = new SimpleChannel(); + QueueChannel channel = new QueueChannel(); channel.addInterceptor(interceptor); assertTrue(channel.send(new StringMessage("test1"))); } @@ -49,7 +49,7 @@ public class MessageSelectingInterceptorTests { final AtomicInteger counter = new AtomicInteger(); MessageSelector selector = new TestMessageSelector(false, counter); MessageSelectingInterceptor interceptor = new MessageSelectingInterceptor(selector); - SimpleChannel channel = new SimpleChannel(); + QueueChannel channel = new QueueChannel(); channel.addInterceptor(interceptor); channel.send(new StringMessage("test1")); } @@ -60,7 +60,7 @@ public class MessageSelectingInterceptorTests { MessageSelector selector1 = new TestMessageSelector(true, counter); MessageSelector selector2 = new TestMessageSelector(true, counter); MessageSelectingInterceptor interceptor = new MessageSelectingInterceptor(selector1, selector2); - SimpleChannel channel = new SimpleChannel(); + QueueChannel channel = new QueueChannel(); channel.addInterceptor(interceptor); assertTrue(channel.send(new StringMessage("test1"))); assertEquals(2, counter.get()); @@ -75,7 +75,7 @@ public class MessageSelectingInterceptorTests { MessageSelector selector3 = new TestMessageSelector(false, counter); MessageSelector selector4 = new TestMessageSelector(true, counter); MessageSelectingInterceptor interceptor = new MessageSelectingInterceptor(selector1, selector2, selector3, selector4); - SimpleChannel channel = new SimpleChannel(); + QueueChannel channel = new QueueChannel(); channel.addInterceptor(interceptor); try { channel.send(new StringMessage("test1")); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java index 49c19c2441..3188c4a8f3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java @@ -27,7 +27,7 @@ import java.util.List; import org.junit.Test; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; import org.springframework.integration.message.selector.MessageSelector; @@ -39,8 +39,8 @@ public class WireTapTests { @Test public void testWireTapWithNoSelectors() { - SimpleChannel mainChannel = new SimpleChannel(); - SimpleChannel secondaryChannel = new SimpleChannel(); + QueueChannel mainChannel = new QueueChannel(); + QueueChannel secondaryChannel = new QueueChannel(); mainChannel.addInterceptor(new WireTap(secondaryChannel)); mainChannel.send(new StringMessage("testing")); Message original = mainChannel.receive(0); @@ -51,8 +51,8 @@ public class WireTapTests { @Test public void testWireTapWithRejectingSelector() { - SimpleChannel mainChannel = new SimpleChannel(); - SimpleChannel secondaryChannel = new SimpleChannel(); + QueueChannel mainChannel = new QueueChannel(); + QueueChannel secondaryChannel = new QueueChannel(); List selectors = new ArrayList(); selectors.add(new TestSelector(true)); selectors.add(new TestSelector(false)); @@ -66,8 +66,8 @@ public class WireTapTests { @Test public void testWireTapWithAcceptingSelectors() { - SimpleChannel mainChannel = new SimpleChannel(); - SimpleChannel secondaryChannel = new SimpleChannel(); + QueueChannel mainChannel = new QueueChannel(); + QueueChannel secondaryChannel = new QueueChannel(); List selectors = new ArrayList(); selectors.add(new TestSelector(true)); selectors.add(new TestSelector(true)); @@ -81,8 +81,8 @@ public class WireTapTests { @Test public void testNewMessageIdGeneratedForDuplicate() { - SimpleChannel mainChannel = new SimpleChannel(); - SimpleChannel secondaryChannel = new SimpleChannel(); + QueueChannel mainChannel = new QueueChannel(); + QueueChannel secondaryChannel = new QueueChannel(); mainChannel.addInterceptor(new WireTap(secondaryChannel)); mainChannel.send(new StringMessage("testing")); Message original = mainChannel.receive(0); @@ -94,8 +94,8 @@ public class WireTapTests { @Test public void testOriginalIdStoredAsAttribute() { - SimpleChannel mainChannel = new SimpleChannel(); - SimpleChannel secondaryChannel = new SimpleChannel(); + QueueChannel mainChannel = new QueueChannel(); + QueueChannel secondaryChannel = new QueueChannel(); mainChannel.addInterceptor(new WireTap(secondaryChannel)); mainChannel.send(new StringMessage("testing")); Message original = mainChannel.receive(0); @@ -107,8 +107,8 @@ public class WireTapTests { @Test public void testNewTimestampGeneratedForDuplicate() throws InterruptedException { - SimpleChannel mainChannel = new SimpleChannel(); - SimpleChannel secondaryChannel = new SimpleChannel(); + QueueChannel mainChannel = new QueueChannel(); + QueueChannel secondaryChannel = new QueueChannel(); mainChannel.addInterceptor(new WireTap(secondaryChannel)); Message message = new StringMessage("testing"); Thread.sleep(3); @@ -120,8 +120,8 @@ public class WireTapTests { } public void testDuplicateMessageContainsAttribute() { - SimpleChannel mainChannel = new SimpleChannel(); - SimpleChannel secondaryChannel = new SimpleChannel(); + QueueChannel mainChannel = new QueueChannel(); + QueueChannel secondaryChannel = new QueueChannel(); mainChannel.addInterceptor(new WireTap(secondaryChannel)); Message message = new StringMessage("testing"); String attributeKey = "testAttribute"; @@ -139,8 +139,8 @@ public class WireTapTests { @Test public void testDuplicateMessageContainsProperty() { - SimpleChannel mainChannel = new SimpleChannel(); - SimpleChannel secondaryChannel = new SimpleChannel(); + QueueChannel mainChannel = new QueueChannel(); + QueueChannel secondaryChannel = new QueueChannel(); mainChannel.addInterceptor(new WireTap(secondaryChannel)); Message message = new StringMessage("testing"); String propertyKey = "testProperty"; diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java index 06f4dd6ce8..12e8815a50 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/EndpointParserTests.java @@ -28,7 +28,7 @@ import org.junit.Test; import org.springframework.context.Lifecycle; import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.HandlerEndpoint; import org.springframework.integration.message.GenericMessage; @@ -111,7 +111,7 @@ public class EndpointParserTests { Target endpoint = (Target) context.getBean("endpoint"); ((Lifecycle) endpoint).start(); Message message = new StringMessage("test"); - MessageChannel replyChannel = new SimpleChannel(); + MessageChannel replyChannel = new QueueChannel(); message.getHeader().setReturnAddress(replyChannel); endpoint.send(message); Message reply = replyChannel.receive(500); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/SubscriberAnnotationPostProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/SubscriberAnnotationPostProcessorTests.java index 329d2b0a89..2522ac28cd 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/SubscriberAnnotationPostProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/SubscriberAnnotationPostProcessorTests.java @@ -34,7 +34,7 @@ import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.annotation.Subscriber; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.StringMessage; /** @@ -46,7 +46,7 @@ public class SubscriberAnnotationPostProcessorTests { public void testAnnotatedSubscriber() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); GenericApplicationContext context = new GenericApplicationContext(); - context.registerBeanDefinition("testChannel", new RootBeanDefinition(SimpleChannel.class)); + context.registerBeanDefinition("testChannel", new RootBeanDefinition(QueueChannel.class)); RootBeanDefinition subscriberDef = new RootBeanDefinition(SubscriberAnnotationTestBean.class); subscriberDef.getConstructorArgumentValues().addGenericArgumentValue(latch); context.registerBeanDefinition("testBean", subscriberDef); @@ -72,7 +72,7 @@ public class SubscriberAnnotationPostProcessorTests { public void testCustomAnnotation() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); GenericApplicationContext context = new GenericApplicationContext(); - context.registerBeanDefinition("testChannel", new RootBeanDefinition(SimpleChannel.class)); + context.registerBeanDefinition("testChannel", new RootBeanDefinition(QueueChannel.class)); RootBeanDefinition subscriberDef = new RootBeanDefinition(CustomAnnotationTestBean.class); subscriberDef.getConstructorArgumentValues().addGenericArgumentValue(latch); context.registerBeanDefinition("testBean", subscriberDef); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SourcePollerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SourcePollerTests.java index 31bce8e325..53e893d0c0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SourcePollerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SourcePollerTests.java @@ -26,7 +26,7 @@ import org.junit.Test; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; @@ -39,7 +39,7 @@ public class SourcePollerTests { public void testSingleMessagePerRetrieval() { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setReceiveTimeout(0); - MessageChannel channel = new SimpleChannel(5, dispatcherPolicy); + MessageChannel channel = new QueueChannel(5, dispatcherPolicy); SourcePoller poller = new SourcePoller(channel); Collection> results = poller.poll(); assertTrue(results.isEmpty()); @@ -58,7 +58,7 @@ public class SourcePollerTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(2); - MessageChannel channel = new SimpleChannel(5, dispatcherPolicy); + MessageChannel channel = new QueueChannel(5, dispatcherPolicy); SourcePoller poller = new SourcePoller(channel); Collection> results = poller.poll(); assertTrue(results.isEmpty()); @@ -80,7 +80,7 @@ public class SourcePollerTests { DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); dispatcherPolicy.setReceiveTimeout(0); dispatcherPolicy.setMaxMessagesPerTask(1); - MessageChannel channel = new SimpleChannel(5, dispatcherPolicy); + MessageChannel channel = new QueueChannel(5, dispatcherPolicy); SourcePoller poller = new SourcePoller(channel); Collection> results = poller.poll(); assertTrue(results.isEmpty()); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java index 3a4c97b6a8..21d9f1b530 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java @@ -31,7 +31,7 @@ import org.junit.Test; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.handler.MessageHandlerNotRunningException; import org.springframework.integration.handler.TestHandlers; @@ -48,7 +48,7 @@ public class HandlerEndpointTests { @Test public void testDefaultReplyChannel() throws Exception { - MessageChannel replyChannel = new SimpleChannel(); + MessageChannel replyChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("replyChannel", replyChannel); MessageHandler handler = new MessageHandler() { @@ -69,7 +69,7 @@ public class HandlerEndpointTests { @Test public void testExplicitReplyChannel() throws Exception { - final MessageChannel replyChannel = new SimpleChannel(); + final MessageChannel replyChannel = new QueueChannel(); MessageHandler handler = new MessageHandler() { public Message handle(Message message) { return new StringMessage("123", "hello " + message.getPayload()); @@ -88,7 +88,7 @@ public class HandlerEndpointTests { @Test public void testExplicitReplyChannelName() throws Exception { - final MessageChannel replyChannel = new SimpleChannel(); + final MessageChannel replyChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("replyChannel", replyChannel); MessageHandler handler = new MessageHandler() { @@ -110,8 +110,8 @@ public class HandlerEndpointTests { @Test public void testDynamicReplyChannel() throws Exception { - final MessageChannel replyChannel1 = new SimpleChannel(); - final MessageChannel replyChannel2 = new SimpleChannel(); + final MessageChannel replyChannel1 = new QueueChannel(); + final MessageChannel replyChannel2 = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("replyChannel2", replyChannel2); MessageHandler handler = new MessageHandler() { @@ -158,7 +158,7 @@ public class HandlerEndpointTests { @Test public void testConcurrentHandlerWithDefaultReplyChannel() throws InterruptedException { - MessageChannel replyChannel = new SimpleChannel(); + MessageChannel replyChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("replyChannel", replyChannel); final CountDownLatch latch = new CountDownLatch(1); @@ -184,7 +184,7 @@ public class HandlerEndpointTests { @Test public void testHandlerReturnsNull() throws InterruptedException { - MessageChannel replyChannel = new SimpleChannel(); + MessageChannel replyChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("replyChannel", replyChannel); final CountDownLatch latch = new CountDownLatch(1); @@ -208,7 +208,7 @@ public class HandlerEndpointTests { @Test public void testConcurrentHandlerReturnsNull() throws InterruptedException { - MessageChannel replyChannel = new SimpleChannel(); + MessageChannel replyChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("replyChannel", replyChannel); final CountDownLatch latch = new CountDownLatch(1); @@ -233,7 +233,7 @@ public class HandlerEndpointTests { @Test public void testConcurrentHandlerWithExplicitReplyChannel() throws InterruptedException { - MessageChannel replyChannel = new SimpleChannel(); + MessageChannel replyChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("replyChannel", replyChannel); final CountDownLatch latch = new CountDownLatch(1); @@ -260,7 +260,7 @@ public class HandlerEndpointTests { @Test public void testGeneratedConcurrentHandlerWithDefaultReplyChannel() throws InterruptedException { - MessageChannel replyChannel = new SimpleChannel(); + MessageChannel replyChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("replyChannel", replyChannel); final CountDownLatch latch = new CountDownLatch(1); @@ -286,7 +286,7 @@ public class HandlerEndpointTests { @Test public void testGeneratedConcurrentHandlerWithExplicitReplyChannel() throws InterruptedException { - MessageChannel replyChannel = new SimpleChannel(); + MessageChannel replyChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("replyChannel", replyChannel); final CountDownLatch latch = new CountDownLatch(1); @@ -433,7 +433,7 @@ public class HandlerEndpointTests { @Test public void testDefaultOutputChannelTimeoutSendsToErrorHandler() { - SimpleChannel output = new SimpleChannel(1); + QueueChannel output = new QueueChannel(1); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("output", output); HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() { @@ -458,7 +458,7 @@ public class HandlerEndpointTests { @Test public void testReturnAddressChannelTimeoutSendsToErrorHandler() { - SimpleChannel replyChannel = new SimpleChannel(1); + QueueChannel replyChannel = new QueueChannel(1); HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() { public Message handle(Message message) { return message; @@ -483,7 +483,7 @@ public class HandlerEndpointTests { @Test public void testReturnAddressChannelNameTimeoutSendsToErrorHandler() { - SimpleChannel replyChannel = new SimpleChannel(1); + QueueChannel replyChannel = new QueueChannel(1); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("replyChannel", replyChannel); HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() { @@ -511,7 +511,7 @@ public class HandlerEndpointTests { @Test public void testCorrelationId() { - SimpleChannel replyChannel = new SimpleChannel(1); + QueueChannel replyChannel = new QueueChannel(1); HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() { public Message handle(Message message) { return message; @@ -527,7 +527,7 @@ public class HandlerEndpointTests { @Test public void testCorrelationIdSetByHandlerTakesPrecedence() { - SimpleChannel replyChannel = new SimpleChannel(1); + QueueChannel replyChannel = new QueueChannel(1); HandlerEndpoint endpoint = new HandlerEndpoint(new MessageHandler() { public Message handle(Message message) { message.getHeader().setCorrelationId("ABC-123"); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java index 7c1e3237ae..52fe63d209 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java @@ -39,7 +39,7 @@ import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.MessageEndpointAnnotationPostProcessor; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.HandlerEndpoint; @@ -63,6 +63,20 @@ public class MessageEndpointAnnotationPostProcessorTests { context.stop(); } + @Test + public void testSimpleHandlerWithAutoCreatedChannels() throws InterruptedException { + AbstractApplicationContext context = new ClassPathXmlApplicationContext( + "simpleAnnotatedEndpointWithAutoCreateChannelTests.xml", this.getClass()); + context.start(); + ChannelRegistry channelRegistry = (ChannelRegistry) context.getBean("bus"); + MessageChannel inputChannel = channelRegistry.lookupChannel("inputChannel"); + MessageChannel outputChannel = channelRegistry.lookupChannel("outputChannel"); + inputChannel.send(new StringMessage("world")); + Message message = outputChannel.receive(1000); + assertEquals("hello world", message.getPayload()); + context.stop(); + } + @Test public void testMessageParameterHandler() throws InterruptedException { AbstractApplicationContext context = new ClassPathXmlApplicationContext("messageParameterAnnotatedEndpointTests.xml", this.getClass()); @@ -90,7 +104,7 @@ public class MessageEndpointAnnotationPostProcessorTests { @Test public void testPolledAnnotation() throws InterruptedException { MessageBus messageBus = new MessageBus(); - SimpleChannel testChannel = new SimpleChannel(); + QueueChannel testChannel = new QueueChannel(); messageBus.registerChannel("testChannel", testChannel); MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor(messageBus); @@ -106,7 +120,7 @@ public class MessageEndpointAnnotationPostProcessorTests { @Test public void testDefaultOutputAnnotation() throws InterruptedException { MessageBus messageBus = new MessageBus(); - SimpleChannel testChannel = new SimpleChannel(); + QueueChannel testChannel = new QueueChannel(); messageBus.registerChannel("testChannel", testChannel); MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor(messageBus); @@ -212,8 +226,8 @@ public class MessageEndpointAnnotationPostProcessorTests { @Test public void testMessageEndpointAnnotationInheritedFromInterface() { MessageBus messageBus = new MessageBus(); - MessageChannel inputChannel = new SimpleChannel(); - MessageChannel outputChannel = new SimpleChannel(); + MessageChannel inputChannel = new QueueChannel(); + MessageChannel outputChannel = new QueueChannel(); messageBus.registerChannel("inputChannel", inputChannel); messageBus.registerChannel("outputChannel", outputChannel); MessageEndpointAnnotationPostProcessor postProcessor = @@ -245,8 +259,8 @@ public class MessageEndpointAnnotationPostProcessorTests { @Test public void testMessageEndpointAnnotationInheritedFromInterfaceWithProxy() { MessageBus messageBus = new MessageBus(); - MessageChannel inputChannel = new SimpleChannel(); - MessageChannel outputChannel = new SimpleChannel(); + MessageChannel inputChannel = new QueueChannel(); + MessageChannel outputChannel = new QueueChannel(); messageBus.registerChannel("inputChannel", inputChannel); messageBus.registerChannel("outputChannel", outputChannel); MessageEndpointAnnotationPostProcessor postProcessor = @@ -264,8 +278,8 @@ public class MessageEndpointAnnotationPostProcessorTests { @Test public void testSplitterAnnotation() throws InterruptedException { MessageBus messageBus = new MessageBus(); - SimpleChannel input = new SimpleChannel(); - SimpleChannel output = new SimpleChannel(); + QueueChannel input = new QueueChannel(); + QueueChannel output = new QueueChannel(); messageBus.registerChannel("input", input); messageBus.registerChannel("output", output); MessageEndpointAnnotationPostProcessor postProcessor = @@ -293,7 +307,7 @@ public class MessageEndpointAnnotationPostProcessorTests { @Test(expected=ConfigurationException.class) public void testEndpointWithNoHandlerMethod() { MessageBus messageBus = new MessageBus(); - SimpleChannel testChannel = new SimpleChannel(); + QueueChannel testChannel = new QueueChannel(); messageBus.registerChannel("testChannel", testChannel); MessageEndpointAnnotationPostProcessor postProcessor = new MessageEndpointAnnotationPostProcessor(messageBus); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml index eff7af7b4b..0b944a260e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml @@ -10,10 +10,8 @@ - - - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java index c88e805771..9b5f3b6b43 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java @@ -27,7 +27,7 @@ import org.junit.Test; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; @@ -123,7 +123,7 @@ public class CorrelationIdTests { adapter.setObject(new TestBean()); adapter.setMethodName("upperCase"); adapter.afterPropertiesSet(); - MessageChannel testChannel = new SimpleChannel(); + MessageChannel testChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("testChannel", testChannel); Map attributes = new HashMap(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java index 566ba00a62..f631943e0b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java @@ -30,7 +30,7 @@ import java.util.concurrent.TimeUnit; import org.junit.Test; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.StringMessage; @@ -54,7 +54,7 @@ public class AggregatingMessageHandlerTests { @Test public void testCompleteGroupWithinTimeout() throws InterruptedException { AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator()); - SimpleChannel replyChannel = new SimpleChannel(); + QueueChannel replyChannel = new QueueChannel(); Message message1 = createMessage("123", "ABC", 3, 1, replyChannel); Message message2 = createMessage("456", "ABC", 3, 2, replyChannel); Message message3 = createMessage("789", "ABC", 3, 3, replyChannel); @@ -70,12 +70,12 @@ public class AggregatingMessageHandlerTests { @Test public void testShouldNotSendPartialResultOnTimeoutByDefault() throws InterruptedException { - SimpleChannel discardChannel = new SimpleChannel(); + QueueChannel discardChannel = new QueueChannel(); AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator()); aggregator.setTimeout(50); aggregator.setReaperInterval(10); aggregator.setDiscardChannel(discardChannel); - SimpleChannel replyChannel = new SimpleChannel(); + QueueChannel replyChannel = new QueueChannel(); Message message = createMessage("123", "ABC", 2, 1, replyChannel); CountDownLatch latch = new CountDownLatch(1); AggregatorTestTask task = new AggregatorTestTask(aggregator, message, latch); @@ -94,7 +94,7 @@ public class AggregatingMessageHandlerTests { aggregator.setTimeout(50); aggregator.setReaperInterval(10); aggregator.setSendPartialResultOnTimeout(true); - SimpleChannel replyChannel = new SimpleChannel(); + QueueChannel replyChannel = new QueueChannel(); Message message1 = createMessage("123", "ABC", 3, 1, replyChannel); Message message2 = createMessage("456", "ABC", 3, 2, replyChannel); CountDownLatch latch = new CountDownLatch(2); @@ -114,8 +114,8 @@ public class AggregatingMessageHandlerTests { @Test public void testMultipleGroupsSimultaneously() throws InterruptedException { AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator()); - SimpleChannel replyChannel1 = new SimpleChannel(); - SimpleChannel replyChannel2 = new SimpleChannel(); + QueueChannel replyChannel1 = new QueueChannel(); + QueueChannel replyChannel2 = new QueueChannel(); Message message1 = createMessage("123", "ABC", 3, 1, replyChannel1); Message message2 = createMessage("456", "ABC", 3, 2, replyChannel1); Message message3 = createMessage("789", "ABC", 3, 3, replyChannel1); @@ -140,8 +140,8 @@ public class AggregatingMessageHandlerTests { @Test public void testDiscardChannelForTrackedCorrelationId() { - SimpleChannel replyChannel = new SimpleChannel(); - SimpleChannel discardChannel = new SimpleChannel(); + QueueChannel replyChannel = new QueueChannel(); + QueueChannel discardChannel = new QueueChannel(); AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator()); aggregator.setDiscardChannel(discardChannel); aggregator.handle(createMessage("test-1a", 1, 1, 1, replyChannel)); @@ -152,8 +152,8 @@ public class AggregatingMessageHandlerTests { @Test public void testTrackedCorrelationIdsCapacityAtLimit() { - SimpleChannel replyChannel = new SimpleChannel(); - SimpleChannel discardChannel = new SimpleChannel(); + QueueChannel replyChannel = new QueueChannel(); + QueueChannel discardChannel = new QueueChannel(); AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator()); aggregator.setTrackedCorrelationIdCapacity(3); aggregator.setDiscardChannel(discardChannel); @@ -169,8 +169,8 @@ public class AggregatingMessageHandlerTests { @Test public void testTrackedCorrelationIdsCapacityPassesLimit() { - SimpleChannel replyChannel = new SimpleChannel(); - SimpleChannel discardChannel = new SimpleChannel(); + QueueChannel replyChannel = new QueueChannel(); + QueueChannel discardChannel = new QueueChannel(); AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator()); aggregator.setTrackedCorrelationIdCapacity(3); aggregator.setDiscardChannel(discardChannel); @@ -190,7 +190,7 @@ public class AggregatingMessageHandlerTests { @Test(expected=MessageHandlingException.class) public void testExceptionThrownIfNoCorrelationId() throws InterruptedException { AggregatingMessageHandler aggregator = new AggregatingMessageHandler(new TestAggregator()); - Message message = createMessage("123", null, 2, 1, new SimpleChannel()); + Message message = createMessage("123", null, 2, 1, new QueueChannel()); aggregator.handle(message); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/MultiChannelRouterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/MultiChannelRouterTests.java index e0891173c9..387f8a6f1d 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/MultiChannelRouterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/MultiChannelRouterTests.java @@ -28,7 +28,7 @@ import org.springframework.integration.ConfigurationException; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.StringMessage; @@ -40,8 +40,8 @@ public class MultiChannelRouterTests { @Test public void testRoutingWithChannelResolver() { - final SimpleChannel channel1 = new SimpleChannel(); - final SimpleChannel channel2 = new SimpleChannel(); + final QueueChannel channel1 = new QueueChannel(); + final QueueChannel channel2 = new QueueChannel(); MultiChannelResolver channelResolver = new MultiChannelResolver() { public List resolve(Message message) { List channels = new ArrayList(); @@ -70,8 +70,8 @@ public class MultiChannelRouterTests { return new String[] {"channel1", "channel2"}; } }; - SimpleChannel channel1 = new SimpleChannel(); - SimpleChannel channel2 = new SimpleChannel(); + QueueChannel channel1 = new QueueChannel(); + QueueChannel channel2 = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("channel1", channel1); channelRegistry.registerChannel("channel2", channel2); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/PayloadTypeRouterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/PayloadTypeRouterTests.java index 35e64e38c7..5c62c698c8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/PayloadTypeRouterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/PayloadTypeRouterTests.java @@ -25,7 +25,7 @@ import java.util.concurrent.ConcurrentHashMap; import org.junit.Test; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; @@ -37,8 +37,8 @@ public class PayloadTypeRouterTests { @Test public void testRoutingByPayloadType() { - SimpleChannel stringChannel = new SimpleChannel(); - SimpleChannel integerChannel = new SimpleChannel(); + QueueChannel stringChannel = new QueueChannel(); + QueueChannel integerChannel = new QueueChannel(); Map, MessageChannel> channelMappings = new ConcurrentHashMap, MessageChannel>(); channelMappings.put(String.class, stringChannel); channelMappings.put(Integer.class, integerChannel); @@ -59,8 +59,8 @@ public class PayloadTypeRouterTests { @Test public void testRoutingToDefaultChannelWhenNoTypeMatches() { - SimpleChannel stringChannel = new SimpleChannel(); - SimpleChannel defaultChannel = new SimpleChannel(); + QueueChannel stringChannel = new QueueChannel(); + QueueChannel defaultChannel = new QueueChannel(); Map, MessageChannel> channelMappings = new ConcurrentHashMap, MessageChannel>(); channelMappings.put(String.class, stringChannel); PayloadTypeRouter router = new PayloadTypeRouter(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/RecipientListRouterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/RecipientListRouterTests.java index 4d32591742..cffb8d3f35 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/RecipientListRouterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/RecipientListRouterTests.java @@ -29,7 +29,7 @@ import org.springframework.integration.ConfigurationException; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; @@ -40,8 +40,8 @@ public class RecipientListRouterTests { @Test public void testRoutingWithChannelList() { - SimpleChannel channel1 = new SimpleChannel(); - SimpleChannel channel2 = new SimpleChannel(); + QueueChannel channel1 = new QueueChannel(); + QueueChannel channel2 = new QueueChannel(); List channels = new ArrayList(); channels.add(channel1); channels.add(channel2); @@ -60,8 +60,8 @@ public class RecipientListRouterTests { @Test public void testRoutingWithChannelNames() { - SimpleChannel channel1 = new SimpleChannel(); - SimpleChannel channel2 = new SimpleChannel(); + QueueChannel channel1 = new QueueChannel(); + QueueChannel channel2 = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("channel1", channel1); channelRegistry.registerChannel("channel2", channel2); @@ -81,8 +81,8 @@ public class RecipientListRouterTests { @Test public void testRoutingToSingleChannelByName() { - SimpleChannel channel1 = new SimpleChannel(); - SimpleChannel channel2 = new SimpleChannel(); + QueueChannel channel1 = new QueueChannel(); + QueueChannel channel2 = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("channel1", channel1); channelRegistry.registerChannel("channel2", channel2); @@ -101,8 +101,8 @@ public class RecipientListRouterTests { @Test(expected=ConfigurationException.class) public void testConfigurationExceptionWhenBothChannelsAndNamesAreProvided() { - SimpleChannel channel1 = new SimpleChannel(); - SimpleChannel channel2 = new SimpleChannel(); + QueueChannel channel1 = new QueueChannel(); + QueueChannel channel2 = new QueueChannel(); List channels = new ArrayList(); channels.add(channel1); channels.add(channel2); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/RouterMessageHandlerAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/RouterMessageHandlerAdapterTests.java index 64174ad183..3d5d3500bc 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/RouterMessageHandlerAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/RouterMessageHandlerAdapterTests.java @@ -33,7 +33,7 @@ import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.ChannelRegistryAware; import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; @@ -50,7 +50,7 @@ public class RouterMessageHandlerAdapterTests { Map attribs = new ConcurrentHashMap(); RouterMessageHandlerAdapter adapter = new RouterMessageHandlerAdapter(testBean, routingMethod, attribs); Message message = new GenericMessage("123", "bar"); - SimpleChannel barChannel = new SimpleChannel(); + QueueChannel barChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("bar-channel", barChannel); adapter.setChannelRegistry(channelRegistry); @@ -70,8 +70,8 @@ public class RouterMessageHandlerAdapterTests { RouterMessageHandlerAdapter adapter = new RouterMessageHandlerAdapter(testBean, routingMethod, attribs); Message message = new GenericMessage("123", "bar"); message.getHeader().setProperty("returnAddress", "baz"); - SimpleChannel barChannel = new SimpleChannel(); - SimpleChannel bazChannel = new SimpleChannel(); + QueueChannel barChannel = new QueueChannel(); + QueueChannel bazChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("bar-channel", barChannel); channelRegistry.registerChannel("baz-channel", bazChannel); @@ -95,9 +95,9 @@ public class RouterMessageHandlerAdapterTests { Message message = new GenericMessage("123", "bar"); message.getHeader().setProperty("returnAddress", "bad"); message.getHeader().setAttribute("returnAddress", "baz"); - SimpleChannel barChannel = new SimpleChannel(); - SimpleChannel badChannel = new SimpleChannel(); - SimpleChannel bazChannel = new SimpleChannel(); + QueueChannel barChannel = new QueueChannel(); + QueueChannel badChannel = new QueueChannel(); + QueueChannel bazChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("bar-channel", barChannel); channelRegistry.registerChannel("bad-channel", badChannel); @@ -135,8 +135,8 @@ public class RouterMessageHandlerAdapterTests { Message fooMessage = new StringMessage("foo"); Message barMessage = new StringMessage("bar"); Message badMessage = new StringMessage("bad"); - SimpleChannel fooChannel = new SimpleChannel(); - SimpleChannel barChannel = new SimpleChannel(); + QueueChannel fooChannel = new QueueChannel(); + QueueChannel barChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("foo-channel", fooChannel); channelRegistry.registerChannel("bar-channel", barChannel); @@ -159,8 +159,8 @@ public class RouterMessageHandlerAdapterTests { @Test public void testChannelInstanceResolutionByPayload() throws Exception { - SimpleChannel fooChannel = new SimpleChannel(); - SimpleChannel barChannel = new SimpleChannel(); + QueueChannel fooChannel = new QueueChannel(); + QueueChannel barChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("foo-channel", fooChannel); channelRegistry.registerChannel("bar-channel", barChannel); @@ -190,8 +190,8 @@ public class RouterMessageHandlerAdapterTests { @Test public void testChannelInstanceResolutionByMessage() throws Exception { - SimpleChannel fooChannel = new SimpleChannel(); - SimpleChannel barChannel = new SimpleChannel(); + QueueChannel fooChannel = new QueueChannel(); + QueueChannel barChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("foo-channel", fooChannel); channelRegistry.registerChannel("bar-channel", barChannel); @@ -221,8 +221,8 @@ public class RouterMessageHandlerAdapterTests { @Test public void testMultiChannelNameResolutionByPayload() throws Exception { - SimpleChannel fooChannel = new SimpleChannel(); - SimpleChannel barChannel = new SimpleChannel(); + QueueChannel fooChannel = new QueueChannel(); + QueueChannel barChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("foo-channel", fooChannel); channelRegistry.registerChannel("bar-channel", barChannel); @@ -258,8 +258,8 @@ public class RouterMessageHandlerAdapterTests { @Test public void testMultiChannelNameResolutionByMessage() throws Exception { - SimpleChannel fooChannel = new SimpleChannel(); - SimpleChannel barChannel = new SimpleChannel(); + QueueChannel fooChannel = new QueueChannel(); + QueueChannel barChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("foo-channel", fooChannel); channelRegistry.registerChannel("bar-channel", barChannel); @@ -295,8 +295,8 @@ public class RouterMessageHandlerAdapterTests { @Test public void testMultiChannelNameArrayResolutionByMessage() throws Exception { - SimpleChannel fooChannel = new SimpleChannel(); - SimpleChannel barChannel = new SimpleChannel(); + QueueChannel fooChannel = new QueueChannel(); + QueueChannel barChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("foo-channel", fooChannel); channelRegistry.registerChannel("bar-channel", barChannel); @@ -332,8 +332,8 @@ public class RouterMessageHandlerAdapterTests { @Test public void testMultiChannelListResolutionByPayload() throws Exception { - SimpleChannel fooChannel = new SimpleChannel(); - SimpleChannel barChannel = new SimpleChannel(); + QueueChannel fooChannel = new QueueChannel(); + QueueChannel barChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("foo-channel", fooChannel); channelRegistry.registerChannel("bar-channel", barChannel); @@ -369,8 +369,8 @@ public class RouterMessageHandlerAdapterTests { @Test public void testMultiChannelListResolutionByMessage() throws Exception { - SimpleChannel fooChannel = new SimpleChannel(); - SimpleChannel barChannel = new SimpleChannel(); + QueueChannel fooChannel = new QueueChannel(); + QueueChannel barChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("foo-channel", fooChannel); channelRegistry.registerChannel("bar-channel", barChannel); @@ -406,8 +406,8 @@ public class RouterMessageHandlerAdapterTests { @Test public void testMultiChannelArrayResolutionByMessage() throws Exception { - SimpleChannel fooChannel = new SimpleChannel(); - SimpleChannel barChannel = new SimpleChannel(); + QueueChannel fooChannel = new QueueChannel(); + QueueChannel barChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("foo-channel", fooChannel); channelRegistry.registerChannel("bar-channel", barChannel); @@ -443,7 +443,7 @@ public class RouterMessageHandlerAdapterTests { @Test public void testChannelRegistryAwareTarget() throws Exception { - SimpleChannel fooChannel = new SimpleChannel(); + QueueChannel fooChannel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("foo-channel", fooChannel); ChannelRegistryAwareTestBean testBean = new ChannelRegistryAwareTestBean(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/SingleChannelRouterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/SingleChannelRouterTests.java index 458b41e6c4..382859fbec 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/SingleChannelRouterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/SingleChannelRouterTests.java @@ -25,7 +25,7 @@ import org.springframework.integration.ConfigurationException; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.DefaultChannelRegistry; import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageHandlingException; import org.springframework.integration.message.StringMessage; @@ -37,7 +37,7 @@ public class SingleChannelRouterTests { @Test public void testRoutingWithChannelResolver() { - final SimpleChannel channel = new SimpleChannel(); + final QueueChannel channel = new QueueChannel(); ChannelResolver channelResolver = new ChannelResolver() { public MessageChannel resolve(Message message) { return channel; @@ -60,7 +60,7 @@ public class SingleChannelRouterTests { return "testChannel"; } }; - SimpleChannel channel = new SimpleChannel(); + QueueChannel channel = new QueueChannel(); ChannelRegistry channelRegistry = new DefaultChannelRegistry(); channelRegistry.registerChannel("testChannel", channel); SingleChannelRouter router = new SingleChannelRouter(); @@ -78,7 +78,7 @@ public class SingleChannelRouterTests { public void testConfiguringBothChannelResolverAndChannelNameResolverIsNotAllowed() { ChannelResolver channelResolver = new ChannelResolver() { public MessageChannel resolve(Message message) { - return new SimpleChannel(); + return new QueueChannel(); } }; ChannelNameResolver channelNameResolver = new ChannelNameResolver() { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java index cd0166334d..f9afb3c431 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java @@ -31,7 +31,7 @@ import org.junit.Test; import org.springframework.integration.ConfigurationException; import org.springframework.integration.channel.ChannelRegistry; import org.springframework.integration.channel.DefaultChannelRegistry; -import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.handler.AbstractMessageHandlerAdapter; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; @@ -41,7 +41,7 @@ import org.springframework.integration.message.StringMessage; */ public class SplitterMessageHandlerAdapterTests { - private SimpleChannel testChannel = new SimpleChannel(); + private QueueChannel testChannel = new QueueChannel(); private ChannelRegistry channelRegistry = new DefaultChannelRegistry(); diff --git a/spring-integration-reference/src/core-api.xml b/spring-integration-reference/src/core-api.xml index 9db8b16ee4..299bfb1cee 100644 --- a/spring-integration-reference/src/core-api.xml +++ b/spring-integration-reference/src/core-api.xml @@ -121,16 +121,16 @@ new GenericMessage<T>(T payload, MessageHeader headerToCopy)true if the message is sent successfully. If the send call times out or is interrupted, then it will return false. Likewise when receiving a message, the return value will be null in the case of a timeout or interrupt. - The SimpleChannel implementation wraps a queue. It provides a no-argument constructor as + The QueueChannel implementation wraps a queue. It provides a no-argument constructor as well as a constructor that accepts the queue capacity: - public SimpleChannel(int capacity) + public QueueChannel(int capacity) Specifying a capacity of 0 will create a "direct-handoff" channel where a sender will block until the channel's receive() method is called. Otherwise a channel that has not reached its capacity limit will store messages in its internal queue, and the send() method will return immediately even if no receiver is ready to handle the message. - Whereas the SimpleChannel enforces first-in/first-out (FIFO) ordering, the + Whereas the QueueChannel enforces first-in/first-out (FIFO) ordering, the PriorityChannel is an alternative implementation that allows for messages to be ordered within the channel based upon a priority. By default the priority is determined by the 'priority' property within each message's header. However, for custom priority determination