Adding ChannelFactory, various implementations for different types of channels and DefaultChannelFactoryBean.

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

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<ChannelInterceptor> interceptors);
+
+}

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<ChannelInterceptor> 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<ChannelInterceptor> 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;
+	}
+
+}

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<Message<?>> comparator;
+
+	public void setCapacity(int capacity) {
+		this.capacity = capacity;
+	}
+
+	public void setComparator(Comparator<Message<?>> comparator) {
+		this.comparator = comparator;
+	}
+
+	@Override
+	protected AbstractMessageChannel createChannelInternal(DispatcherPolicy dispatcherPolicy) {
+		return new PriorityChannel(capacity, dispatcherPolicy, comparator);
+	}
+
+}

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

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

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

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<ChannelInterceptor> interceptors = null;
+
+	DispatcherPolicy dispatcherPolicy = null;
+
+	@Before
+	public void createInterceptorsList() {
+		interceptors = new ArrayList<ChannelInterceptor>();
+		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<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) {
+			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<Message<?>> clear() {
+			return null;
+		}
+
+		public List<Message<?>> purge(MessageSelector selector) {
+			return null;
+		}
+		
+	}
+	
+	static class StubChannelFactory extends AbstractChannelFactory {
+
+		@Override
+		protected AbstractMessageChannel createChannelInternal(DispatcherPolicy dispatcherPolicy) {
+			return new StubChannel(dispatcherPolicy);
+		}
+		
+	}
+
+}