From 1c8f44a41c3b8695d71fdc37018bb8514b4575c9 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 19 May 2008 21:10:11 +0000 Subject: [PATCH] Factored out a new ThreadLocalChannel implementation from SynchronousChannel, and then renamed SynchronousChannel to DirectChannel. --- .../integration/bus/SubscriptionManager.java | 8 +- .../integration/channel/DirectChannel.java | 112 ++++++++++++++++++ .../ThreadLocalChannel.java} | 74 ++---------- .../channel/config/DirectChannelParser.java | 4 +- ...Factory.java => DirectChannelFactory.java} | 11 +- ...essageEndpointAnnotationPostProcessor.java | 4 +- ...va => DirectChannelSubscriptionTests.java} | 9 +- .../channel/ThreadLocalChannelTests.java | 63 ++++++++++ .../config/DirectChannelParserTests.java | 16 +-- ...lFactory.java => ChannelFactoryTests.java} | 44 ++++--- .../config/MessageBusParserTests.java | 8 +- .../config/messageBusWithChannelFactory.xml | 4 +- ...nnelTests.java => DirectChannelTests.java} | 43 ++----- .../AggregatingMessageHandlerTests.java | 1 - 14 files changed, 241 insertions(+), 160 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java rename spring-integration-core/src/main/java/org/springframework/integration/{dispatcher/SynchronousChannel.java => channel/ThreadLocalChannel.java} (55%) rename spring-integration-core/src/main/java/org/springframework/integration/channel/factory/{SynchronousChannelFactory.java => DirectChannelFactory.java} (74%) rename spring-integration-core/src/test/java/org/springframework/integration/bus/{SynchronousChannelSubscriptionTests.java => DirectChannelSubscriptionTests.java} (93%) create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/channel/ThreadLocalChannelTests.java rename spring-integration-core/src/test/java/org/springframework/integration/channel/factory/{TestChannelFactory.java => ChannelFactoryTests.java} (86%) rename spring-integration-core/src/test/java/org/springframework/integration/dispatcher/{SynchronousChannelTests.java => DirectChannelTests.java} (76%) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/SubscriptionManager.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/SubscriptionManager.java index 689dfc72f7..f4adb66db8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/SubscriptionManager.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/SubscriptionManager.java @@ -27,10 +27,10 @@ import org.apache.commons.logging.LogFactory; import org.springframework.context.Lifecycle; import org.springframework.integration.ConfigurationException; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.dispatcher.DefaultPollingDispatcher; import org.springframework.integration.dispatcher.PollingDispatcherTask; -import org.springframework.integration.dispatcher.SynchronousChannel; import org.springframework.integration.endpoint.TargetEndpoint; import org.springframework.integration.message.MessagingException; import org.springframework.integration.message.Target; @@ -87,7 +87,7 @@ public class SubscriptionManager { if (schedule == null) { schedule = this.defaultSchedule; } - else if (this.channel instanceof SynchronousChannel) { + else if (this.channel instanceof DirectChannel) { if (logger.isInfoEnabled()) { logger.info("Subscribing to a SynchronousChannel. The provided schedule will be ignored."); } @@ -106,8 +106,8 @@ public class SubscriptionManager { ((Lifecycle) target).start(); } } - if (this.channel instanceof SynchronousChannel) { - ((SynchronousChannel) this.channel).subscribe(target); + if (this.channel instanceof DirectChannel) { + ((DirectChannel) this.channel).subscribe(target); if (target instanceof TargetEndpoint) { ((TargetEndpoint) target).setErrorHandler(new ErrorHandler() { public void handle(Throwable t) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java new file mode 100644 index 0000000000..abed877ed0 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/DirectChannel.java @@ -0,0 +1,112 @@ +/* + * 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.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +import org.springframework.integration.dispatcher.SimpleDispatcher; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.Source; +import org.springframework.integration.message.Subscribable; +import org.springframework.integration.message.Target; +import org.springframework.integration.message.selector.MessageSelector; + +/** + * A channel that invokes the subscribed {@link MessageHandler handler(s)} in a + * sender's thread (returning after at most one handles the message). If a + * {@link Source} is provided, then that source will likewise be polled + * within a receiver's thread. + * + * @author Dave Syer + * @author Mark Fisher + */ +public class DirectChannel extends AbstractMessageChannel implements Subscribable { + + private volatile Source source; + + private final SimpleDispatcher dispatcher; + + private final AtomicInteger handlerCount = new AtomicInteger(); + + + public DirectChannel() { + this(null); + } + + public DirectChannel(Source source) { + super(defaultDispatcherPolicy()); + this.source = source; + this.dispatcher = new SimpleDispatcher(this.getDispatcherPolicy()); + } + + + public boolean subscribe(Target target) { + boolean added = this.dispatcher.subscribe(target); + if (added) { + this.handlerCount.incrementAndGet(); + } + return added; + } + + public boolean unsubscribe(Target target) { + boolean removed = this.dispatcher.unsubscribe(target); + if (removed) { + this.handlerCount.decrementAndGet(); + } + return removed; + } + + + @Override + protected Message doReceive(long timeout) { + if (this.source != null) { + return this.source.receive(); + } + return null; + } + + @Override + protected boolean doSend(Message message, long timeout) { + if (message != null && this.handlerCount.get() > 0) { + return this.dispatcher.dispatch(message); + } + return false; + } + + public List> clear() { + return new ArrayList>(); + } + + public List> purge(MessageSelector selector) { + return new ArrayList>(); + } + + + private static DispatcherPolicy defaultDispatcherPolicy() { + DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(false); + dispatcherPolicy.setMaxMessagesPerTask(1); + dispatcherPolicy.setReceiveTimeout(0); + dispatcherPolicy.setRejectionLimit(1); + dispatcherPolicy.setRetryInterval(0); + dispatcherPolicy.setShouldFailOnRejectionLimit(false); + return dispatcherPolicy; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ThreadLocalChannel.java similarity index 55% rename from spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java rename to spring-integration-core/src/main/java/org/springframework/integration/channel/ThreadLocalChannel.java index 5c1d7e2adc..eb49d1b6db 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ThreadLocalChannel.java @@ -14,89 +14,37 @@ * limitations under the License. */ -package org.springframework.integration.dispatcher; +package org.springframework.integration.channel; import java.util.ArrayList; import java.util.List; import java.util.Queue; import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.atomic.AtomicInteger; -import org.springframework.integration.channel.AbstractMessageChannel; -import org.springframework.integration.channel.DispatcherPolicy; -import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; -import org.springframework.integration.message.Source; -import org.springframework.integration.message.Subscribable; -import org.springframework.integration.message.Target; import org.springframework.integration.message.selector.MessageSelector; /** - * A channel that invokes the subscribed {@link MessageHandler handler(s)} in a - * sender's thread (returning after at most one handles the message). If a - * {@link PollableSource} is provided, then that source will likewise be polled - * within a receiver's thread. - *

- * If the channel has no subscribed handlers and no configured source, then it - * will store messages in a thread-bound queue. In other words, send() will put - * a message at the tail of the queue for the current thread, and receive() will - * retrieve a message from the head of the queue. + * A channel implementation that stores messages in a thread-bound queue. In + * other words, send() will put a message at the tail of the queue for the + * current thread, and receive() will retrieve a message from the head of the + * queue. * * @author Dave Syer * @author Mark Fisher */ -public class SynchronousChannel extends AbstractMessageChannel implements Subscribable { +public class ThreadLocalChannel extends AbstractMessageChannel { private static final ThreadLocalMessageHolder messageHolder = new ThreadLocalMessageHolder(); - private volatile Source source; - - private final SimpleDispatcher dispatcher; - - private final AtomicInteger handlerCount = new AtomicInteger(); - - - public SynchronousChannel() { - this(null); - } - - public SynchronousChannel(Source source) { + public ThreadLocalChannel() { super(defaultDispatcherPolicy()); - this.source = source; - this.dispatcher = new SimpleDispatcher(this.getDispatcherPolicy()); - } - - - public void setSource(Source source) { - this.source = source; - } - - public boolean subscribe(Target target) { - boolean added = this.dispatcher.subscribe(target); - if (added) { - this.handlerCount.incrementAndGet(); - } - return added; - } - - public boolean unsubscribe(Target target) { - boolean removed = this.dispatcher.unsubscribe(target); - if (removed) { - this.handlerCount.decrementAndGet(); - } - return removed; } @Override protected Message doReceive(long timeout) { - if (this.source != null) { - Message result = this.source.receive(); - if (result != null) { - return result; - } - } return messageHolder.get().poll(); } @@ -105,13 +53,7 @@ public class SynchronousChannel extends AbstractMessageChannel implements Subscr if (message == null) { return false; } - if (this.handlerCount.get() > 0) { - return this.dispatcher.dispatch(message); - } - else if (this.source == null) { - return messageHolder.get().add(message); - } - return false; + return messageHolder.get().add(message); } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/config/DirectChannelParser.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/config/DirectChannelParser.java index e4e134741c..7c11983077 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/config/DirectChannelParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/config/DirectChannelParser.java @@ -19,8 +19,8 @@ package org.springframework.integration.channel.config; import org.w3c.dom.Element; import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.DispatcherPolicy; -import org.springframework.integration.dispatcher.SynchronousChannel; import org.springframework.util.StringUtils; /** @@ -32,7 +32,7 @@ public class DirectChannelParser extends AbstractChannelParser { @Override protected Class getBeanClass(Element element) { - return SynchronousChannel.class; + return DirectChannel.class; } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/SynchronousChannelFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/DirectChannelFactory.java similarity index 74% rename from spring-integration-core/src/main/java/org/springframework/integration/channel/factory/SynchronousChannelFactory.java rename to spring-integration-core/src/main/java/org/springframework/integration/channel/factory/DirectChannelFactory.java index 418dbefe1e..5cce001245 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/SynchronousChannelFactory.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/DirectChannelFactory.java @@ -17,20 +17,19 @@ package org.springframework.integration.channel.factory; import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.DispatcherPolicy; -import org.springframework.integration.channel.PriorityChannel; -import org.springframework.integration.dispatcher.SynchronousChannel; /** - * A {@link ChannelFactory} for creating {@link SynchronousChannel} instances. - * @author Marius Bogoevici + * A {@link ChannelFactory} for creating {@link DirectChannel} instances. * + * @author Marius Bogoevici */ -public class SynchronousChannelFactory extends AbstractChannelFactory { +public class DirectChannelFactory extends AbstractChannelFactory { @Override protected AbstractMessageChannel createChannelInternal(DispatcherPolicy dispatcherPolicy) { - return new SynchronousChannel(null); + return new DirectChannel(null); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java index c70a648659..fc1f26114e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/MessageEndpointAnnotationPostProcessor.java @@ -45,7 +45,7 @@ import org.springframework.integration.annotation.Router; import org.springframework.integration.annotation.Splitter; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.ChannelRegistryAware; -import org.springframework.integration.dispatcher.SynchronousChannel; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.endpoint.ConcurrencyPolicy; import org.springframework.integration.endpoint.HandlerEndpoint; import org.springframework.integration.endpoint.SourceEndpoint; @@ -159,7 +159,7 @@ public class MessageEndpointAnnotationPostProcessor implements BeanPostProcessor MethodInvokingSource source = new MethodInvokingSource(); source.setObject(bean); source.setMethod(method.getName()); - SynchronousChannel channel = new SynchronousChannel(); + DirectChannel channel = new DirectChannel(); PollingSchedule schedule = new PollingSchedule(period); schedule.setInitialDelay(initialDelay); schedule.setFixedRate(fixedRate); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/SynchronousChannelSubscriptionTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java similarity index 93% rename from spring-integration-core/src/test/java/org/springframework/integration/bus/SynchronousChannelSubscriptionTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java index 35ebad60a8..bd1a1c0a7b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/SynchronousChannelSubscriptionTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/DirectChannelSubscriptionTests.java @@ -23,10 +23,11 @@ import org.junit.Test; import org.springframework.integration.annotation.Handler; import org.springframework.integration.annotation.MessageEndpoint; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.channel.ThreadLocalChannel; import org.springframework.integration.config.MessageEndpointAnnotationPostProcessor; -import org.springframework.integration.dispatcher.SynchronousChannel; import org.springframework.integration.endpoint.HandlerEndpoint; import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; @@ -37,13 +38,13 @@ import org.springframework.integration.scheduling.Subscription; /** * @author Mark Fisher */ -public class SynchronousChannelSubscriptionTests { +public class DirectChannelSubscriptionTests { private MessageBus bus = new MessageBus(); - private MessageChannel sourceChannel = new SynchronousChannel(); + private MessageChannel sourceChannel = new DirectChannel(); - private MessageChannel targetChannel = new SynchronousChannel(); + private MessageChannel targetChannel = new ThreadLocalChannel(); @Before diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/ThreadLocalChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/ThreadLocalChannelTests.java new file mode 100644 index 0000000000..2a48962365 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/ThreadLocalChannelTests.java @@ -0,0 +1,63 @@ +/* + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.util.List; + +import org.junit.Test; + +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Mark Fisher + */ +public class ThreadLocalChannelTests { + + @Test + public void testSendAndReceive() { + ThreadLocalChannel channel = new ThreadLocalChannel(); + StringMessage message = new StringMessage("test"); + assertNull(channel.receive()); + assertTrue(channel.send(message)); + Message response = channel.receive(); + assertNotNull(response); + assertEquals(response, message); + assertNull(channel.receive()); + } + + @Test + public void testSendAndClear() { + ThreadLocalChannel channel = new ThreadLocalChannel(); + StringMessage message1 = new StringMessage("test1"); + StringMessage message2 = new StringMessage("test2"); + assertNull(channel.receive()); + assertTrue(channel.send(message1)); + assertTrue(channel.send(message2)); + List> clearedMessages = channel.clear(); + assertEquals(2, clearedMessages.size()); + assertEquals(message1, clearedMessages.get(0)); + assertEquals(message2, clearedMessages.get(1)); + assertNull(channel.receive()); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/DirectChannelParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/DirectChannelParserTests.java index 9f48440793..312a9c0678 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/config/DirectChannelParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/config/DirectChannelParserTests.java @@ -20,12 +20,11 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; import org.junit.Test; import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.integration.dispatcher.SynchronousChannel; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.StringMessage; @@ -35,23 +34,18 @@ import org.springframework.integration.message.StringMessage; public class DirectChannelParserTests { @Test - public void testChannelWithoutSource() { + public void testReceivesNullFromChannelWithoutSource() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "directChannelParserTests.xml", DirectChannelParserTests.class); - SynchronousChannel channel = (SynchronousChannel) context.getBean("channelWithoutSource"); + DirectChannel channel = (DirectChannel) context.getBean("channelWithoutSource"); assertNull(channel.receive()); - Message message = new StringMessage("test"); - assertTrue(channel.send(message)); - Message reply = channel.receive(); - assertNotNull(reply); - assertEquals(message, reply); } @Test - public void testChannelWithSource() { + public void testReceivesMessageFromChannelWithSource() { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( "directChannelParserTests.xml", DirectChannelParserTests.class); - SynchronousChannel channel = (SynchronousChannel) context.getBean("channelWithSource"); + DirectChannel channel = (DirectChannel) context.getBean("channelWithSource"); assertFalse(channel.send(new StringMessage("test"))); Message reply = channel.receive(); assertNotNull(reply); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/factory/TestChannelFactory.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/factory/ChannelFactoryTests.java similarity index 86% rename from spring-integration-core/src/test/java/org/springframework/integration/channel/factory/TestChannelFactory.java rename to spring-integration-core/src/test/java/org/springframework/integration/channel/factory/ChannelFactoryTests.java index 86472416a3..64c2c773ba 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/factory/TestChannelFactory.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/factory/ChannelFactoryTests.java @@ -25,29 +25,25 @@ import java.util.List; import org.junit.Before; import org.junit.Test; + import org.springframework.beans.DirectFieldAccessor; -import org.springframework.beans.factory.config.BeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.context.ApplicationContext; -import org.springframework.context.support.GenericApplicationContext; import org.springframework.context.support.StaticApplicationContext; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.channel.ChannelInterceptor; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.DispatcherPolicy; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.channel.RendezvousChannel; -import org.springframework.integration.config.MessageBusParser; -import org.springframework.integration.dispatcher.SynchronousChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.selector.MessageSelector; /** - * * @author Marius Bogoevici */ -public class TestChannelFactory { +public class ChannelFactoryTests { ArrayList interceptors = null; @@ -66,6 +62,7 @@ public class TestChannelFactory { dispatcherPolicy.setMaxMessagesPerTask(100); } + @Test public void testQueueChannelFactory() { QueueChannelFactory channelFactory = new QueueChannelFactory(); @@ -75,12 +72,12 @@ public class TestChannelFactory { } @Test - public void testSynchronousChannelFactory() { - SynchronousChannelFactory channelFactory = new SynchronousChannelFactory(); + public void testDirectChannelFactory() { + DirectChannelFactory channelFactory = new DirectChannelFactory(); assertNotNull(interceptors); - AbstractMessageChannel channel = (AbstractMessageChannel) channelFactory.getChannel(dispatcherPolicy, - interceptors); - assertEquals(SynchronousChannel.class, channel.getClass()); + AbstractMessageChannel channel = (AbstractMessageChannel) + channelFactory.getChannel(dispatcherPolicy, interceptors); + assertEquals(DirectChannel.class, channel.getClass()); assertInterceptors(channel); } @@ -113,32 +110,33 @@ public class TestChannelFactory { assertInterceptors(channel); } + private void genericChannelFactoryTests(ChannelFactory channelFactory, Class expectedChannelClass) { assertNotNull(dispatcherPolicy); assertNotNull(interceptors); - AbstractMessageChannel channel = (AbstractMessageChannel) channelFactory.getChannel(dispatcherPolicy, - interceptors); + AbstractMessageChannel channel = (AbstractMessageChannel) + channelFactory.getChannel(dispatcherPolicy, interceptors); assertEquals(expectedChannelClass, channel.getClass()); assertTrue(channel.getDispatcherPolicy() == dispatcherPolicy); assertInterceptors(channel); } + @SuppressWarnings("unchecked") private void assertInterceptors(AbstractMessageChannel channel) { Object interceptorsWrapper = new DirectFieldAccessor(channel).getPropertyValue("interceptors"); - List interceptors = (List) new DirectFieldAccessor(interceptorsWrapper) - .getPropertyValue("interceptors"); + List interceptors = (List) + new DirectFieldAccessor(interceptorsWrapper).getPropertyValue("interceptors"); assertTrue(interceptors.get(0) == interceptors.get(0)); assertTrue(interceptors.get(1) == interceptors.get(1)); } + static class TestChannelInterceptor implements ChannelInterceptor { public void postReceive(Message message, MessageChannel channel) { - } public void postSend(Message message, MessageChannel channel, boolean sent) { - } public boolean preReceive(MessageChannel channel) { @@ -150,13 +148,14 @@ public class TestChannelFactory { } } - + + static class StubChannel extends AbstractMessageChannel { public StubChannel(DispatcherPolicy dispatcherPolicy) { super(dispatcherPolicy); } - + @Override protected Message doReceive(long timeout) { return null; @@ -174,16 +173,15 @@ public class TestChannelFactory { public List> purge(MessageSelector selector) { return null; } - } - + + static class StubChannelFactory extends AbstractChannelFactory { @Override protected AbstractMessageChannel createChannelInternal(DispatcherPolicy dispatcherPolicy) { return new StubChannel(dispatcherPolicy); } - } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java index efb8ed1509..59568afdc6 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/MessageBusParserTests.java @@ -29,8 +29,8 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.ConfigurationException; import org.springframework.integration.bus.MessageBus; import org.springframework.integration.bus.TestMessageBusAwareImpl; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.integration.dispatcher.SynchronousChannel; import org.springframework.integration.endpoint.TargetEndpoint; import org.springframework.integration.handler.TestHandlers; import org.springframework.integration.scheduling.Subscription; @@ -140,8 +140,8 @@ public class MessageBusParserTests { @Test public void testMessageBusAwareAutomaticallyAddedByNamespace() { - ApplicationContext context = new ClassPathXmlApplicationContext("messageBusWithMessageBusAware.xml", - this.getClass()); + ApplicationContext context = new ClassPathXmlApplicationContext( + "messageBusWithMessageBusAware.xml", this.getClass()); TestMessageBusAwareImpl messageBusAware = (TestMessageBusAwareImpl) context.getBean("messageBusAwareBean"); assertTrue(messageBusAware.getMessageBus() == context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME)); } @@ -150,7 +150,7 @@ public class MessageBusParserTests { public void testMessageBusWithChannelFactory() { ApplicationContext context = new ClassPathXmlApplicationContext("messageBusWithChannelFactory.xml", this.getClass()); - assertEquals(SynchronousChannel.class, context.getBean("defaultTypeChannel").getClass()); + assertEquals(DirectChannel.class, context.getBean("defaultTypeChannel").getClass()); assertEquals(QueueChannel.class, context.getBean("specifiedTypeChannel").getClass()); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/messageBusWithChannelFactory.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/messageBusWithChannelFactory.xml index 141e1a03c1..523b5d3b1f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/messageBusWithChannelFactory.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/messageBusWithChannelFactory.xml @@ -7,9 +7,9 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-core-1.0.xsd"> - + - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DirectChannelTests.java similarity index 76% rename from spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DirectChannelTests.java index 528fb4c750..592f1c2b80 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/DirectChannelTests.java @@ -21,13 +21,13 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; -import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.SynchronousQueue; import java.util.concurrent.TimeUnit; import org.junit.Test; +import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.message.Message; import org.springframework.integration.message.Source; import org.springframework.integration.message.StringMessage; @@ -36,14 +36,14 @@ import org.springframework.integration.message.Target; /** * @author Mark Fisher */ -public class SynchronousChannelTests { +public class DirectChannelTests { private static final String HANDLER_THREAD = "handler-thread"; @Test public void testSend() { - SynchronousChannel channel = new SynchronousChannel(); + DirectChannel channel = new DirectChannel(); channel.subscribe(new ThreadNameSettingTestTarget()); StringMessage message = new StringMessage("test"); assertTrue(channel.send(message)); @@ -51,37 +51,10 @@ public class SynchronousChannelTests { assertEquals(Thread.currentThread().getName(), handlerThreadName); } - @Test - public void testSendAndReceiveWithNoHandler() { - SynchronousChannel channel = new SynchronousChannel(); - StringMessage message = new StringMessage("test"); - assertNull(channel.receive()); - assertTrue(channel.send(message)); - Message response = channel.receive(); - assertNotNull(response); - assertEquals(response, message); - assertNull(channel.receive()); - } - - @Test - public void testSendAndClearWithNoHandler() { - SynchronousChannel channel = new SynchronousChannel(); - StringMessage message1 = new StringMessage("test1"); - StringMessage message2 = new StringMessage("test2"); - assertNull(channel.receive()); - assertTrue(channel.send(message1)); - assertTrue(channel.send(message2)); - List> clearedMessages = channel.clear(); - assertEquals(2, clearedMessages.size()); - assertEquals(message1, clearedMessages.get(0)); - assertEquals(message2, clearedMessages.get(1)); - assertNull(channel.receive()); - } - @Test public void testSendInSeparateThread() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); - final SynchronousChannel channel = new SynchronousChannel(); + final DirectChannel channel = new DirectChannel(); channel.subscribe(new ThreadNameSettingTestTarget(latch)); final StringMessage message = new StringMessage("test"); new Thread(new Runnable() { @@ -96,7 +69,7 @@ public class SynchronousChannelTests { @Test public void testReceive() { - SynchronousChannel channel = new SynchronousChannel(new Source() { + DirectChannel channel = new DirectChannel(new Source() { public Message receive() { return new StringMessage("foo"); } @@ -110,7 +83,7 @@ public class SynchronousChannelTests { @Test public void testReceiveWithMessageResult() { - SynchronousChannel channel = new SynchronousChannel(new MessageReturningTestSource("foo")); + DirectChannel channel = new DirectChannel(new MessageReturningTestSource("foo")); Message message = channel.receive(); assertNotNull(message); assertNotNull(message.getPayload()); @@ -122,7 +95,7 @@ public class SynchronousChannelTests { @Test public void testReceiveInSeparateThread() throws InterruptedException { - final SynchronousChannel channel = new SynchronousChannel(new MessageReturningTestSource("foo")); + final DirectChannel channel = new DirectChannel(new MessageReturningTestSource("foo")); final SynchronousQueue> messageHolder = new SynchronousQueue>(); new Thread(new Runnable() { public void run() { @@ -147,7 +120,7 @@ public class SynchronousChannelTests { @Test public void testReceiveWithNoSource() { - SynchronousChannel channel = new SynchronousChannel(); + DirectChannel channel = new DirectChannel(); assertNull(channel.receive()); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java index f631943e0b..440f066a0e 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/router/AggregatingMessageHandlerTests.java @@ -21,7 +21,6 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import java.util.ArrayList; -import java.util.Collection; import java.util.Collections; import java.util.List; import java.util.concurrent.CountDownLatch;