From 0d8e8fc6ee29693aedfbf7a43b9183b744afdb78 Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Fri, 2 May 2008 21:10:18 +0000 Subject: [PATCH] Adding ChannelFactory, various implementations for different types of channels and DefaultChannelFactoryBean. --- .../factory/AbstractChannelFactory.java | 52 +++++ .../channel/factory/ChannelFactory.java | 37 ++++ .../factory/DefaultChannelFactoryBean.java | 73 +++++++ .../factory/PriorityChannelFactory.java | 50 +++++ .../channel/factory/QueueChannelFactory.java | 53 +++++ .../factory/RendezvousChannelFactory.java | 36 ++++ .../factory/SynchronousChannelFactory.java | 36 ++++ .../channel/factory/TestChannelFactory.java | 182 ++++++++++++++++++ 8 files changed, 519 insertions(+) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/factory/AbstractChannelFactory.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/factory/ChannelFactory.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/factory/DefaultChannelFactoryBean.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/factory/PriorityChannelFactory.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/factory/QueueChannelFactory.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/factory/RendezvousChannelFactory.java create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/factory/SynchronousChannelFactory.java create mode 100644 spring-integration-core/src/test/java/org/springframework/integration/channel/factory/TestChannelFactory.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/AbstractChannelFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/AbstractChannelFactory.java new file mode 100644 index 0000000000..ea5b6928a2 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/AbstractChannelFactory.java @@ -0,0 +1,52 @@ +/* + * 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 java.util.List; + +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.channel.ChannelInterceptor; +import org.springframework.integration.channel.DispatcherPolicy; +import org.springframework.integration.channel.MessageChannel; + +/** + * Base class for {@link ChannelFactory} implementations. Subclasses should + * override {@literal createChannelInternal()}. + * + * @author Marius Bogoevici + */ +public abstract class AbstractChannelFactory implements ChannelFactory { + + public AbstractChannelFactory() { + super(); + } + + public final MessageChannel getChannel(DispatcherPolicy dispatcherPolicy, List interceptors) { + AbstractMessageChannel channel = createChannelInternal(dispatcherPolicy); + if (null != interceptors) { + channel.setInterceptors(interceptors); + } + return channel; + } + + /** + * Factory method to be overridden by subclasses. It assumes that subclasses will return + * subclasses of AbstractMessageChannel. + */ + protected abstract AbstractMessageChannel createChannelInternal(DispatcherPolicy dispatcherPolicy); + +} \ No newline at end of file diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/ChannelFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/ChannelFactory.java new file mode 100644 index 0000000000..94323f99e0 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/ChannelFactory.java @@ -0,0 +1,37 @@ +/* + * 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 java.util.List; + +import org.springframework.integration.channel.ChannelInterceptor; +import org.springframework.integration.channel.DispatcherPolicy; +import org.springframework.integration.channel.MessageChannel; + +/** + * Interface for a channel factory. + * @author Marius Bogoevici + */ +public interface ChannelFactory { + + /** + * Creates a channel, based on the provided dispatcher policy, and with the given interceptors. + * @return + */ + MessageChannel getChannel(DispatcherPolicy dispatcherPolicy, List interceptors); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/DefaultChannelFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/DefaultChannelFactoryBean.java new file mode 100644 index 0000000000..77fdbe7755 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/DefaultChannelFactoryBean.java @@ -0,0 +1,73 @@ +/* + * 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 java.util.List; + +import org.springframework.beans.factory.FactoryBean; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.bus.MessageBusAware; +import org.springframework.integration.channel.ChannelInterceptor; +import org.springframework.integration.channel.DispatcherPolicy; +import org.springframework.integration.channel.MessageChannel; + +/** + * Creates a channel by delegating to the current message bus-configured + * ChannelFactory. + * @author Marius Bogoevici + */ +public class DefaultChannelFactoryBean implements FactoryBean, MessageBusAware, InitializingBean { + + private volatile ChannelFactory channelFactory; + + private volatile List interceptors; + + private volatile DispatcherPolicy dispatcherPolicy; + + public void setMessageBus(MessageBus messageBus) { + this.channelFactory = messageBus.getChannelFactory(); + } + + public void afterPropertiesSet() throws Exception { + if (null == this.channelFactory) { + this.channelFactory = new QueueChannelFactory(); + } + + } + + public void setInterceptors(List interceptors) { + this.interceptors = interceptors; + } + + public void setDispatcherPolicy(DispatcherPolicy dispatcherPolicy) { + this.dispatcherPolicy = dispatcherPolicy; + } + + public Object getObject() throws Exception { + return channelFactory.getChannel(dispatcherPolicy, interceptors); + } + + public Class getObjectType() { + return MessageChannel.class; + } + + public boolean isSingleton() { + return true; + } + +} 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 new file mode 100644 index 0000000000..074ef4e66a --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/PriorityChannelFactory.java @@ -0,0 +1,50 @@ +/* + * 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 java.util.Comparator; + +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.channel.DispatcherPolicy; +import org.springframework.integration.channel.PriorityChannel; +import org.springframework.integration.message.Message; + +/** + * A {@link ChannelFactory} for creating {@link PriorityChannel} instances. + * @author Marius Bogoevici + * + */ +public class PriorityChannelFactory extends AbstractChannelFactory { + + private int capacity; + + private Comparator> comparator; + + public void setCapacity(int capacity) { + this.capacity = capacity; + } + + public void setComparator(Comparator> comparator) { + this.comparator = comparator; + } + + @Override + protected AbstractMessageChannel createChannelInternal(DispatcherPolicy dispatcherPolicy) { + return new PriorityChannel(capacity, dispatcherPolicy, comparator); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/QueueChannelFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/QueueChannelFactory.java new file mode 100644 index 0000000000..87228abff8 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/QueueChannelFactory.java @@ -0,0 +1,53 @@ +/* + * 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.QueueChannel; +import org.springframework.util.Assert; + +/** + * Default implementation of a {@link ChannelFactory}, which will create instances of a + * {@link QueueChannel}. + * + * @author Marius Bogoevici + */ +public class QueueChannelFactory extends AbstractChannelFactory{ + + int queueCapacity = QueueChannel.DEFAULT_CAPACITY; + + public int getQueueCapacity() { + return queueCapacity; + } + + /** + * Sets the queue capacity for the newly created channels. + * By default, the queue capacity is {@literal QueueChannel.DEFAULT_CAPACITY} + * @param queueCapacity + */ + public void setQueueCapacity(int queueCapacity) { + Assert.state(queueCapacity > 0, "Queue capacity must be greater than zero"); + this.queueCapacity = queueCapacity; + } + + protected AbstractMessageChannel createChannelInternal(DispatcherPolicy dispatcherPolicy) { + return new QueueChannel(queueCapacity, dispatcherPolicy); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/RendezvousChannelFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/RendezvousChannelFactory.java new file mode 100644 index 0000000000..9ec8321050 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/RendezvousChannelFactory.java @@ -0,0 +1,36 @@ +/* + * 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.PriorityChannel; +import org.springframework.integration.channel.RendezvousChannel; + +/** + * A {@link ChannelFactory} for creating {@link RendezvousChannel} instances. + * @author Marius Bogoevici + * + */ +public class RendezvousChannelFactory extends AbstractChannelFactory { + + @Override + protected AbstractMessageChannel createChannelInternal(DispatcherPolicy dispatcherPolicy) { + return new RendezvousChannel(dispatcherPolicy); + } + +} 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/SynchronousChannelFactory.java new file mode 100644 index 0000000000..418dbefe1e --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/factory/SynchronousChannelFactory.java @@ -0,0 +1,36 @@ +/* + * 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.PriorityChannel; +import org.springframework.integration.dispatcher.SynchronousChannel; + +/** + * A {@link ChannelFactory} for creating {@link SynchronousChannel} instances. + * @author Marius Bogoevici + * + */ +public class SynchronousChannelFactory extends AbstractChannelFactory { + + @Override + protected AbstractMessageChannel createChannelInternal(DispatcherPolicy dispatcherPolicy) { + return new SynchronousChannel(null); + } + +} 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/TestChannelFactory.java new file mode 100644 index 0000000000..bc8f83e3fe --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/factory/TestChannelFactory.java @@ -0,0 +1,182 @@ +/* + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.List; + +import org.junit.Before; +import org.junit.Test; +import org.omg.CORBA.REBIND; +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.FactoryBean; +import org.springframework.integration.bus.MessageBus; +import org.springframework.integration.channel.AbstractMessageChannel; +import org.springframework.integration.channel.ChannelInterceptor; +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.dispatcher.SynchronousChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.selector.MessageSelector; + +/** + * + * @author Marius Bogoevici + */ +public class TestChannelFactory { + + ArrayList interceptors = null; + + DispatcherPolicy dispatcherPolicy = null; + + @Before + public void createInterceptorsList() { + interceptors = new ArrayList(); + interceptors.add(new TestChannelInterceptor()); + interceptors.add(new TestChannelInterceptor()); + } + + @Before + public void createDispatcherPolicy() { + dispatcherPolicy = new DispatcherPolicy(); + dispatcherPolicy.setMaxMessagesPerTask(100); + } + + @Test + public void testQueueChannelFactory() { + QueueChannelFactory channelFactory = new QueueChannelFactory(); + channelFactory.setQueueCapacity(99); + genericChannelFactoryTests(channelFactory, QueueChannel.class); + assertEquals(99, channelFactory.getQueueCapacity()); + } + + @Test + public void testSynchronousChannelFactory() { + SynchronousChannelFactory channelFactory = new SynchronousChannelFactory(); + assertNotNull(interceptors); + AbstractMessageChannel channel = (AbstractMessageChannel) channelFactory.getChannel(dispatcherPolicy, + interceptors); + assertEquals(SynchronousChannel.class, channel.getClass()); + assertInterceptors(channel); + } + + @Test + public void testRendezvousChannelFactory() { + RendezvousChannelFactory channelFactory = new RendezvousChannelFactory(); + genericChannelFactoryTests(channelFactory, RendezvousChannel.class); + } + + @Test + public void testPriorityChannelFactory() { + RendezvousChannelFactory channelFactory = new RendezvousChannelFactory(); + genericChannelFactoryTests(channelFactory, RendezvousChannel.class); + } + + @Test + public void testDefaultChannelFactoryBean() throws Exception{ + MessageBus messageBus = new MessageBus(); + ChannelFactory channelFactory = new StubChannelFactory(); + messageBus.setChannelFactory(channelFactory); + DefaultChannelFactoryBean channelFactoryBean = new DefaultChannelFactoryBean(); + channelFactoryBean.setDispatcherPolicy(dispatcherPolicy); + channelFactoryBean.setInterceptors(interceptors); + channelFactoryBean.setMessageBus(messageBus); + StubChannel channel = (StubChannel)channelFactoryBean.getObject(); + assertTrue(dispatcherPolicy == channel.getDispatcherPolicy()); + assertInterceptors(channel); + } + + private void genericChannelFactoryTests(ChannelFactory channelFactory, Class expectedChannelClass) { + assertNotNull(dispatcherPolicy); + assertNotNull(interceptors); + AbstractMessageChannel channel = (AbstractMessageChannel) channelFactory.getChannel(dispatcherPolicy, + interceptors); + assertEquals(expectedChannelClass, channel.getClass()); + assertTrue(channel.getDispatcherPolicy() == dispatcherPolicy); + assertInterceptors(channel); + } + + private void assertInterceptors(AbstractMessageChannel channel) { + Object interceptorsWrapper = new DirectFieldAccessor(channel).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) { + return false; + } + + public boolean preSend(Message message, MessageChannel channel) { + return false; + } + + } + + static class StubChannel extends AbstractMessageChannel { + + public StubChannel(DispatcherPolicy dispatcherPolicy) { + super(dispatcherPolicy); + } + + @Override + protected Message doReceive(long timeout) { + return null; + } + + @Override + protected boolean doSend(Message message, long timeout) { + return false; + } + + public List> clear() { + return null; + } + + public List> purge(MessageSelector selector) { + return null; + } + + } + + static class StubChannelFactory extends AbstractChannelFactory { + + @Override + protected AbstractMessageChannel createChannelInternal(DispatcherPolicy dispatcherPolicy) { + return new StubChannel(dispatcherPolicy); + } + + } + +}