INT-1895 Use FactoryBean to Allow Late-Binding of using-nio Attribute

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2011 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.
@@ -183,19 +183,6 @@ public abstract class IpAdapterParserUtils {
 		return multicast;
 	}
 
-	/**
-	 * Gets the use-nio attribute, if present; if not returns 'false'.
-	 * @param element
-	 * @return The value of the attribute or false.
-	 */
-	static String getUseNio(Element element) {
-		String useNio = element.getAttribute(IpAdapterParserUtils.USING_NIO);
-		if (!StringUtils.hasText(useNio)) {
-			useNio = "false";
-		}
-		return useNio;
-	}
-
 	/**
 	 * Sets the common port attributes on the bean being built (timeout, receive buffer size,
 	 * send buffer size).

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java

@@ -0,0 +1,367 @@
+/*
+ * Copyright 2002-2011 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.ip.config;
+
+import java.util.concurrent.Executor;
+
+import org.springframework.beans.factory.config.AbstractFactoryBean;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.core.serializer.Deserializer;
+import org.springframework.core.serializer.Serializer;
+import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain;
+import org.springframework.integration.ip.tcp.connection.TcpListener;
+import org.springframework.integration.ip.tcp.connection.TcpMessageMapper;
+import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory;
+import org.springframework.integration.ip.tcp.connection.TcpSender;
+import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer;
+
+/**
+ * Instantiates a TcpN(et|io)(Server|Client)ConnectionFactory, depending
+ * on type and using-nio attributes.
+ * 
+ * @author Gary Russell
+ * @since 2.0.5
+ *
+ */
+public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<AbstractConnectionFactory> 
+		implements SmartLifecycle {
+
+	private AbstractConnectionFactory connectionFactory;
+	
+	private String type;
+
+	protected String host;
+	
+	protected int port;
+	
+	protected TcpListener listener;
+
+	protected TcpSender sender;
+
+	protected int soTimeout;
+
+	private int soSendBufferSize;
+
+	private int soReceiveBufferSize;
+	
+	private boolean soTcpNoDelay;
+
+	private int soLinger  = -1; // don't set by default
+
+	private boolean soKeepAlive;
+
+	private int soTrafficClass = -1; // don't set by default
+	
+	private Executor taskExecutor;
+	
+	protected Deserializer<?> deserializer = new ByteArrayCrLfSerializer();
+	
+	protected Serializer<?> serializer = new ByteArrayCrLfSerializer();
+	
+	protected TcpMessageMapper mapper = new TcpMessageMapper();
+
+	protected boolean singleUse;
+
+	protected int poolSize = 5;
+
+	protected volatile boolean active;
+
+	protected TcpConnectionInterceptorFactoryChain interceptorFactoryChain;
+	
+	private boolean lookupHost = true;
+	
+	private String localAddress;
+
+	private boolean usingNio;
+	
+	private boolean usingDirectBuffers;
+	
+	@Override
+	public Class<?> getObjectType() {
+		return this.connectionFactory != null ? this.connectionFactory.getClass() 
+											  : AbstractConnectionFactory.class;
+	}
+
+	@Override
+	protected AbstractConnectionFactory createInstance() throws Exception {
+		if (this.usingNio) {
+			if ("server".equals(this.type)) {
+				TcpNioServerConnectionFactory connectionFactory = new TcpNioServerConnectionFactory(this.port);
+				this.setCommonAttributes(connectionFactory);
+				this.setServerAttributes(connectionFactory);
+				connectionFactory.setUsingDirectBuffers(this.usingDirectBuffers);
+				this.connectionFactory = connectionFactory;
+			} else {
+				TcpNioClientConnectionFactory connectionFactory = new TcpNioClientConnectionFactory(
+						this.host, this.port);
+				this.setCommonAttributes(connectionFactory);
+				connectionFactory.setUsingDirectBuffers(this.usingDirectBuffers);
+				this.connectionFactory = connectionFactory;
+			}
+		} else {
+			if ("server".equals(this.type)) {
+				TcpNetServerConnectionFactory connectionFactory = new TcpNetServerConnectionFactory(this.port);
+				this.setCommonAttributes(connectionFactory);
+				this.setServerAttributes(connectionFactory);
+				this.connectionFactory = connectionFactory;
+			} else {
+				TcpNetClientConnectionFactory connectionFactory = new TcpNetClientConnectionFactory(
+						this.host, this.port);
+				this.setCommonAttributes(connectionFactory);
+				this.connectionFactory = connectionFactory;
+			}
+		}
+		return this.connectionFactory;
+	}
+
+	private void setCommonAttributes(AbstractConnectionFactory factory) {
+		factory.setDeserializer(this.deserializer);
+		factory.setInterceptorFactoryChain(this.interceptorFactoryChain);
+		factory.setLookupHost(this.lookupHost);
+		factory.setMapper(this.mapper);
+		factory.setPoolSize(this.poolSize);
+		factory.setSerializer(this.serializer);
+		factory.setSingleUse(this.singleUse);
+		factory.setSoKeepAlive(this.soKeepAlive);
+		factory.setSoLinger(this.soLinger);
+		factory.setSoReceiveBufferSize(this.soReceiveBufferSize);
+		factory.setSoSendBufferSize(this.soSendBufferSize);
+		factory.setSoTcpNoDelay(this.soTcpNoDelay);
+		factory.setSoTimeout(this.soTimeout);
+		factory.setSoTrafficClass(this.soTrafficClass);
+		factory.setTaskExecutor(this.taskExecutor);
+	}
+
+	private void setServerAttributes(AbstractServerConnectionFactory factory) {
+		factory.setLocalAddress(this.localAddress);
+	}
+
+	/**
+	 * @param port the port to set
+	 */
+	public void setPort(int port) {
+		this.port = port;
+	}
+
+	/**
+	 * @param host the host to set
+	 */
+	public void setHost(String host) {
+		this.host = host;
+	}
+
+	/**
+	 * @param server the server to set
+	 */
+	public void setType(String type) {
+		this.type = type;
+	}
+
+	/**
+	 * @param localAddress
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory#setLocalAddress(java.lang.String)
+	 */
+	public void setLocalAddress(String localAddress) {
+		this.localAddress = localAddress;
+	}
+
+	/**
+	 * @param soTimeout
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setSoTimeout(int)
+	 */
+	public void setSoTimeout(int soTimeout) {
+		this.soTimeout = soTimeout;
+	}
+
+	/**
+	 * @param soReceiveBufferSize
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setSoReceiveBufferSize(int)
+	 */
+	public void setSoReceiveBufferSize(int soReceiveBufferSize) {
+		this.soReceiveBufferSize = soReceiveBufferSize;
+	}
+
+	/**
+	 * @param soSendBufferSize
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setSoSendBufferSize(int)
+	 */
+	public void setSoSendBufferSize(int soSendBufferSize) {
+		this.soSendBufferSize = soSendBufferSize;
+	}
+
+	/**
+	 * @param soTcpNoDelay
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setSoTcpNoDelay(boolean)
+	 */
+	public void setSoTcpNoDelay(boolean soTcpNoDelay) {
+		this.soTcpNoDelay = soTcpNoDelay;
+	}
+
+	/**
+	 * @param soLinger
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setSoLinger(int)
+	 */
+	public void setSoLinger(int soLinger) {
+		this.soLinger = soLinger;
+	}
+
+	/**
+	 * @param soKeepAlive
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setSoKeepAlive(boolean)
+	 */
+	public void setSoKeepAlive(boolean soKeepAlive) {
+		this.soKeepAlive = soKeepAlive;
+	}
+
+	/**
+	 * @param soTrafficClass
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setSoTrafficClass(int)
+	 */
+	public void setSoTrafficClass(int soTrafficClass) {
+		this.soTrafficClass = soTrafficClass;
+	}
+
+	/**
+	 * @param usingNio the usingNio to set
+	 */
+	public void setUsingNio(boolean usingNio) {
+		this.usingNio = usingNio;
+	}
+
+	/**
+	 * @param usingDirectBuffers
+	 * @see org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory#setUsingDirectBuffers(boolean)
+	 */
+	public void setUsingDirectBuffers(boolean usingDirectBuffers) {
+		this.usingDirectBuffers = usingDirectBuffers;
+	}
+
+	/**
+	 * @param taskExecutor
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setTaskExecutor(java.util.concurrent.Executor)
+	 */
+	public void setTaskExecutor(Executor taskExecutor) {
+		this.taskExecutor = taskExecutor;
+	}
+
+	/**
+	 * @param deserializer
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setDeserializer(org.springframework.core.serializer.Deserializer)
+	 */
+	public void setDeserializer(Deserializer<?> deserializer) {
+		this.deserializer = deserializer;
+	}
+
+	/**
+	 * @param serializer
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setSerializer(org.springframework.core.serializer.Serializer)
+	 */
+	public void setSerializer(Serializer<?> serializer) {
+		this.serializer = serializer;
+	}
+
+	/**
+	 * @param mapper
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setMapper(org.springframework.integration.ip.tcp.connection.TcpMessageMapper)
+	 */
+	public void setMapper(TcpMessageMapper mapper) {
+		this.mapper = mapper; 
+	}
+
+	/**
+	 * @param singleUse
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setSingleUse(boolean)
+	 */
+	public void setSingleUse(boolean singleUse) {
+		this.singleUse = singleUse;
+	}
+
+	/**
+	 * @param poolSize
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setPoolSize(int)
+	 */
+	public void setPoolSize(int poolSize) {
+		this.poolSize = poolSize;
+	}
+
+	/**
+	 * @param interceptorFactoryChain
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setInterceptorFactoryChain(org.springframework.integration.ip.tcp.connection.TcpConnectionInterceptorFactoryChain)
+	 */
+	public void setInterceptorFactoryChain(
+			TcpConnectionInterceptorFactoryChain interceptorFactoryChain) {
+		this.interceptorFactoryChain = interceptorFactoryChain;
+	}
+
+	/**
+	 * @param lookupHost
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#setLookupHost(boolean)
+	 */
+	public void setLookupHost(boolean lookupHost) {
+		this.lookupHost = lookupHost;
+	}
+
+	/**
+	 * 
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#start()
+	 */
+	public void start() {
+		this.connectionFactory.start();
+	}
+
+	/**
+	 * 
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#stop()
+	 */
+	public void stop() {
+		this.connectionFactory.stop();
+	}
+
+	/**
+	 * @return
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#getPhase()
+	 */
+	public int getPhase() {
+		return this.connectionFactory.getPhase();
+	}
+
+	/**
+	 * @return
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#isAutoStartup()
+	 */
+	public boolean isAutoStartup() {
+		return this.connectionFactory.isAutoStartup();
+	}
+
+	/**
+	 * @param callback
+	 * @see org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory#stop(java.lang.Runnable)
+	 */
+	public void stop(Runnable callback) {
+		this.connectionFactory.stop(callback);
+	}
+
+	public boolean isRunning() {
+		return this.connectionFactory.isRunning();
+	}
+
+}
+

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2011 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.
@@ -31,42 +31,29 @@ import org.w3c.dom.Element;
  */
 public class TcpConnectionParser extends AbstractBeanDefinitionParser {
 	
-	private static final String BASE_PACKAGE = "org.springframework.integration.ip.tcp.connection";
+	private static final String BASE_PACKAGE = "org.springframework.integration.ip.config";
 
 	@Override
 	protected AbstractBeanDefinition parseInternal(Element element,
 			ParserContext parserContext) {
 		BeanDefinitionBuilder builder = null;
-		String useNio = IpAdapterParserUtils.getUseNio(element);
 		String type = element.getAttribute(IpAdapterParserUtils.TCP_CONNECTION_TYPE);
 		if (!StringUtils.hasText(type)) {
 			parserContext.getReaderContext().error(IpAdapterParserUtils.TCP_CONNECTION_TYPE + 
 					" is required for a tcp connection", element);
-		}
-		if (type.equals("client")) {
-			if (useNio.equals("true")) {
-				builder = BeanDefinitionBuilder.genericBeanDefinition(BASE_PACKAGE + 
-						".TcpNioClientConnectionFactory");
-				IpAdapterParserUtils.addHostAndPortToConstructor(element, builder, parserContext);				
-			} else {
-				builder = BeanDefinitionBuilder.genericBeanDefinition(BASE_PACKAGE +
-						".TcpNetClientConnectionFactory");
-				IpAdapterParserUtils.addHostAndPortToConstructor(element, builder, parserContext);	
-			}
-		} else if (type.equals("server")) {
-			if (useNio.equals("true")) {
-				builder = BeanDefinitionBuilder.genericBeanDefinition(BASE_PACKAGE +
-						".TcpNioServerConnectionFactory");
-				IpAdapterParserUtils.addPortToConstructor(element, builder, parserContext);
-			} else {
-				builder = BeanDefinitionBuilder.genericBeanDefinition(BASE_PACKAGE +
-						".TcpNetServerConnectionFactory");
-				IpAdapterParserUtils.addPortToConstructor(element, builder, parserContext);
-			}
-		} else {
+		} else if (!"server".equals(type) && !"client".equals(type)) {
 			parserContext.getReaderContext().error(IpAdapterParserUtils.TCP_CONNECTION_TYPE + 
 					" must be 'client' or 'server' for a TCP Connection Factory", element);
-		}
+		}		
+		builder = BeanDefinitionBuilder.genericBeanDefinition(BASE_PACKAGE + 
+				".TcpConnectionFactoryFactoryBean");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "type");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, 
+				IpAdapterParserUtils.HOST);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, 
+				IpAdapterParserUtils.PORT);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, 
+				IpAdapterParserUtils.USING_NIO);
 		IpAdapterParserUtils.addCommonSocketOptions(builder, element);
 		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.RECEIVE_BUFFER_SIZE);

spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml

@@ -4,10 +4,14 @@
 	xmlns:int="http://www.springframework.org/schema/integration"
 	xmlns:ip="http://www.springframework.org/schema/integration/ip"
 	xmlns:task="http://www.springframework.org/schema/task"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:util="http://www.springframework.org/schema/util"
 	xsi:schemaLocation="http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd
 		http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
 		http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.0.xsd
-		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd">
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
+		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util-3.0.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd">
 
 	<bean id="tcpIpUtils" class="org.springframework.integration.ip.util.SocketTestUtils" />
 
@@ -162,7 +166,7 @@
 		so-tcp-no-delay="true"
 		so-timeout="1236"
 		so-traffic-class="12"
-		using-nio="true"
+		using-nio="#{props['use.nio']}"
 		single-use="true"
 		task-executor="externalTE"
 		pool-size="321"		
@@ -170,6 +174,10 @@
 		interceptor-factory-chain="interceptors"
 	/>
 	
+	<util:properties id="props">
+		<prop key="use.nio">true</prop>
+	</util:properties>
+	
 	<ip:tcp-connection-factory
 		id="server1"
 		type="server"