Added ThreadLocalChannelFactory (INT-221).

spring-integration-core/src/main/java/org/springframework/integration/channel/factory/PriorityChannelFactory.java

@@ -22,6 +22,7 @@ import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.DispatcherPolicy;
 import org.springframework.integration.channel.PriorityChannel;
 import org.springframework.integration.message.Message;
+import org.springframework.util.Assert;
 
 /**
  * A {@link ChannelFactory} for creating {@link PriorityChannel} instances.
@@ -30,11 +31,13 @@ import org.springframework.integration.message.Message;
  */
 public class PriorityChannelFactory extends AbstractChannelFactory {
 
-	private int capacity;
+	private int capacity = PriorityChannel.DEFAULT_CAPACITY;
 
 	private Comparator<Message<?>> comparator;
 
+
 	public void setCapacity(int capacity) {
+		Assert.isTrue(capacity > 0, "capacity must be a positive value");
 		this.capacity = capacity;
 	}
 
@@ -44,7 +47,7 @@ public class PriorityChannelFactory extends AbstractChannelFactory {
 
 	@Override
 	protected AbstractMessageChannel createChannelInternal(DispatcherPolicy dispatcherPolicy) {
-		return new PriorityChannel(capacity, dispatcherPolicy, comparator);
+		return new PriorityChannel(this.capacity, dispatcherPolicy, this.comparator);
 	}
 
 }

spring-integration-core/src/main/java/org/springframework/integration/channel/factory/ThreadLocalChannelFactory.java

@@ -0,0 +1,35 @@
+/*
+ * 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.factory;
+
+import org.springframework.integration.channel.AbstractMessageChannel;
+import org.springframework.integration.channel.DispatcherPolicy;
+import org.springframework.integration.channel.ThreadLocalChannel;
+
+/**
+ * A {@link ChannelFactory} implementation for creating {@link ThreadLocalChannel} instances.
+ * 
+ * @author Mark Fisher
+ */
+public class ThreadLocalChannelFactory extends AbstractChannelFactory {
+
+	@Override
+	protected AbstractMessageChannel createChannelInternal(DispatcherPolicy dispatcherPolicy) {
+		return new ThreadLocalChannel();
+	}
+
+}

spring-integration-core/src/test/java/org/springframework/integration/channel/factory/ChannelFactoryTests.java

@@ -35,30 +35,32 @@ 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.PriorityChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.channel.RendezvousChannel;
+import org.springframework.integration.channel.ThreadLocalChannel;
 import org.springframework.integration.message.Message;
 import org.springframework.integration.message.selector.MessageSelector;
 
 /**
  * @author Marius Bogoevici
+ * @author Mark Fisher
  */
 public class ChannelFactoryTests {
 
-	ArrayList<ChannelInterceptor> interceptors = null;
+	private final ArrayList<ChannelInterceptor> interceptors = new ArrayList<ChannelInterceptor>();
+
+	private final DispatcherPolicy dispatcherPolicy = new DispatcherPolicy();
 
-	DispatcherPolicy dispatcherPolicy = null;
 
 	@Before
-	public void createInterceptorsList() {
-		interceptors = new ArrayList<ChannelInterceptor>();
+	public void initInterceptorsList() {
 		interceptors.add(new TestChannelInterceptor());
 		interceptors.add(new TestChannelInterceptor());
 	}
 
 	@Before
-	public void createDispatcherPolicy() {
-		dispatcherPolicy = new DispatcherPolicy();
+	public void initDispatcherPolicy() {
 		dispatcherPolicy.setMaxMessagesPerTask(100);
 	}
 
@@ -89,8 +91,18 @@ public class ChannelFactoryTests {
 
 	@Test
 	public void testPriorityChannelFactory() {
-		RendezvousChannelFactory channelFactory = new RendezvousChannelFactory();
-		genericChannelFactoryTests(channelFactory, RendezvousChannel.class);
+		PriorityChannelFactory channelFactory = new PriorityChannelFactory();
+		genericChannelFactoryTests(channelFactory, PriorityChannel.class);
+	}
+
+	@Test
+	public void testThreadLocalChannelFactory() {
+		ThreadLocalChannelFactory channelFactory = new ThreadLocalChannelFactory();
+		assertNotNull(interceptors);
+		AbstractMessageChannel channel = (AbstractMessageChannel)
+				channelFactory.getChannel(dispatcherPolicy, interceptors);
+		assertEquals(ThreadLocalChannel.class, channel.getClass());
+		assertInterceptors(channel);
 	}
 
 	@Test