diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java index 2c1c734998..2ba9fc45a4 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java +++ b/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). diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java new file mode 100644 index 0000000000..ab2238400b --- /dev/null +++ b/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 + 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(); + } + +} + diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java index 5903425e24..df94e239f9 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java +++ b/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); diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml index 64e0efbe6c..75fb5d5479 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml +++ b/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"> @@ -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" /> + + true + +