Factored out a new ThreadLocalChannel implementation from SynchronousChannel, and then renamed SynchronousChannel to DirectChannel.

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

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<Message<?>> clear() {
+		return new ArrayList<Message<?>>();
+	}
+
+	public List<Message<?>> purge(MessageSelector selector) {
+		return new ArrayList<Message<?>>();
+	}
+
+
+	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;
+	}
+
+}

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.
- * <p>
- * 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);
 	}
 
 	/**

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

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

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

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

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<Message<?>> clearedMessages = channel.clear();
+		assertEquals(2, clearedMessages.size());
+		assertEquals(message1, clearedMessages.get(0));
+		assertEquals(message2, clearedMessages.get(1));
+		assertNull(channel.receive());
+	}
+
+}

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

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<ChannelInterceptor> 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<ChannelInterceptor> interceptors = (List<ChannelInterceptor>) new DirectFieldAccessor(interceptorsWrapper)
-				.getPropertyValue("interceptors");
+		List<ChannelInterceptor> interceptors = (List<ChannelInterceptor>)
+				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<Message<?>> purge(MessageSelector selector) {
 			return null;
 		}
-		
 	}
-	
+
+
 	static class StubChannelFactory extends AbstractChannelFactory {
 
 		@Override
 		protected AbstractMessageChannel createChannelInternal(DispatcherPolicy dispatcherPolicy) {
 			return new StubChannel(dispatcherPolicy);
 		}
-		
 	}
 
 }

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

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">
 
-	<message-bus channel-factory="synchronousChannelFactory"/>
+	<message-bus channel-factory="directChannelFactory"/>
 
-	<beans:bean id="synchronousChannelFactory" class="org.springframework.integration.channel.factory.SynchronousChannelFactory"/>	
+	<beans:bean id="directChannelFactory" class="org.springframework.integration.channel.factory.DirectChannelFactory"/>	
 
     <channel id="defaultTypeChannel"/>
 

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<Message<?>> 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<String>() {
+		DirectChannel channel = new DirectChannel(new Source<String>() {
 			public Message<String> 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<Message<?>> messageHolder = new SynchronousQueue<Message<?>>();
 		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());
 	}
 

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;