From 2c731b239f05df522cce763c23e1cafb2975e241 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 20 May 2008 16:41:42 +0000 Subject: [PATCH] Added ThreadLocalChannelFactory (INT-221). --- .../factory/PriorityChannelFactory.java | 7 ++-- .../factory/ThreadLocalChannelFactory.java | 35 +++++++++++++++++++ .../channel/factory/ChannelFactoryTests.java | 28 ++++++++++----- 3 files changed, 60 insertions(+), 10 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/factory/ThreadLocalChannelFactory.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/PriorityChannelFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/PriorityChannelFactory.java index 074ef4e66a..85a6a93924 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/PriorityChannelFactory.java +++ b/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> 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); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/ThreadLocalChannelFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/ThreadLocalChannelFactory.java new file mode 100644 index 0000000000..ce55bfed24 --- /dev/null +++ b/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(); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/factory/ChannelFactoryTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/factory/ChannelFactoryTests.java index 64c2c773ba..b6f1ef110b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/factory/ChannelFactoryTests.java +++ b/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 interceptors = null; + private final ArrayList interceptors = new ArrayList(); + + private final DispatcherPolicy dispatcherPolicy = new DispatcherPolicy(); - DispatcherPolicy dispatcherPolicy = null; @Before - public void createInterceptorsList() { - interceptors = new ArrayList(); + 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