Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

SimpleChannel is now QueueChannel. Also added RendezvousChannel for the zero-capacity SynchronousQueue-based version.

Browse Source
This commit is contained in:
Mark Fisher
2008-04-22 20:26:19 +00:00
parent 4b01a3a09c
commit bf0f8b16ec
50 changed files with 354 additions and 330 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapterTests.java
View File

@@ -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<Class<? extends ApplicationEvent>> eventTypes = new ArrayList<Class<? extends ApplicationEvent>>();
eventTypes.add(TestApplicationEvent1.class);

4
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/ApplicationEventTargetAdapterTests.java
View File

@@ -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();

2
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/event/applicationEventSourceAdapterTests.xml
View File

@@ -6,7 +6,7 @@
<bean id="bus" class="org.springframework.integration.bus.MessageBus"/>
<bean id="channel" class="org.springframework.integration.channel.SimpleChannel"/>
<bean id="channel" class="org.springframework.integration.channel.QueueChannel"/>
<bean id="adapter" class="org.springframework.integration.adapter.event.ApplicationEventSourceAdapter">
<constructor-arg ref="channel"/>

6
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/httpinvoker/HttpInvokerSourceAdapterTests.java
View File

@@ -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();

4
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/rmi/config/RmiTargetAdapterParserTests.java
View File

@@ -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

14
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/ByteStreamSourceAdapterTests.java
View File

@@ -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);

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

@@ -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<byte[]>(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<byte[]>(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<byte[]>(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<byte[]>(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<byte[]>(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<byte[]>(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<byte[]>(new byte[] {1,2,3}), 0);

10
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java
View File

@@ -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);

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

@@ -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);

4
spring-integration-core/src/main/java/org/springframework/integration/bus/DefaultErrorChannel.java
View File

@@ -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());

6
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -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) {

3
spring-integration-core/src/main/java/org/springframework/integration/channel/MessageChannel.java
View File

@@ -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.
*/

2
spring-integration-core/src/main/java/org/springframework/integration/channel/PriorityChannel.java
View File

@@ -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;

39
spring-integration-core/src/main/java/org/springframework/integration/channel/BaseBlockingQueueChannel.java → spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java
View File

@@ -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<Message<?>> queue;
@@ -38,12 +46,37 @@ public class BaseBlockingQueueChannel extends AbstractMessageChannel {
/**
* Create a channel with the specified queue and dispatcher policy.
*/
public BaseBlockingQueueChannel(BlockingQueue<Message<?>> queue, DispatcherPolicy dispatcherPolicy) {
public QueueChannel(BlockingQueue<Message<?>> 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<Message<?>>(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");

40
spring-integration-core/src/main/java/org/springframework/integration/channel/RendezvousChannel.java Normal file
View File

@@ -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<Message<?>>(), dispatcherPolicy);
}
public RendezvousChannel() {
this(null);
}
}

2
spring-integration-core/src/main/java/org/springframework/integration/channel/RequestReplyTemplate.java
View File

@@ -96,7 +96,7 @@ public class RequestReplyTemplate {
* @return the reply message or <code>null</code>
*/
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);

55
spring-integration-core/src/main/java/org/springframework/integration/channel/SimpleChannel.java
View File

@@ -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<Message<?>>(capacity) :
new SynchronousQueue<Message<?>>(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);
}
}

10
spring-integration-core/src/main/java/org/springframework/integration/config/ChannelParser.java
View File

@@ -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) {

4
spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingTargetTests.java
View File

@@ -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<String> message = new GenericMessage<String>("123", "testing");
channel.send(message);

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

@@ -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);

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

@@ -8,7 +8,7 @@
<property name="autoStartup" value="false"/>
</bean>
<bean id="channel" class="org.springframework.integration.channel.SimpleChannel"/>
<bean id="channel" class="org.springframework.integration.channel.QueueChannel"/>
<bean id="sourceAdapter" class="org.springframework.integration.adapter.PollingSourceAdapter">
<constructor-arg>

16
spring-integration-core/src/test/java/org/springframework/integration/aop/MessagePublishingInterceptorTests.java
View File

@@ -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));
}

8
spring-integration-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java
View File

@@ -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);

2
spring-integration-core/src/test/java/org/springframework/integration/aop/publisherAnnotationPostProcessorTests.xml
View File

@@ -6,7 +6,7 @@
<bean id="messageBus" class="org.springframework.integration.bus.MessageBus"/>
<bean id="testChannel" class="org.springframework.integration.channel.SimpleChannel"/>
<bean id="testChannel" class="org.springframework.integration.channel.QueueChannel"/>
<bean id="testBean" class="org.springframework.integration.aop.PublisherAnnotationTestBean"/>

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

@@ -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);

34
spring-integration-core/src/test/java/org/springframework/integration/bus/SubscriptionManagerTests.java
View File

@@ -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);

6
spring-integration-core/src/test/java/org/springframework/integration/bus/SynchronousChannelSubscriptionTests.java
View File

@@ -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();

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

@@ -6,9 +6,9 @@
<bean id="bus" class="org.springframework.integration.bus.MessageBus"/>
<bean id="sourceChannel" class="org.springframework.integration.channel.SimpleChannel"/>
<bean id="sourceChannel" class="org.springframework.integration.channel.QueueChannel"/>
<bean id="targetChannel" class="org.springframework.integration.channel.SimpleChannel"/>
<bean id="targetChannel" class="org.springframework.integration.channel.QueueChannel"/>
<bean id="endpoint" class="org.springframework.integration.endpoint.HandlerEndpoint">
<constructor-arg ref="handler"/>

20
spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelPurgerTests.java
View File

@@ -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"));

6
spring-integration-core/src/test/java/org/springframework/integration/channel/DefaultChannelRegistryTests.java
View File

@@ -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");

22
spring-integration-core/src/test/java/org/springframework/integration/channel/SimpleChannelTests.java → spring-integration-core/src/test/java/org/springframework/integration/channel/QueueChannelTests.java
View File

@@ -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<String>(1, "test-1"));
assertTrue(result1);
boolean result2 = channel.send(new GenericMessage<String>(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<String>(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<String>(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<Message<?>> 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);

2
spring-integration-core/src/test/java/org/springframework/integration/channel/RequestReplyTemplateTests.java
View File

@@ -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() {

6
spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/ChannelInterceptorTests.java
View File

@@ -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) {

10
spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/MessageSelectingInterceptorTests.java
View File

@@ -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"));

34
spring-integration-core/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java
View File

@@ -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<MessageSelector> selectors = new ArrayList<MessageSelector>();
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<MessageSelector> selectors = new ArrayList<MessageSelector>();
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";

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

@@ -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);

6
spring-integration-core/src/test/java/org/springframework/integration/config/SubscriberAnnotationPostProcessorTests.java
View File

@@ -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);

8
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SourcePollerTests.java
View File

@@ -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<Message<?>> 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<Message<?>> 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<Message<?>> results = poller.poll();
assertTrue(results.isEmpty());

34
spring-integration-core/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java
View File

@@ -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");

34
spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/MessageEndpointAnnotationPostProcessorTests.java
View File

@@ -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);

4
spring-integration-core/src/test/java/org/springframework/integration/endpoint/annotation/simpleAnnotatedEndpointWithAutoCreateChannelTests.xml
View File

@@ -10,10 +10,8 @@
<bean id="endpoint" class="org.springframework.integration.endpoint.annotation.SimpleAnnotatedEndpoint"/>
<bean id="outputChannel" class="org.springframework.integration.channel.SimpleChannel"/>
<bean class="org.springframework.integration.config.MessageEndpointAnnotationPostProcessor">
<property name="messageBus" ref="bus"/>
<constructor-arg ref="bus"/>
</bean>
</beans>

4
spring-integration-core/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java
View File

@@ -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<String, String> attributes = new HashMap<String, String>();

28
spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java
View File

@@ -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);
}

10
spring-integration-core/src/test/java/org/springframework/integration/router/MultiChannelRouterTests.java
View File

@@ -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<MessageChannel> resolve(Message<?> message) {
List<MessageChannel> channels = new ArrayList<MessageChannel>();
@@ -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);

10
spring-integration-core/src/test/java/org/springframework/integration/router/PayloadTypeRouterTests.java
View File

@@ -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<Class<?>, MessageChannel> channelMappings = new ConcurrentHashMap<Class<?>, 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<Class<?>, MessageChannel> channelMappings = new ConcurrentHashMap<Class<?>, MessageChannel>();
channelMappings.put(String.class, stringChannel);
PayloadTypeRouter router = new PayloadTypeRouter();

18
spring-integration-core/src/test/java/org/springframework/integration/router/RecipientListRouterTests.java
View File

@@ -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<MessageChannel> channels = new ArrayList<MessageChannel>();
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<MessageChannel> channels = new ArrayList<MessageChannel>();
channels.add(channel1);
channels.add(channel2);

52
spring-integration-core/src/test/java/org/springframework/integration/router/RouterMessageHandlerAdapterTests.java
View File

@@ -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<String, Object> attribs = new ConcurrentHashMap<String, Object>();
RouterMessageHandlerAdapter adapter = new RouterMessageHandlerAdapter(testBean, routingMethod, attribs);
Message<String> message = new GenericMessage<String>("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<String> message = new GenericMessage<String>("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<String> message = new GenericMessage<String>("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<String> fooMessage = new StringMessage("foo");
Message<String> barMessage = new StringMessage("bar");
Message<String> 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();

8
spring-integration-core/src/test/java/org/springframework/integration/router/SingleChannelRouterTests.java
View File

@@ -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() {

4
spring-integration-core/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java
View File

@@ -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();

6
spring-integration-reference/src/core-api.xml
View File

@@ -121,16 +121,16 @@ new GenericMessage&lt;T&gt;(T payload, MessageHeader headerToCopy)</programlisti
When sending a message, the return value will be <emphasis>true</emphasis> if the message is sent successfully.
If the send call times out or is interrupted, then it will return <emphasis>false</emphasis>. Likewise when
receiving a message, the return value will be <emphasis>null</emphasis> in the case of a timeout or interrupt.
The <classname>SimpleChannel</classname> implementation wraps a queue. It provides a no-argument constructor as
The <classname>QueueChannel</classname> implementation wraps a queue. It provides a no-argument constructor as
well as a constructor that accepts the queue capacity:
<programlisting>public SimpleChannel(int capacity)</programlisting>
<programlisting>public QueueChannel(int capacity)</programlisting>
Specifying a capacity of 0 will create a "direct-handoff" channel where a sender will block until the channel's
<methodname>receive()</methodname> method is called. Otherwise a channel that has not reached its capacity limit
will store messages in its internal queue, and the <methodname>send()</methodname> method will return immediately
even if no receiver is ready to handle the message.
</para>
<para>
Whereas the <classname>SimpleChannel</classname> enforces first-in/first-out (FIFO) ordering, the
Whereas the <classname>QueueChannel</classname> enforces first-in/first-out (FIFO) ordering, the
<classname>PriorityChannel</classname> 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
'<literal>priority</literal>' property within each message's header. However, for custom priority determination