From 46bc422174879eaf2fce0d231e9eeb13472c75d6 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 26 Jul 2010 20:54:51 +0000 Subject: [PATCH] INT-1269 Refactor TCP Adapters to Separate Out Connection Factories - Docs to Follow --- .../integration/ip/IpHeaders.java | 2 + .../ip/config/IpAdapterParserUtils.java | 20 + .../ip/config/IpNamespaceHandler.java | 3 + .../ip/config/TcpConnectionParser.java | 101 +++ .../TcpInboundChannelAdapterParser.java | 45 ++ .../TcpOutboundChannelAdapterParser.java | 42 ++ .../integration/ip/tcp/SocketIoUtils.java | 11 + .../ip/tcp/TcpReceivingChannelAdapter.java | 89 +++ .../ip/tcp/TcpSendingMessageHandler.java | 157 ++++ .../AbstractClientConnectionFactory.java | 75 ++ .../connection/AbstractConnectionFactory.java | 313 ++++++++ .../AbstractServerConnectionFactory.java | 105 +++ .../tcp/connection/AbstractTcpConnection.java | 138 ++++ .../ip/tcp/connection/ConnectionFactory.java | 32 + .../ip/tcp/connection/TcpConnection.java | 124 ++++ .../ip/tcp/connection/TcpListener.java | 44 ++ .../ip/tcp/connection/TcpMessageMapper.java | 114 +++ .../TcpNetClientConnectionFactory.java | 74 ++ .../ip/tcp/connection/TcpNetConnection.java | 152 ++++ .../TcpNetServerConnectionFactory.java | 100 +++ .../TcpNioClientConnectionFactory.java | 194 +++++ .../ip/tcp/connection/TcpNioConnection.java | 353 +++++++++ .../TcpNioServerConnectionFactory.java | 218 ++++++ .../ip/tcp/connection/TcpSender.java | 42 ++ .../ip/tcp/connection/package-info.java | 6 + .../AbstractByteArrayStreamingConverter.java | 66 ++ .../tcp/converter/ByteArrayCrLfConverter.java | 76 ++ .../ByteArrayLengthHeaderConverter.java | 113 +++ .../converter/ByteArrayStxEtxConverter.java | 81 +++ .../converter/SoftEndOfStreamException.java | 49 ++ .../ip/tcp/converter/package-info.java | 6 + .../ip/config/spring-integration-ip-2.0.xsd | 177 +++++ .../ip/config/ParserUnitTests-context.xml | 59 ++ .../ip/config/ParserUnitTests.java | 96 +++ .../ConnectionToConnectionTests-context.xml | 26 + .../ip/tcp/ConnectionToConnectionTests.java | 81 +++ .../ip/tcp/SharedConnectionTests-context.xml | 58 ++ .../ip/tcp/SharedConnectionTests.java | 80 +++ .../tcp/TcpReceivingChannelAdapterTests.java | 367 ++++++++++ .../ip/tcp/TcpSendingMessageHandlerTests.java | 670 +++++++++++++++++- .../tcp/connection/TcpNioConnectionTests.java | 121 ++++ spring-integration-ip/template.mf | 1 + 42 files changed, 4677 insertions(+), 4 deletions(-) create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundChannelAdapterParser.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ConnectionFactory.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSender.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/package-info.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/AbstractByteArrayStreamingConverter.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayCrLfConverter.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayLengthHeaderConverter.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayStxEtxConverter.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/SoftEndOfStreamException.java create mode 100644 spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/package-info.java create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests-context.xml create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests.java create mode 100644 spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java index 7114ff0944..041f6f36ca 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/IpHeaders.java @@ -44,4 +44,6 @@ public abstract class IpHeaders { public static final String REMOTE_PORT = TCP + "remote_port"; + public static final String CONNECTION_ID = IP + "connection_id"; + } 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 8d6275e177..6581bc3c33 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 @@ -93,6 +93,15 @@ public abstract class IpAdapterParserUtils { static final String TASK_EXECUTOR = "task-executor"; + static final String TCP_CONNECTION_TYPE = "type"; + + static final String INPUT_CONVERTER = "input-converter"; + + static final String OUTPUT_CONVERTER = "output-converter"; + + static final String SINGLE_USE = "single-use"; + + static final String TCP_CONNECTION_FACTORY = "connection-factory"; /** * Adds a constructor-arg to the provided bean definition builder @@ -129,6 +138,17 @@ public abstract class IpAdapterParserUtils { builder.addConstructorArgValue(port); } + /** + * @param element + * @param builder + * @param parserContext + */ + public static void addPortToConstructor(Element element, + BeanDefinitionBuilder builder, ParserContext parserContext) { + String port = IpAdapterParserUtils.getPort(element, parserContext); + builder.addConstructorArgValue(port); + } + /** * Asserts that a protocol attribute (udp or tcp) is supplied, * @param element diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java index 00f5dba003..1a542eaaa1 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpNamespaceHandler.java @@ -31,6 +31,9 @@ public class IpNamespaceHandler extends AbstractIntegrationNamespaceHandler { this.registerBeanDefinitionParser("outbound-channel-adapter", new IpOutboundChannelAdapterParser()); this.registerBeanDefinitionParser("inbound-gateway", new IpInboundGatewayParser()); this.registerBeanDefinitionParser("outbound-gateway", new IpOutboundGatewayParser()); + this.registerBeanDefinitionParser("tcp-connection-factory", new TcpConnectionParser()); + this.registerBeanDefinitionParser("tcp-inbound-channel-adapter", new TcpInboundChannelAdapterParser()); + this.registerBeanDefinitionParser("tcp-outbound-channel-adapter", new TcpOutboundChannelAdapterParser()); } } 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 new file mode 100644 index 0000000000..aa8f3baf72 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionParser.java @@ -0,0 +1,101 @@ +/* + * Copyright 2002-2010 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 org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.AbstractBeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.core.Conventions; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +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.util.StringUtils; +import org.w3c.dom.Element; + +/** + * @author Gary Russell + * @since 2.0 + * + */ +public class TcpConnectionParser extends AbstractBeanDefinitionParser { + + @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( + TcpNioClientConnectionFactory.class); + IpAdapterParserUtils.addHostAndPortToConstructor(element, builder, parserContext); + } else { + builder = BeanDefinitionBuilder.genericBeanDefinition( + TcpNetClientConnectionFactory.class); + IpAdapterParserUtils.addHostAndPortToConstructor(element, builder, parserContext); + } + } else if (type.equals("server")) { + if (useNio.equals("true")) { + builder = BeanDefinitionBuilder.genericBeanDefinition( + TcpNioServerConnectionFactory.class); + IpAdapterParserUtils.addPortToConstructor(element, builder, parserContext); + } else { + builder = BeanDefinitionBuilder.genericBeanDefinition( + TcpNetServerConnectionFactory.class); + IpAdapterParserUtils.addPortToConstructor(element, builder, parserContext); + } + } else { + parserContext.getReaderContext().error(IpAdapterParserUtils.TCP_CONNECTION_TYPE + + " must be 'client' or 'server' for an IP channel adapter", element); + } + IpAdapterParserUtils.addCommonSocketOptions(builder, element); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.RECEIVE_BUFFER_SIZE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.USING_DIRECT_BUFFERS); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.SO_KEEP_ALIVE); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.SO_LINGER); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.SO_TCP_NODELAY); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.SO_TRAFFIC_CLASS); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.POOL_SIZE); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + IpAdapterParserUtils.TASK_EXECUTOR); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + IpAdapterParserUtils.INPUT_CONVERTER); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + IpAdapterParserUtils.OUTPUT_CONVERTER); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.SINGLE_USE); + + return builder.getBeanDefinition(); + } + + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java new file mode 100644 index 0000000000..4a69039149 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpInboundChannelAdapterParser.java @@ -0,0 +1,45 @@ +/* + * Copyright 2002-2010 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 org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractChannelAdapterParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter; +import org.w3c.dom.Element; + +/** + * Channel Adapter that receives UDP datagram packets and maps them to Messages. + * + * @author Gary Russell + * @since 2.0 + */ +public class TcpInboundChannelAdapterParser extends AbstractChannelAdapterParser { + + protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( + TcpReceivingChannelAdapter.class); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + IpAdapterParserUtils.TCP_CONNECTION_FACTORY); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, + element, "channel", "outputChannel"); + return builder.getBeanDefinition(); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundChannelAdapterParser.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundChannelAdapterParser.java new file mode 100644 index 0000000000..2f0f2ba0dd --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpOutboundChannelAdapterParser.java @@ -0,0 +1,42 @@ +/* + * Copyright 2002-2010 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 org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.ip.tcp.TcpSendingMessageHandler; +import org.w3c.dom.Element; + +/** + * @author Gary Russell + * @since 2.0 + */ +public class TcpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser { + + protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( + TcpSendingMessageHandler.class); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, + IpAdapterParserUtils.TCP_CONNECTION_FACTORY); + return builder.getBeanDefinition(); + } + + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketIoUtils.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketIoUtils.java index 51d6d5fc8c..412ee5d549 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketIoUtils.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/SocketIoUtils.java @@ -16,6 +16,7 @@ package org.springframework.integration.ip.tcp; import java.lang.reflect.Constructor; +import java.net.InetAddress; import java.net.Socket; import java.nio.channels.SocketChannel; @@ -122,4 +123,14 @@ public class SocketIoUtils { return writer; } + public static String getSocketId(Socket socket) { + InetAddress inetAddress = socket.getInetAddress(); + String hostName = ""; + if (inetAddress != null) { + hostName = inetAddress.getHostName(); + } + return hostName + ":" + socket.getPort() + ":" + + socket.hashCode(); + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java new file mode 100644 index 0000000000..4b313148d9 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapter.java @@ -0,0 +1,89 @@ +/* + * Copyright 2002-2010 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.tcp; + +import java.net.ServerSocket; + +import org.springframework.integration.core.Message; +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; +import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.ConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpListener; + +/** + * Tcp inbound channel adapter using a TcpConnection to + * receive data - if the connection factory is a server + * factory, this Listener owns the connections. If it is + * a client factory, the sender owns the connection. + * + * @author Gary Russell + * @since 2.0 + * + */ +public class TcpReceivingChannelAdapter + extends MessageProducerSupport implements TcpListener { + + protected ServerSocket serverSocket; + + protected Class customSocketReaderClass; + + protected ConnectionFactory clientConnectionFactory; + + protected ConnectionFactory serverConnectionFactory; + + public void onMessage(Message message) { + sendMessage(message); + } + + @Override + protected void doStart() { + // Nothing to do; we're passive + } + + @Override + protected void doStop() { + // Nothing to do; we're passive + } + + /** + * Sets the client or server connection factory; for this (an inbound adapter), if + * the factory is a client connection factory, the sockets are owned by a sending + * channel adapter and this adapter is used to receive replies. + * + * @param connectionFactory the connectionFactory to set + */ + public void setConnectionFactory(AbstractConnectionFactory connectionFactory) { + if (connectionFactory instanceof AbstractClientConnectionFactory) { + this.clientConnectionFactory = connectionFactory; + } else { + this.serverConnectionFactory = connectionFactory; + } + connectionFactory.registerListener(this); + } + + public boolean isListening() { + if (this.serverConnectionFactory == null) { + return false; + } + if (this.serverConnectionFactory instanceof AbstractServerConnectionFactory) { + return ((AbstractServerConnectionFactory) this.serverConnectionFactory).isListening(); + } + return false; + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java new file mode 100644 index 0000000000..5258f14b79 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandler.java @@ -0,0 +1,157 @@ +/* + * Copyright 2002-2010 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.tcp; + +import java.io.IOException; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.integration.core.Message; +import org.springframework.integration.ip.IpHeaders; +import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; +import org.springframework.integration.ip.tcp.connection.ConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpConnection; +import org.springframework.integration.ip.tcp.connection.TcpSender; +import org.springframework.integration.message.MessageDeliveryException; +import org.springframework.integration.message.MessageHandler; +import org.springframework.integration.message.MessageHandlingException; +import org.springframework.integration.message.MessageMappingException; +import org.springframework.integration.message.MessageRejectedException; + +/** + * Tcp outbound channel adapter using a TcpConnection to + * send data - if the connection factory is a server + * factory, the TcpListener owns the connections. If it is + * a client factory, this object owns the connection. + * @author Gary Russell + * @since 2.0 + * + */ +public class TcpSendingMessageHandler implements MessageHandler, TcpSender { + + protected Log logger = LogFactory.getLog(this.getClass()); + + protected TcpConnection connection; + + protected ConnectionFactory clientConnectionFactory; + + protected ConnectionFactory serverConnectionFactory; + + protected Map connections = new ConcurrentHashMap(); + + protected synchronized TcpConnection getConnection() { + try { + this.connection = clientConnectionFactory.getConnection(); + } catch (Exception e) { + logger.error("Error creating SocketWriter", e); + } + return this.connection; + } + + /** + * Close the underlying socket and prepare to establish a new socket on + * the next write. + */ + protected void close() { + this.connection.close(); + this.connection = null; + } + + /** + * Writes the message payload to the underlying socket, using the specified + * message format. + * @see org.springframework.integration.message.MessageHandler#handleMessage(org.springframework.integration.core.Message) + */ + public void handleMessage(final Message message) throws MessageRejectedException, + MessageHandlingException, MessageDeliveryException { + if (this.serverConnectionFactory != null) { + // We don't own the connection + Object connectionId = message.getHeaders().get(IpHeaders.CONNECTION_ID); + TcpConnection connection = connections.get(connectionId); + if (connection != null) { + try { + connection.send(message); + } catch (Exception e) { + logger.error("Error sending message", e); + connection.close(); + } + } else { + logger.error("Unable to find incoming socket for " + message); + } + return; + } + + + try { + doWrite(message); + } catch (MessageMappingException e) { + // retry - socket may have closed + if (e.getCause() instanceof IOException) { + doWrite(message); + } else { + throw e; + } + } + } + + /** + * Method that actually does the write. + * @param message The message to write. + */ + protected void doWrite(Message message) { + try { + TcpConnection connection = getConnection(); + if (connection == null) { + throw new MessageMappingException(message, "Failed to create connection"); + } + connection.send(message); + } catch (Exception e) { + this.connection = null; + if (e instanceof MessageMappingException) { + throw (MessageMappingException) e; + } + throw new MessageMappingException(message, "Failed to map message", e); + } + } + + /** + * Sets the client or server connection factory; for this (an outbound adapter), if + * the factory is a server connection factory, the sockets are owned by a receiving + * channel adapter and this adapter is used to send replies. + * + * @param connectionFactory the connectionFactory to set + */ + public void setConnectionFactory(AbstractConnectionFactory connectionFactory) { + if (connectionFactory instanceof AbstractClientConnectionFactory) { + this.clientConnectionFactory = connectionFactory; + } else { + this.serverConnectionFactory = connectionFactory; + connectionFactory.registerSender(this); + } + } + + public void addNewConnection(TcpConnection connection) { + connections.put(connection.getConnectionId(), connection); + } + + public void removeDeadConnection(TcpConnection connection) { + connections.remove(connection.getConnectionId()); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java new file mode 100644 index 0000000000..5e715fa931 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractClientConnectionFactory.java @@ -0,0 +1,75 @@ +/* + * Copyright 2002-2010 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.tcp.connection; + +import java.net.Socket; +import java.net.SocketException; +import java.util.concurrent.Executors; + +import org.springframework.util.Assert; + +/** + * Abstract class for client connection factories; client connection factories + * establish outgoing connections. + * @author Gary Russell + * @since 2.0 + * + */ +public abstract class AbstractClientConnectionFactory extends AbstractConnectionFactory { + + /** + * Constructs a factory that will established connections to the host and port. + * @param host The host. + * @param port The port. + */ + public AbstractClientConnectionFactory(String host, int port) { + Assert.notNull(host, "host must not be null"); + this.host = host; + this.port = port; + } + + /** + * Transfers attributes such as converters, singleUse etc to a new connection. + * When the connection factory has a reference to a TCPListener (to read + * responses), or for single use connections, the connection is executed. + * Single use connections need to read from the connection in order to + * close it after the socket timeout. + * @param connection The new connection. + * @param socket The new socket. + */ + protected void initializeConnection(TcpConnection connection, Socket socket) { + if (this.listener != null) { + connection.registerListener(this.listener); + } + if (this.listener != null || this.singleUse) { + if (this.taskExecutor == null) { + this.taskExecutor = Executors.newFixedThreadPool(this.poolSize); + } + if (this.soTimeout <= 0) { + try { + socket.setSoTimeout(DEFAULT_REPLY_TIMEOUT); + } catch (SocketException e) { + logger.error("Error setting default reply timeout", e); + } + } + } + connection.setMapper(this.mapper); + connection.setInputConverter(this.inputConverter); + connection.setOutputConverter(this.outputConverter); + connection.setSingleUse(this.singleUse); + } +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java new file mode 100644 index 0000000000..b498d3f50a --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java @@ -0,0 +1,313 @@ +/* + * Copyright 2002-2010 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.tcp.connection; + +import java.net.Socket; +import java.net.SocketException; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.commons.serializer.InputStreamingConverter; +import org.springframework.commons.serializer.OutputStreamingConverter; +import org.springframework.context.Lifecycle; +import org.springframework.core.task.TaskExecutor; +import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter; +import org.springframework.util.Assert; + +/** + * Base class for all connection factories. + * + * @author Gary Russell + * @since 2.0 + * + */ +public abstract class AbstractConnectionFactory + implements ConnectionFactory, Runnable, Lifecycle { + + protected Log logger = LogFactory.getLog(this.getClass()); + + protected final static int DEFAULT_REPLY_TIMEOUT = 10000; + + 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; + + private boolean soKeepAlive; + + private int soTrafficClass; + + protected Executor taskExecutor; + + protected InputStreamingConverter inputConverter = new ByteArrayCrLfConverter(); + + protected OutputStreamingConverter outputConverter = new ByteArrayCrLfConverter(); + + protected TcpMessageMapper mapper = new TcpMessageMapper(); + + protected boolean singleUse; + + protected int poolSize = 5; + + protected boolean active; + + /** + * Sets socket attributes on the socket. + * @param socket The socket. + * @throws SocketException + */ + protected void setSocketAttributes(Socket socket) throws SocketException { + if (this.soTimeout >= 0) { + socket.setSoTimeout(this.soTimeout); + } + if (this.soSendBufferSize > 0) { + socket.setSendBufferSize(this.soSendBufferSize); + } + if (this.soReceiveBufferSize > 0) { + socket.setReceiveBufferSize(this.soReceiveBufferSize); + } + socket.setTcpNoDelay(this.soTcpNoDelay); + if (this.soLinger >= 0) { + socket.setSoLinger(true, this.soLinger); + } + if (this.soTrafficClass >= 0) { + socket.setTrafficClass(this.soTrafficClass); + } + socket.setKeepAlive(this.soKeepAlive); + } + + /** + * @return the soTimeout + */ + public int getSoTimeout() { + return soTimeout; + } + + /** + * @param soTimeout the soTimeout to set + */ + public void setSoTimeout(int soTimeout) { + this.soTimeout = soTimeout; + } + + /** + * @return the soReceiveBufferSize + */ + public int getSoReceiveBufferSize() { + return soReceiveBufferSize; + } + + /** + * @param soReceiveBufferSize the soReceiveBufferSize to set + */ + public void setSoReceiveBufferSize(int soReceiveBufferSize) { + this.soReceiveBufferSize = soReceiveBufferSize; + } + + /** + * @return the soSendBufferSize + */ + public int getSoSendBufferSize() { + return soSendBufferSize; + } + + /** + * @param soSendBufferSize the soSendBufferSize to set + */ + public void setSoSendBufferSize(int soSendBufferSize) { + this.soSendBufferSize = soSendBufferSize; + } + + /** + * @return the soTcpNoDelay + */ + public boolean isSoTcpNoDelay() { + return soTcpNoDelay; + } + + /** + * @param soTcpNoDelay the soTcpNoDelay to set + */ + public void setSoTcpNoDelay(boolean soTcpNoDelay) { + this.soTcpNoDelay = soTcpNoDelay; + } + + /** + * @return the soLinger + */ + public int getSoLinger() { + return soLinger; + } + + /** + * @param soLinger the soLinger to set + */ + public void setSoLinger(int soLinger) { + this.soLinger = soLinger; + } + + /** + * @return the soKeepAlive + */ + public boolean isSoKeepAlive() { + return soKeepAlive; + } + + /** + * @param soKeepAlive the soKeepAlive to set + */ + public void setSoKeepAlive(boolean soKeepAlive) { + this.soKeepAlive = soKeepAlive; + } + + /** + * @return the soTrafficClass + */ + public int getSoTrafficClass() { + return soTrafficClass; + } + + /** + * @param soTrafficClass the soTrafficClass to set + */ + public void setSoTrafficClass(int soTrafficClass) { + this.soTrafficClass = soTrafficClass; + } + + /** + * @return the host + */ + public String getHost() { + return host; + } + + /** + * @return the port + */ + public int getPort() { + return port; + } + + /** + * Registers a TcpListener to receive messages after + * the payload has been converted from the input data. + * @param listener the TcpListener. + */ + public void registerListener(TcpListener listener) { + Assert.isNull(this.listener, this.getClass().getName() + + " may only be used by one inbound adapter"); + this.listener = listener; + } + + /** + * Registers a TcpSender; for server sockets, used to + * provide connection information so a sender can be used + * to reply to incoming messages. + * @param tcpSendingMessageHandler + */ + public void registerSender(TcpSender sender) { + Assert.isNull(this.sender, this.getClass().getName() + + " may only be used by one outbound adapter"); + this.sender = sender; + } + + /** + * @param taskExecutor the taskExecutor to set + */ + public void setTaskExecutor(Executor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + /** + * + * @param converter the inputConverter to set + */ + public void setInputConverter(InputStreamingConverter converter) { + this.inputConverter = converter; + } + + /** + * + * @param outputConverter the outputConverter to set + */ + public void setOutputConverter(OutputStreamingConverter outputConverter) { + this.outputConverter = outputConverter; + } + + /** + * + * @param mapper the mapper to set; defaults to a {@link TcpMessageMapper} + */ + public void setMapper(TcpMessageMapper mapper) { + this.mapper = mapper; + } + + /** + * If true, sockets created by this factory will be used once. + * @param singleUse + */ + public void setSingleUse(boolean singleUse) { + this.singleUse = singleUse; + } + + public void setPoolSize(int poolSize) { + this.poolSize = poolSize; + } + + /** + * Closes the server. + */ + public abstract void close(); + + /** + * Creates a taskExecutor (if one was not provided) and starts + * the listening process on one of its threads. + */ + public void start() { + if (this.taskExecutor == null) { + this.taskExecutor = Executors.newFixedThreadPool(this.poolSize); + } + this.active = true; + this.taskExecutor.execute(this); + } + + /** + * Stops the server. + */ + public void stop() { + this.active = false; + this.close(); + } + + + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java new file mode 100644 index 0000000000..3cff726112 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java @@ -0,0 +1,105 @@ +/* + * Copyright 2002-2010 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.tcp.connection; + +import java.net.Socket; +import java.net.SocketException; + + + +/** + * Base class for all server connection factories. Server connection factories + * listen on a port for incoming connections and create new TcpConnection objects + * for each new connection. + * @author Gary Russell + * + */ +public abstract class AbstractServerConnectionFactory extends AbstractConnectionFactory { + + protected boolean listening; + + protected String localAddress; + + /** + * The port on which the factory will listen. + * @param port + */ + public AbstractServerConnectionFactory(int port) { + this.port = port; + } + + /** + * Not supported because the factory manages multiple connections and this + * method cannot discriminate. + */ + public TcpConnection getConnection() throws Exception { + throw new UnsupportedOperationException("Getting a connection from a server factory is not supported"); + } + + /** + * + * @return true if the server is listening on the port. + */ + public boolean isListening() { + return listening; + } + + /** + * Transfers attributes such as converters, singleUse etc to a new connection. + * For single use sockets, enforces a socket timeout (default 10 seconds). + * @param connection The new connection. + * @param socket The new socket. + */ + protected void initializeConnection(TcpConnection connection, Socket socket) { + connection.registerListener(this.listener); + connection.registerSender(this.sender); + connection.setMapper(this.mapper); + connection.setInputConverter(this.inputConverter); + connection.setOutputConverter(this.outputConverter); + connection.setSingleUse(this.singleUse); + /* + * If we have a collaborating outbound channel adapter and we are configured + * for single use; need to enforce a timeout on the socket so we will close + * it some period after the response was sent (timeout on the next read). + */ + if (this.singleUse && this.soTimeout <= 0 && this.listener != null) { + try { + socket.setSoTimeout(DEFAULT_REPLY_TIMEOUT); + } catch (SocketException e) { + logger.error("Error setting default reply timeout", e); + } + } + + } + + /** + * + * @return the localAddress + */ + public String getLocalAddress() { + return localAddress; + } + + /** + * Used on multi-homed systems to enforce the server to listen + * on a specfic network address instead of all network adapters. + * @param localAddress the ip address of the required adapter. + */ + public void setLocalAddress(String localAddress) { + this.localAddress = localAddress; + } +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java new file mode 100644 index 0000000000..af1889ca60 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractTcpConnection.java @@ -0,0 +1,138 @@ +/* + * Copyright 2002-2010 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.tcp.connection; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.commons.serializer.InputStreamingConverter; +import org.springframework.commons.serializer.OutputStreamingConverter; +import org.springframework.integration.ip.tcp.converter.AbstractByteArrayStreamingConverter; +import org.springframework.util.Assert; + +/** + * Base class for TcpConnections. TcpConnections are established by + * client connection factories (outgoing) or server connection factories + * (incoming). + * + * @author Gary Russell + * @since 2.0 + * + */ +public abstract class AbstractTcpConnection implements TcpConnection { + + protected Log logger = LogFactory.getLog(this.getClass()); + + @SuppressWarnings("rawtypes") + protected InputStreamingConverter inputConverter; + + @SuppressWarnings("rawtypes") + protected OutputStreamingConverter outputConverter; + + protected TcpMessageMapper mapper; + + protected TcpListener listener; + + protected TcpSender sender; + + protected boolean singleUse; + + /** + * Closes this connection. + */ + public void close() { + if (this.sender != null) { + this.sender.removeDeadConnection(this); + } + } + + /** + * @return the mapper + */ + public TcpMessageMapper getMapper() { + return mapper; + } + + /** + * @param mapper the mapper to set + */ + public void setMapper(TcpMessageMapper mapper) { + Assert.notNull(mapper, this.getClass().getName() + " Mapper may not be null"); + this.mapper = mapper; + if (this.outputConverter != null && + !(this.outputConverter instanceof AbstractByteArrayStreamingConverter)) { + mapper.setStringToBytes(false); + } + } + + /** + * @param converter the input converter to set + */ + public void setInputConverter(InputStreamingConverter inputConverter) { + this.inputConverter = inputConverter; + } + + /** + * @param converter the output converter to set + */ + public void setOutputConverter(OutputStreamingConverter outputConverter) { + this.outputConverter = outputConverter; + if (!(outputConverter instanceof AbstractByteArrayStreamingConverter)) { + mapper.setStringToBytes(false); + } + } + + /** + * @param sender the listener to set + */ + public void registerListener(TcpListener listener) { + this.listener = listener; + } + + /** + * @param sender the sender to set + */ + public void registerSender(TcpSender sender) { + this.sender = sender; + if (sender != null) { + sender.addNewConnection(this); + } + } + + /** + * @return the listener + */ + public TcpListener getListener() { + return this.listener; + } + + /** + * @param singleUse true if this socket is to used once and + * discarded. + */ + public void setSingleUse(boolean singleUse) { + this.singleUse = singleUse; + } + + /** + * + * @return True if connection is used once. + */ + public boolean isSingleUse() { + return this.singleUse; + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ConnectionFactory.java new file mode 100644 index 0000000000..ea5fa3a148 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/ConnectionFactory.java @@ -0,0 +1,32 @@ +/* + * Copyright 2002-2010 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.tcp.connection; + + + +/** + * A factory used to create TcpConnection objects. + * + * @author Gary Russell + * @since 2.0 + * + */ +public interface ConnectionFactory { + + public TcpConnection getConnection() throws Exception; + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java new file mode 100644 index 0000000000..d017af5b93 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnection.java @@ -0,0 +1,124 @@ +/* + * Copyright 2002-2010 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.tcp.connection; + +import java.net.Socket; +import java.nio.channels.SocketChannel; + +import org.springframework.commons.serializer.InputStreamingConverter; +import org.springframework.commons.serializer.OutputStreamingConverter; +import org.springframework.integration.core.Message; + +/** + * An abstraction over {@link Socket} and {@link SocketChannel} that + * sends {@link Message} objects by converting the payload + * and streaming it to the destination. Requires a {@link TcpListener} + * to receive incoming messages. + * + * @author Gary Russell + * @since 2.0 + * + */ +public interface TcpConnection extends Runnable { + + /** + * Closes the connection. + */ + public void close(); + + /** + * @return true if the connection is open. + */ + public boolean isOpen(); + + /** + * Converts and sends the message. + * @param message The message + * @throws Exception + */ + public void send(Message message) throws Exception; + + /** + * Uses the input converter to obtain the message payload + * from the connection's input stream. + * @return + * @throws Exception + */ + public Object getPayload() throws Exception; + + /** + * @return the host name + */ + public String getHostName(); + + /** + * @return the host address + */ + public String getHostAddress(); + + /** + * @return the port + */ + public int getPort(); + + /** + * Sets the listener that will receive incoming Messages. + * @param listener + */ + public void registerListener(TcpListener listener); + + /** + * Registers a sender. Used on server side sockets so a + * sender can determine which connection to send a reply + * to. + * @param sender the sender + */ + public void registerSender(TcpSender sender); + + /** + * @return a string uniquely representing a connection. + */ + public String getConnectionId(); + + /** + * When true, the socket is used once and discarded. + * @param singleUse + */ + public void setSingleUse(boolean singleUse); + + /** + * + * @return True if connection is used once. + */ + public boolean isSingleUse(); + + /** + * @param mapper the mapper + */ + public void setMapper(TcpMessageMapper mapper); + + /** + * @param inputConverter the inputConverter to set + */ + public void setInputConverter(InputStreamingConverter inputConverter); + + /** + * @param outputConverter the outputConverter to set + */ + public void setOutputConverter(OutputStreamingConverter outputConverter); + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java new file mode 100644 index 0000000000..6a16f96f44 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpListener.java @@ -0,0 +1,44 @@ +/* + * Copyright 2002-2010 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.tcp.connection; + +import org.springframework.integration.core.Message; + +/** + * Classes that implement this interface may register with a + * connection factory to receive messages retrieved from a + * {@link TcpConnection} + * @author Gary Russell + * @since 2.0 + * + */ +public interface TcpListener { + + /** + * Called by a TCPConnection when a new message arrives. + * @param message The message. + */ + public abstract void onMessage(Message message); + + /** + * Return true if the connection factory is a server + * and it is listening. + * @return + */ + public boolean isListening(); + +} \ No newline at end of file diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java new file mode 100644 index 0000000000..61d4a0d0ea --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpMessageMapper.java @@ -0,0 +1,114 @@ +/* + * Copyright 2002-2010 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.tcp.connection; + +import java.io.UnsupportedEncodingException; + +import org.springframework.integration.core.Message; +import org.springframework.integration.ip.IpHeaders; +import org.springframework.integration.message.InboundMessageMapper; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.message.MessageHandlingException; +import org.springframework.integration.message.OutboundMessageMapper; + +/** + * Maps incoming data from a {@link TcpConnection} to a {@link Message}. + * If StringToBytes is true (default), + * payloads of type String are converted to a byte[] using the supplied + * charset (UTF-8 by default). + * Inbound messages include headers representing the remote end of the + * connection as well as a connection id that can be used by a {@link TcpSender} + * to correlate which connection to send a reply. + * @author Gary Russell + * @since 2.0 + * + */ +public class TcpMessageMapper implements + InboundMessageMapper, + OutboundMessageMapper { + + private volatile String charset = "UTF-8"; + + private volatile boolean stringToBytes = true; + + public Message toMessage(TcpConnection connection) throws Exception { + Message message = null; + Object payload = connection.getPayload(); + if (payload != null) { + message = MessageBuilder.withPayload(payload) + .setHeader(IpHeaders.HOSTNAME, connection.getHostName()) + .setHeader(IpHeaders.IP_ADDRESS, connection.getHostAddress()) + .setHeader(IpHeaders.REMOTE_PORT, connection.getPort()) + .setHeader(IpHeaders.CONNECTION_ID, connection.getConnectionId()) + .build(); + } + return message; + + } + + public Object fromMessage(Message message) throws Exception { + if (this.stringToBytes) { + return getPayloadAsBytes(message); + } + return message.getPayload(); + } + + /** + * Extracts the payload as a byte array. + * @param message + * @return + */ + private byte[] getPayloadAsBytes(Message message) { + byte[] bytes = null; + Object payload = message.getPayload(); + if (payload instanceof byte[]) { + bytes = (byte[]) payload; + } + else if (payload instanceof String) { + try { + bytes = ((String) payload).getBytes(this.charset); + } + catch (UnsupportedEncodingException e) { + throw new MessageHandlingException(message, e); + } + } + else { + throw new MessageHandlingException(message, + "When using a byte array streaming converter, the socket mapper expects " + + "either a byte array or String payload, but received: " + payload.getClass()); + } + return bytes; + } + + + /** + * @param charset the charset to set + */ + public void setCharset(String charset) { + this.charset = charset; + } + + + /** + * Sets whether outbound String payloads are to be converted + * to byte[]. Default is true. + * @param stringToBytes + */ + public void setStringToBytes(boolean stringToBytes) { + this.stringToBytes = stringToBytes; + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java new file mode 100644 index 0000000000..b10c056308 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java @@ -0,0 +1,74 @@ +/* + * Copyright 2002-2010 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.tcp.connection; + +import java.net.Socket; + +import javax.net.SocketFactory; + +/** + * A client connection factory that creates {@link TcpNetConection}s. + * @author Gary Russell + * @since 2.0 + * + */ +public class TcpNetClientConnectionFactory extends + AbstractClientConnectionFactory { + + protected TcpNetConnection theConnection; + + /** + * Creates a TcpNetClientConnectionFactory for connections to the host and port. + * @param host the host + * @param port the port + */ + public TcpNetClientConnectionFactory(String host, int port) { + super(host, port); + } + + /** + * Obtains a connection - if {@link #setSingleUse(boolean)} was called with + * true, a new connection is returned; otherwise a single connection is + * reused for all requests while the connection remains open. + */ + public TcpNetConnection getConnection() throws Exception { + if (this.theConnection != null && this.theConnection.isOpen()) { + return this.theConnection; + } + logger.debug("Opening new socket connection to " + this.host + ":" + this.port); + Socket socket = SocketFactory.getDefault().createSocket(this.host, this.port); + setSocketAttributes(socket); + TcpNetConnection connection = new TcpNetConnection(socket, false); + initializeConnection(connection, socket); + this.taskExecutor.execute(connection); + if (!this.singleUse) { + this.theConnection = connection; + } + return connection; + } + + public void close() { + } + + public void run() { + } + + public boolean isRunning() { + return this.active; + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java new file mode 100644 index 0000000000..a4cf5d14f0 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java @@ -0,0 +1,152 @@ +/* + * Copyright 2002-2010 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.tcp.connection; + +import java.net.Socket; +import java.net.SocketTimeoutException; + +import org.springframework.commons.serializer.InputStreamingConverter; +import org.springframework.integration.core.Message; +import org.springframework.integration.ip.tcp.SocketIoUtils; +import org.springframework.integration.ip.tcp.converter.SoftEndOfStreamException; + +/** + * A TcpConnection that uses and underlying {@link Socket}. + * + * @author Gary Russell + * @since 2.0 + * + */ +public class TcpNetConnection extends AbstractTcpConnection { + + private final Socket socket; + + private final boolean server; + + /** + * Constructs a TcpNetConnection for the socket. + * @param socket the socket + * @param server if true this connection was created as + * a result of an incoming request. + */ + public TcpNetConnection(Socket socket, boolean server) { + this.socket = socket; + this.server = server; + } + + /** + * Closes this connection. + */ + public void close() { + try { + this.socket.close(); + } catch (Exception e) {} + super.close(); + } + + public boolean isOpen() { + return !this.socket.isClosed(); + } + + @SuppressWarnings("unchecked") + public void send(Message message) throws Exception { + Object object = mapper.fromMessage(message); + this.outputConverter.convert(object, this.socket.getOutputStream()); + if (logger.isDebugEnabled()) + logger.debug("Message sent " + message); + } + + public String getHostAddress() { + return this.socket.getInetAddress().getHostAddress(); + } + + public String getHostName() { + return this.socket.getInetAddress().getHostName(); + } + + public Object getPayload() throws Exception { + return this.inputConverter.convert(this.socket.getInputStream()); + } + + public int getPort() { + return this.socket.getPort(); + } + + /** + * If there is no listener, and this connection is not for single use, + * this method exits. When there is a listener, the method runs in a + * loop reading input from the connections's stream, data is converted + * to an object using the {@link InputStreamingConverter} and the listener's + * {@link TcpListener#onMessage(Message)} method is called. For single use + * connections with no listener, the socket is closed after its timeout + * expires. If data is received on a single use socket with no listener, + * a warning is logged. + */ + public void run() { + if (this.listener == null && !this.singleUse) { + logger.debug("TcpListener exiting - no listener and not single use"); + return; + } + Message message = null; + boolean okToRun = true; + logger.debug("Reading..."); + while (okToRun) { + try { + message = this.mapper.toMessage(this); + } catch (Exception e) { + this.close(); + if (!(e instanceof SoftEndOfStreamException)) { + if (e instanceof SocketTimeoutException && this.singleUse) { + logger.debug("Closing single use socket after timeout"); + } else { + logger.error("Read exception " + + this.getConnectionId() + " " + + e.getClass().getSimpleName() + + ":" + e.getCause() + ":" + e.getMessage()); + } + } + break; + } + /* + * For single use sockets, we close after receipt if we are on the client + * side, or the server side has no outbound adapter registered + */ + if (this.singleUse && (!this.server || this.sender == null)) { + logger.debug("Closing single use socket after inbound message"); + this.close(); + okToRun = false; + } + if (logger.isDebugEnabled()) + logger.debug("Message received " + message); + try { + if (listener == null) { + logger.warn("Unexpected message - no inbound adapter registered with connection " + message); + continue; + } + listener.onMessage(message); + } catch (Exception e) { + logger.error("Exception sending meeeage: " + message, e); + } + } + } + + public String getConnectionId() { + return SocketIoUtils.getSocketId(this.socket); + } + + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java new file mode 100644 index 0000000000..001edfb9ca --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java @@ -0,0 +1,100 @@ +/* + * Copyright 2002-2010 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.tcp.connection; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.ServerSocket; +import java.net.Socket; + +import javax.net.ServerSocketFactory; + +/** + * Implements a server connection factory that produces {@link TcpNetConnection}s using + * a {@link ServerSocket}. Must have a {@link TcpListener} registered. + * @author Gary Russell + * @since 2.0 + * + */ +public class TcpNetServerConnectionFactory extends AbstractServerConnectionFactory { + + protected ServerSocket serverSocket; + + /** + * Listens for incoming connections on the port. + * @param port The port. + */ + public TcpNetServerConnectionFactory(int port) { + super(port); + } + + /** + * If no listener registers, exits. + * Accepts incoming connections and creates TcpConnections for each new connection. + * Invokes {{@link #initializeConnection(TcpConnection, Socket)} and executes the + * connection {@link TcpConnection#run()} using the task executor. + * I/O errors on the server socket/channel are logged and the factory is stopped. + */ + public void run() { + if (this.listener == null) { + logger.info("No listener bound to server connection factory; will not read; exiting..."); + return; + } + try { + if (this.localAddress == null) { + this.serverSocket = ServerSocketFactory.getDefault() + .createServerSocket(this.port, Math.abs(this.poolSize)); + } else { + InetAddress whichNic = InetAddress.getByName(this.localAddress); + this.serverSocket = ServerSocketFactory.getDefault() + .createServerSocket(port, Math.abs(poolSize), whichNic); + } + this.listening = true; + logger.info("Listening on port " + this.port); + while (true) { + final Socket socket = serverSocket.accept(); + logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress()); + setSocketAttributes(socket); + TcpNetConnection connection = new TcpNetConnection(socket, true); + this.initializeConnection(connection, socket); + this.taskExecutor.execute(connection); + } + } catch (IOException e) { + this.listening = false; + if (this.active) { + logger.error("Error on ServerSocket", e); + } + this.active = false; + } + } + + public boolean isRunning() { + return this.active; + } + + public void close() { + if (this.serverSocket == null) { + return; + } + try { + this.serverSocket.close(); + } catch (IOException e) {} + this.serverSocket = null; + } + + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java new file mode 100644 index 0000000000..7d7b2e10f9 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java @@ -0,0 +1,194 @@ +/* + * Copyright 2002-2010 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.tcp.connection; + +import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.nio.channels.SelectionKey; +import java.nio.channels.Selector; +import java.nio.channels.SocketChannel; +import java.util.Iterator; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; + + +/** + * A client connection factory that creates {@link TcpNioConnection}s. + * @author Gary Russell + * @since 2.0 + * + */ +public class TcpNioClientConnectionFactory extends + AbstractClientConnectionFactory { + + protected TcpNioConnection theConnection; + + protected boolean usingDirectBuffers; + + private Selector selector; + + protected Map connections = new ConcurrentHashMap(); + + protected BlockingQueue newChannels = new LinkedBlockingQueue(); + + /** + * Creates a TcpNioClientConnectionFactory for connections to the host and port. + * @param host the host + * @param port the port + */ + public TcpNioClientConnectionFactory(String host, int port) { + super(host, port); + } + + /** + * Obtains a connection - if {@link #setSingleUse(boolean)} was called with + * true, a new connection is returned; otherwise a single connection is + * reused for all requests while the connection remains open. + */ + public TcpNioConnection getConnection() throws Exception { + int n = 0; + while (this.selector == null) { + Thread.sleep(100); + if (n++ > 600) { + throw new Exception("Factory failed to start"); + } + } + if (this.theConnection != null && this.theConnection.isOpen()) { + return this.theConnection; + } + logger.debug("Opening new socket channel connection to " + this.host + ":" + this.port); + SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(this.host, this.port)); + setSocketAttributes(socketChannel.socket()); + TcpNioConnection connection = new TcpNioConnection(socketChannel, false); + if (this.taskExecutor == null) { + connection.setTaskExecutor(Executors.newSingleThreadExecutor()); + } else { + connection.setTaskExecutor(this.taskExecutor); + } + initializeConnection(connection, socketChannel.socket()); + connection.setUsingDirectBuffers(this.usingDirectBuffers); + socketChannel.configureBlocking(false); + if (this.soTimeout > 0) { + connection.setLastRead(System.currentTimeMillis()); + } + this.connections.put(socketChannel, connection); + newChannels.add(socketChannel); + selector.wakeup(); + if (!this.singleUse) { + this.theConnection = connection; + } + return connection; + } + + /** + * When set to true, connections created by this factory attempt + * to use direct buffers where possible. + * @param usingDirectBuffers + * @see ByteBuffer + */ + public void setUsingDirectBuffers(boolean usingDirectBuffers) { + this.usingDirectBuffers = usingDirectBuffers; + } + + public void close() { + } + + public void run() { + logger.debug("Read selector running for connections to " + host + ":" + port); + try { + this.selector = Selector.open(); + while (this.active) { + int selectionCount = selector.select(this.soTimeout); + SocketChannel newChannel; + while ((newChannel = newChannels.poll()) != null) { + newChannel.register(this.selector, SelectionKey.OP_READ, connections.get(newChannel)); + } + if (logger.isTraceEnabled()) + logger.trace("Connection " + host + ":" + port + " SelectionCount: " + selectionCount); + long now = 0; + if (this.soTimeout > 0) { + Iterator it = connections.keySet().iterator(); + now = System.currentTimeMillis(); + while (it.hasNext()) { + SocketChannel channel = it.next(); + if (!channel.isOpen()) { + logger.debug("Removing closed channel"); + it.remove(); + } else { + TcpNioConnection connection = this.connections.get(channel); + if (now - connection.getLastRead() > this.soTimeout) { + logger.warn("Timing out TcpNioConnection " + connection.getConnectionId()); + connection.timeout(); + } + } + } + } + if (selectionCount > 0) { + Set keys = selector.selectedKeys(); + Iterator iterator = keys.iterator(); + while (iterator.hasNext()) { + final SelectionKey key = iterator.next(); + iterator.remove(); + if (!key.isValid()) { + logger.debug("Selection key no longer valid"); + } + else if (key.isReadable()) { + key.interestOps(key.interestOps() - key.readyOps()); + final TcpNioConnection connection; + connection = (TcpNioConnection) key.attachment(); + connection.setLastRead(System.currentTimeMillis()); + this.taskExecutor.execute(new Runnable() { + public void run() { + try { + connection.readPacket(); + } catch (Exception e) { + if (connection.isOpen()) { + logger.error("Exception on read " + + connection.getConnectionId() + " " + + e.getMessage()); + connection.close(); + } else { + logger.debug("Conection closed"); + } + } + if (key.channel().isOpen()) { + key.interestOps(SelectionKey.OP_READ); + selector.wakeup(); + } + }}); + } + else { + logger.error("Unexpected key: " + key); + } + } + } + } + } catch (Exception e) { + logger.error("Exception in reader thread", e); + } + logger.debug("Read selector exiting for connections to " + host + ":" + port); + } + + public boolean isRunning() { + return this.active; + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java new file mode 100644 index 0000000000..399a593971 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java @@ -0,0 +1,353 @@ +/* + * Copyright 2002-2010 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.tcp.connection; + +import java.io.IOException; +import java.io.OutputStream; +import java.io.PipedInputStream; +import java.io.PipedOutputStream; +import java.net.SocketTimeoutException; +import java.nio.ByteBuffer; +import java.nio.channels.SelectionKey; +import java.nio.channels.Selector; +import java.nio.channels.SocketChannel; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; + +import org.springframework.integration.core.Message; +import org.springframework.integration.ip.tcp.SocketIoUtils; +import org.springframework.integration.ip.tcp.converter.SoftEndOfStreamException; + +/** + * A TcpConnection that uses and underlying {@link SocketChannel}. + * + * @author Gary Russell + * @since 2.0 + * + */ +public class TcpNioConnection extends AbstractTcpConnection { + + private final SocketChannel socketChannel; + + private final boolean server; + + private OutputStream channelOutputStream; + + private PipedOutputStream pipedOutputStream; + + private PipedInputStream pipedInputStream; + + private boolean usingDirectBuffers; + + private Executor taskExecutor; + + private ByteBuffer rawBuffer; + + private int maxMessageSize = 60 * 1024; + + private boolean active = true; + + private long lastRead; + + /** + * Constructs a TcpNetConnection for the SocketChannel. + * @param socketChannel the socketChannel + * @param server if true this connection was created as + * a result of an incoming request. + */ + public TcpNioConnection(SocketChannel socketChannel, boolean server) throws Exception { + this.socketChannel = socketChannel; + this.server = server; + this.pipedInputStream = new PipedInputStream(); + this.pipedOutputStream = new PipedOutputStream(this.pipedInputStream); + this.channelOutputStream = new ChannelOutputStream(); + } + + public void close() { + doClose(); + } + + private void doClose() { + this.active = false; + if (pipedOutputStream != null) { + try { + pipedOutputStream.close(); + } catch (IOException e) {} + } + try { + this.socketChannel.close(); + } catch (Exception e) {} + super.close(); + } + + public boolean isOpen() { + return this.socketChannel.isOpen(); + } + + @SuppressWarnings("unchecked") + public void send(Message message) throws Exception { + Object object = mapper.fromMessage(message); + this.outputConverter.convert(object, this.channelOutputStream); + } + + public String getHostAddress() { + return this.socketChannel.socket().getInetAddress().getHostAddress(); + } + + public String getHostName() { + return this.socketChannel.socket().getInetAddress().getHostName(); + } + + public Object getPayload() throws Exception { + return this.inputConverter.convert(pipedInputStream); + } + + public int getPort() { + return this.socketChannel.socket().getPort(); + } + + /** + * Allocates a ByteBuffer of the requested length using normal or + * direct buffers, depending on the usingDirectBuffers field. + */ + protected ByteBuffer allocate(int length) { + ByteBuffer buffer; + if (this.usingDirectBuffers) { + buffer = ByteBuffer.allocateDirect(length); + } else { + buffer = ByteBuffer.allocate(length); + } + return buffer; + } + + /** + * If there is no listener, and this connection is not for single use, + * this method exits. When there is a listener, this method assembles + * data into messages by invoking convertAndSend whenever there is + * data in the input Stream. Method exits when a message is complete + * and there is no more data; thus freeing the thread to work on other + * sockets. + */ + public void run() { + logger.debug("Nio message assembler running..."); + try { + if (this.listener == null && !this.singleUse) { + logger.debug("TcpListener exiting - no listener and not single use"); + return; + } + if (active) { + try { + while (pipedInputStream.available() > 0) { + convertAndSend(); + } + } catch (IOException e) { + logger.error("Unexpected exception, exiting...", e); + return; + } + } + } finally { + logger.debug("Nio message assembler exiting..."); + } + } + + private synchronized void convertAndSend() throws IOException { + if (this.pipedInputStream.available() <= 0) { + System.err.println("NO WORK"); + return; + } + Message message = null; + try { + message = this.mapper.toMessage(this); + } catch (Exception e) { + this.close(); + if (e instanceof SocketTimeoutException && this.singleUse) { + logger.debug("Closing single use socket after timeout"); + } else { + if (!(e instanceof SoftEndOfStreamException)) { + logger.error("Read exception " + + this.getConnectionId() + " " + + e.getClass().getSimpleName() + + ":" + e.getCause() + ":" + e.getMessage()); + + } + } + return; + } + /* + * For single use sockets, we close after receipt if we are on the client + * side, or the server side has no outbound adapter registered + */ + if (this.singleUse && (!this.server || this.sender == null)) { + logger.debug("Closing single use socket after inbound message"); + this.close(); + } + try { + if (message != null) { + listener.onMessage(message); + } + } catch (Exception e) { + logger.error("Exception sending meeeage: " + message, e); + } + } + + private void doRead() throws Exception { + if (rawBuffer == null) { + rawBuffer = allocate(maxMessageSize); + } + rawBuffer.clear(); + int len = socketChannel.read(rawBuffer); + if (len < 0) { + this.close(); + throw new IOException("Channel closed"); + } + rawBuffer.flip(); + if (logger.isDebugEnabled()) { + logger.debug("Read " + rawBuffer.limit() + " into raw buffer"); + } + pipedOutputStream.write(rawBuffer.array(), 0, rawBuffer.limit()); + pipedOutputStream.flush(); + + if (!socketChannel.isBlocking()) { + if (this.taskExecutor == null) { + this.taskExecutor = Executors.newSingleThreadExecutor(); + } + this.taskExecutor.execute(this); + } + } + + /** + * Invoked by the factory when there is data to be read. + */ + public void readPacket() { + try { + doRead(); + } catch (Exception e) { + logger.error("Exception on Read " + + this.getConnectionId() + " " + + e.getMessage()); + this.close(); + } + } + + /** + * Close the socket due to timeout. + */ + void timeout() { + this.close(); + } + + /** + * + * @param taskExecutor the taskExecutor to set + */ + public void setTaskExecutor(Executor taskExecutor) { + this.taskExecutor = taskExecutor; + } + + /** + * If true, connection will attempt to use direct buffers where + * possible. + * @param usingDirectBuffers + */ + public void setUsingDirectBuffers(boolean usingDirectBuffers) { + this.usingDirectBuffers = usingDirectBuffers; + } + + public String getConnectionId() { + return SocketIoUtils.getSocketId(this.socketChannel.socket()); + } + + /** + * + * @return Time of last read. + */ + public long getLastRead() { + return lastRead; + } + + /** + * + * @param lastRead The time of the last read. + */ + public void setLastRead(long lastRead) { + this.lastRead = lastRead; + } + + /** + * OutputStream to wrap a SocketChannel; implements timeout on write. + * + */ + class ChannelOutputStream extends OutputStream { + + private Selector selector; + + private int soTimeout; + + @Override + public void write(int b) throws IOException { + byte[] bytes = new byte[1]; + bytes[0] = (byte) b; + ByteBuffer buffer = ByteBuffer.wrap(bytes); + doWrite(buffer); + } + + @Override + public void close() throws IOException { + doClose(); + } + + @Override + public void flush() throws IOException { + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + ByteBuffer buffer = ByteBuffer.wrap(b, off, len); + doWrite(buffer); + } + + @Override + public void write(byte[] b) throws IOException { + ByteBuffer buffer = ByteBuffer.wrap(b); + doWrite(buffer); + } + + private void doWrite(ByteBuffer buffer) throws IOException { + socketChannel.write(buffer); + int remaining = buffer.remaining(); + if (remaining == 0) { + return; + } + if (this.selector == null) { + this.selector = Selector.open(); + this.soTimeout = socketChannel.socket().getSoTimeout(); + } + socketChannel.register(selector, SelectionKey.OP_WRITE); + while (remaining > 0) { + int selectionCount = this.selector.select(this.soTimeout); + if (selectionCount == 0) { + throw new SocketTimeoutException("Timeout on write"); + } + selector.selectedKeys().clear(); + socketChannel.write(buffer); + remaining = buffer.remaining(); + } + } + + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java new file mode 100644 index 0000000000..1bc75f32c6 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java @@ -0,0 +1,218 @@ +/* + * Copyright 2002-2010 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.tcp.connection; + +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.Socket; +import java.net.SocketException; +import java.nio.channels.ClosedChannelException; +import java.nio.channels.SelectionKey; +import java.nio.channels.Selector; +import java.nio.channels.ServerSocketChannel; +import java.nio.channels.SocketChannel; +import java.util.HashMap; +import java.util.Iterator; +import java.util.Map; +import java.util.Set; + +/** +/** + * Implements a server connection factory that produces {@link TcpNioConnection}s using + * a {@link ServerSocketChannel}. Must have a {@link TcpListener} registered. + * @author Gary Russell + * @since 2.0 + * + */ +public class TcpNioServerConnectionFactory extends AbstractServerConnectionFactory { + + protected ServerSocketChannel serverChannel; + + protected boolean usingDirectBuffers; + + protected Map connections = new HashMap(); + + /** + * Listens for incoming connections on the port. + * @param port The port. + */ + public TcpNioServerConnectionFactory(int port) { + super(port); + } + + /** + * If no listener registers, exits. + * Accepts incoming connections and creates TcpConnections for each new connection. + * Invokes {{@link #initializeConnection(TcpConnection, Socket)} and executes the + * connection {@link TcpConnection#run()} using the task executor. + * I/O errors on the server socket/channel are logged and the factory is stopped. + */ + public void run() { + if (this.listener == null) { + logger.info("No listener bound to server connection factory; will not read; exiting..."); + return; + } + try { + this.serverChannel = ServerSocketChannel.open(); + this.listening = true; + this.serverChannel.configureBlocking(false); + if (this.localAddress == null) { + this.serverChannel.socket().bind(new InetSocketAddress(this.port), + Math.abs(this.poolSize)); + } else { + InetAddress whichNic = InetAddress.getByName(this.localAddress); + this.serverChannel.socket().bind(new InetSocketAddress(whichNic, this.port), + Math.abs(this.poolSize)); + } + final Selector selector = Selector.open(); + this.serverChannel.register(selector, SelectionKey.OP_ACCEPT); + doSelect(this.serverChannel, selector); + + } catch (IOException e) { + this.close(); + this.listening = false; + if (this.active) { + logger.error("Error on ServerSocketChannel", e); + this.active = false; + } + } + } + + /** + * Listens for incoming connections and for notifications that a connected + * socket is ready for reading. + * Accepts incoming connections, registers the new socket with the + * selector for reading. + * When a socket is ready for reading, unregisters the read interest and + * schedules a call to doRead which reads all available data. When the read + * is complete, the socket is again registered for read interest. + * @param server + * @param selector + * @throws IOException + * @throws ClosedChannelException + * @throws SocketException + */ + private void doSelect(ServerSocketChannel server, final Selector selector) + throws IOException, ClosedChannelException, SocketException { + while (this.active) { + int selectionCount = selector.select(this.soTimeout); + if (logger.isTraceEnabled()) + logger.trace("Port " + port + " SelectionCount: " + selectionCount); + long now = 0; + if (this.soTimeout > 0) { + Iterator it = connections.keySet().iterator(); + now = System.currentTimeMillis(); + while (it.hasNext()) { + SocketChannel channel = it.next(); + if (!channel.isOpen()) { + logger.debug("Removing closed channel"); + it.remove(); + } else { + TcpNioConnection connection = this.connections.get(channel); + if (now - connection.getLastRead() > this.soTimeout) { + logger.warn("Timing out TcpNioConnection " + + this.port + " : " + + connection.getConnectionId()); + connection.timeout(); + } + } + } + } + if (selectionCount > 0) { + Set keys = selector.selectedKeys(); + Iterator iterator = keys.iterator(); + SocketChannel channel = null; + while (iterator.hasNext()) { + final SelectionKey key = iterator.next(); + iterator.remove(); + if (!key.isValid()) { + logger.debug("Selection key no longer valid"); + } + else if (key.isAcceptable()) { + channel = server.accept(); + channel.configureBlocking(false); + Socket socket = channel.socket(); + setSocketAttributes(socket); + TcpNioConnection connection = createTcpNioConnection(channel); + connection.setLastRead(now); + connections.put(channel, connection); + channel.register(selector, SelectionKey.OP_READ, connection); + } + else if (key.isReadable()) { + key.interestOps(key.interestOps() - key.readyOps()); + final TcpNioConnection connection; + connection = (TcpNioConnection) key.attachment(); + connection.setLastRead(System.currentTimeMillis()); + this.taskExecutor.execute(new Runnable() { + public void run() { + try { + connection.readPacket(); + } catch (Exception e) { + if (connection.isOpen()) { + logger.error("Exception on read " + e.getMessage()); + connection.close(); + } else { + logger.debug("Conection closed"); + } + } + if (key.channel().isOpen()) { + key.interestOps(SelectionKey.OP_READ); + selector.wakeup(); + } + }}); + } + else { + logger.error("Unexpected key: " + key); + } + } + } + } + } + + private TcpNioConnection createTcpNioConnection(SocketChannel socketChannel) { + try { + TcpNioConnection connection = new TcpNioConnection(socketChannel, true); + this.initializeConnection(connection, socketChannel.socket()); + connection.setUsingDirectBuffers(this.usingDirectBuffers); + return connection; + } catch (Exception e) { + logger.error("Failed to establish new incoming connection", e); + return null; + } + } + + public boolean isRunning() { + return this.active; + } + + public void close() { + if (this.serverChannel == null) { + return; + } + try { + this.serverChannel.close(); + } catch (IOException e) {} + this.serverChannel = null; + } + + public void setUsingDirectBuffers(boolean usingDirectBuffers) { + this.usingDirectBuffers = usingDirectBuffers; + } + + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSender.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSender.java new file mode 100644 index 0000000000..9476e4430e --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpSender.java @@ -0,0 +1,42 @@ +/* + * Copyright 2002-2010 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.tcp.connection; + +/** + * An interface representing a sending client of a connection + * factory. + * @author Gary Russell + * @since 2.0 + * + */ +public interface TcpSender { + + /** + * When we are using sockets owned by a {@link TcpListener}, this + * method is called each time a new connection is made. + * @param connection The connection. + */ + void addNewConnection(TcpConnection connection); + + /** + * When we are using sockets owned by a {@link TcpListener}, this + * method is called each time a connection is closed. + * @param connection The connection. + */ + void removeDeadConnection(TcpConnection connection); + +} \ No newline at end of file diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/package-info.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/package-info.java new file mode 100644 index 0000000000..4c77c12dca --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/package-info.java @@ -0,0 +1,6 @@ +/** + * All things related to tcp connections - client and + * server factories; listener and sender interfaces. + */ +package org.springframework.integration.ip.tcp.connection; + diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/AbstractByteArrayStreamingConverter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/AbstractByteArrayStreamingConverter.java new file mode 100644 index 0000000000..5ee928e6ad --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/AbstractByteArrayStreamingConverter.java @@ -0,0 +1,66 @@ +/* + * Copyright 2002-2010 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.tcp.converter; + +import java.io.IOException; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.commons.serializer.InputStreamingConverter; +import org.springframework.commons.serializer.OutputStreamingConverter; + +/** + * Base class for streaming converters that convert to/from a byte array. + * + * @author Gary Russell + * @since 2.0 + * + */ +public abstract class AbstractByteArrayStreamingConverter implements + InputStreamingConverter, + OutputStreamingConverter { + + protected int maxMessageSize = 2048; + + protected Log logger = LogFactory.getLog(this.getClass()); + + /** + * The maximum supported message size for this converter. + * Default 2048. + * @return The max message size. + */ + public int getMaxMessageSize() { + return maxMessageSize; + } + + /** + * The maximum supported message size for this converter. + * Default 2048. + * @param maxMessageSize The max message size. + */ + public void setMaxMessageSize(int maxMessageSize) { + this.maxMessageSize = maxMessageSize; + } + + protected void checkClosure(int bite) throws IOException { + if (bite < 0) { + logger.debug("Socket closed"); + throw new IOException("Socket closed"); + } + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayCrLfConverter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayCrLfConverter.java new file mode 100644 index 0000000000..ca89503e33 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayCrLfConverter.java @@ -0,0 +1,76 @@ +/* + * Copyright 2002-2010 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.tcp.converter; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +/** + * Converts data in an InputStream to a byte[]; data is terminated by \r\n + * (not included in resulting byte[]). + * Writes a byte[] to an OutputStream and adds \r\n. + * + * @author Gary Russell + * @since 2.0 + * + */ +public class ByteArrayCrLfConverter extends AbstractByteArrayStreamingConverter { + + /** + * Converts the data in the inputstream to a byte[]. Data must be terminated + * by CRLF (\r\n). Throws a {@link SoftEndOfStreamException} if the stream + * is closed immediately after the \r\n (i.e. no data is in the process of + * being read). + */ + public byte[] convert(InputStream inputStream) throws IOException { + byte[] buffer = new byte[this.maxMessageSize]; + int n = 0; + int bite; + if (logger.isDebugEnabled()) + logger.debug("Available to read:" + inputStream.available()); + while (true) { + bite = inputStream.read(); +// logger.debug("Read:" + (char) bite); + if (bite < 0 && n == 0) { + throw new SoftEndOfStreamException("Stream closed between payloads"); + } + checkClosure(bite); + if (n > 0 && bite == '\n' && buffer[n-1] == '\r') + break; + buffer[n++] = (byte) bite; + if (n >= this.maxMessageSize) { + throw new IOException("CRLF not found before max message length: " + + this.maxMessageSize); + } + }; + byte[] assembledData = new byte[n-1]; + System.arraycopy(buffer, 0, assembledData, 0, n-1); + return assembledData; + } + + /** + * Writes the byte[] to the stream and appends \r\n. + */ + public void convert(byte[] bytes, OutputStream outputStream) throws IOException { + outputStream.write(bytes); + outputStream.write('\r'); + outputStream.write('\n'); + outputStream.flush(); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayLengthHeaderConverter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayLengthHeaderConverter.java new file mode 100644 index 0000000000..4bcd39b255 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayLengthHeaderConverter.java @@ -0,0 +1,113 @@ +/* + * Copyright 2002-2010 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.tcp.converter; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.ByteBuffer; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +/** + * Converts data in an InputStream to a byte[]; data is preceded by + * a 4 byte binary length (network byte order, + * not included in resulting byte[]). + * Writes a byte[] to an OutputStream after a 4 byte binary length.\ + * The length field contains the length of data following the length + * field. + * (network byte order). + * @author Gary Russell + * + */ +public class ByteArrayLengthHeaderConverter extends AbstractByteArrayStreamingConverter { + + private Log logger = LogFactory.getLog(this.getClass()); + + /** + * Reads a 4 byte length from the stream and then reads that length + * from the stream and returns the data in a byte[]. Throws an + * IOException if the length field exceeds the maxMessageSize. + * Throws a {@link SoftEndOfStreamException} if the stream + * is closed between messages. + */ + public byte[] convert(InputStream inputStream) throws IOException { + byte[] lengthPart = new byte[4]; + int status = read(inputStream, lengthPart, true); + if (status < 0) { + throw new SoftEndOfStreamException("Stream closed between payloads"); + } + int messageLength = ByteBuffer.wrap(lengthPart).getInt(); + if (logger.isDebugEnabled()) { + logger.debug("Message length is " + messageLength); + } + if (messageLength > this.maxMessageSize) { + throw new IOException("Message length " + messageLength + + " exceeds max message length: " + this.maxMessageSize); + } + byte[] messagePart = new byte[messageLength]; + read(inputStream, messagePart, false); + return messagePart; + } + + /** + * Writes the byte[] to the output stream, preceded by a 4 byte + * length in network byte order (big endian). + */ + public void convert(byte[] bytes, OutputStream outputStream) + throws IOException { + ByteBuffer lengthPart = ByteBuffer.allocate(4); + lengthPart.putInt(bytes.length); + outputStream.write(lengthPart.array()); + outputStream.write(bytes); + outputStream.flush(); + } + + /** + * Reads data from the socket and puts the data in buffer. Blocks until + * buffer is full or a socket timeout occurs. + * @param buffer + * @param header true if we are reading the header + * @return < 0 if socket closed and not in the middle of a message + * @throws IOException + */ + protected int read(InputStream inputStream, byte[] buffer, boolean header) + throws IOException { + int lengthRead = 0; + int needed = buffer.length; + while (lengthRead < needed) { + int len; + len = inputStream.read(buffer, lengthRead, + needed - lengthRead); + if (len < 0 && header && lengthRead == 0) { + return len; + } + if (len < 0) { + throw new IOException("Stream closed after " + lengthRead + " of " + needed); + } + lengthRead += len; + if (logger.isDebugEnabled()) { + logger.debug("Read " + len + " bytes, buffer is now at " + + lengthRead + " of " + + needed); + } + } + return 0; + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayStxEtxConverter.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayStxEtxConverter.java new file mode 100644 index 0000000000..2ffe4aff63 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/ByteArrayStxEtxConverter.java @@ -0,0 +1,81 @@ +/* + * Copyright 2002-2010 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.tcp.converter; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + +import org.springframework.integration.message.MessageMappingException; + +/** + * Converts data in an InputStream to a byte[]; data is prefixed by <stx> and + * terminated by <etx> (not included in resulting byte[]). + * Writes a byte[] to an OutputStream and prefixed by <stx> terminated by <etx> + * + * @author Gary Russell + * @since 2.0 + * + */ +public class ByteArrayStxEtxConverter extends AbstractByteArrayStreamingConverter { + + public static final int STX = 0x02; + + public static final int ETX = 0x03; + + /** + * Converts the data in the inputstream to a byte[]. Data must be prefixed + * with an ASCII STX character, and terminated with an ASCII ETX character. + * Throws a {@link SoftEndOfStreamException} if the stream + * is closed immediately before the STX (i.e. no data is in the process of + * being read). + * + */ + public byte[] convert(InputStream inputStream) throws IOException { + int bite = inputStream.read(); + if (bite < 0) { + throw new SoftEndOfStreamException("Stream closed between payloads"); + } + if (bite != STX) + throw new MessageMappingException("Expected STX to begin message"); + byte[] buffer = new byte[this.maxMessageSize]; + int n = 0; + while ((bite = inputStream.read()) != ETX) { + checkClosure(bite); + buffer[n++] = (byte) bite; + if (n >= this.maxMessageSize) { + throw new IOException("ETX not found before max message length: " + + this.maxMessageSize); + } + } + byte[] assembledData = new byte[n]; + System.arraycopy(buffer, 0, assembledData, 0, n); + return assembledData; + } + + /** + * Writes the byte[] to the stream, prefixed by an ASCII STX character and + * terminated with an ASCII ETX character. + */ + public void convert(byte[] bytes, OutputStream outputStream) throws IOException { + outputStream.write(STX); + outputStream.write(bytes); + outputStream.write(ETX); + outputStream.flush(); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/SoftEndOfStreamException.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/SoftEndOfStreamException.java new file mode 100644 index 0000000000..49a0f8976b --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/SoftEndOfStreamException.java @@ -0,0 +1,49 @@ +/* + * Copyright 2002-2010 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.tcp.converter; + +import java.io.IOException; + +/** + * Used to communicate that a stream has closed, but between logical + * messages. + * + * @author Gary Russell + * @since 2.0 + * + */ +public class SoftEndOfStreamException extends IOException { + + private static final long serialVersionUID = 7309907445617226978L; + + public SoftEndOfStreamException() { + super(); + } + + public SoftEndOfStreamException(String message, Throwable cause) { + super(message, cause); + } + + public SoftEndOfStreamException(String message) { + super(message); + } + + public SoftEndOfStreamException(Throwable cause) { + super(cause); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/package-info.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/package-info.java new file mode 100644 index 0000000000..ebcec3651d --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/converter/package-info.java @@ -0,0 +1,6 @@ +/** + * Byte array converters for putting some protocol on the + * wire so we can delimit incoming messages. + */ +package org.springframework.integration.ip.tcp.converter; + diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd index 84cb0f2bd5..7e89fe622d 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd @@ -189,6 +189,183 @@ the custom message format. See java docs for TcpNetSendingChannelAdapter and Tcp + + + + + + + + + + + + +A connection factory is needed by an outbound adapter. If the connection factory has a type 'client', +the factory is 'owned' by this adapter. If it has a type 'server', it is owned by an inbound channel +adapter and this adapter will attempt to correlate messages to the connection on which an original +inbound message was received. + + + + + + + + + + + + + + + + + + + + + + + + + + +A connection factory is needed by an inbound adapter. If the connection factory has a type 'server', +the factory is 'owned' by this adapter. If it has a type 'client', it is owned by an outbound channel +adapter and this adapter will receive any incoming messages on the connection created by the outbound +adapter. + + + + + + + + + + + + + + + + + + + + + +Connection factories can be 'client' or 'server'. Client factories +open a connection to a server using a host and port. Server factories +listen on a port and create a separate connection for each incoming +connection request. + + + + + + + + + + + + +The host to which a client connection factory will connect. + + + + + + +For client factories, the port to which a client connection factory will connect. +For server factories, the port on which the factory will listen for incoming +connections. + + + + + + +If true, the factory will use java.nio.channel.SocketChannel for communication; +for a large number of connections on the server side, this can provide better +performance and may use fewer threads. + + + + + + + + + + + + + +If true, a new connection will be created for each use. For inbound adapters +where there is no outbound adapter sharing the factory, the connection will +be closed after a message is received. For outbound adapters where there is +no inbound adapter sharing the factory, or for inbound adapters where an +outbound adapter shares the factory, the connection will be closed after +so-timeout milliseconds. For outbound adapters where an inbound adapter shares +the factory, the connection will be closed after a response is received. + + + + + + + + + + + +An InputStreamingConverter that converts message payloads to/from output streams/input streams +associated with the connection. Default is ByteArrayCrLfConverter. Input and output converters +would normally be the same but this is not required. + + + + + + + + + + + +An OutputStreamingConverter that converts message payloads to/from output streams/input streams +associated with the connection. Default is ByteArrayCrLfConverter. Input and output converters +would normally be the same but this is not required. + + + + + + +On a multi-homed system, specifies the ip address of the network interface used to communicate. +For inbound adapters and gateways, specifies the interface used to listen for incoming connections. +If omitted, the endpoint will listen on all available adapters. For the UDP multicast outbound adapter +specifies the interface to which multicast packets will be sent. For UDP unicast and multicast +adapters, specifies which interface to which the acknowledgment socket will be bound. Does not +apply to TCP outbound adapters and gateways. + + + + + + + +The number of threads that will be used for socket/channel handling. Only applies +if an external task-executor is NOT being used. When using an external task executor, +its configuration specifies the number of threads. + + + + + 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 80bc8d7916..4572195901 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 @@ -277,5 +277,64 @@ so-timeout="226" close="true" /> + + + + + + + + + + + + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java index 83176408cf..b7dcf846e6 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java @@ -25,6 +25,7 @@ import org.junit.runner.RunWith; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.commons.serializer.InputStreamingConverter; import org.springframework.context.ApplicationContext; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.ip.tcp.CustomNetSocketReader; @@ -38,6 +39,11 @@ import org.springframework.integration.ip.tcp.TcpNetReceivingChannelAdapter; import org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler; import org.springframework.integration.ip.tcp.TcpNioReceivingChannelAdapter; import org.springframework.integration.ip.tcp.TcpNioSendingMessageHandler; +import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter; +import org.springframework.integration.ip.tcp.TcpSendingMessageHandler; +import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory; import org.springframework.integration.ip.udp.DatagramPacketMessageMapper; import org.springframework.integration.ip.udp.MulticastReceivingChannelAdapter; import org.springframework.integration.ip.udp.MulticastSendingMessageHandler; @@ -49,6 +55,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Gary Russell + * @since 2.0 */ @ContextConfiguration @RunWith(SpringJUnit4ClassRunner.class) @@ -125,6 +132,33 @@ public class ParserUnitTests { @Qualifier(value="externalTE") TaskExecutor taskExecutor; + @Autowired + @Qualifier(value="client1") + AbstractConnectionFactory client1; + + @Autowired + InputStreamingConverter converter; + + @Autowired + @Qualifier(value="server1") + AbstractConnectionFactory server1; + + @Autowired + @Qualifier(value="org.springframework.integration.ip.tcp.TcpSendingMessageHandler#0") + TcpSendingMessageHandler tcpNewOut1; + + @Autowired + @Qualifier(value="org.springframework.integration.ip.tcp.TcpSendingMessageHandler#1") + TcpSendingMessageHandler tcpNewOut2; + + @Autowired + @Qualifier(value="tcpNewIn1") + TcpReceivingChannelAdapter tcpNewIn1; + + @Autowired + @Qualifier(value="tcpNewIn2") + TcpReceivingChannelAdapter tcpNewIn2; + @Test public void testInUdp() { DirectFieldAccessor dfa = new DirectFieldAccessor(udpIn); @@ -383,4 +417,66 @@ public class ParserUnitTests { assertEquals(226, delegateDfa.getPropertyValue("soTimeout")); assertEquals(true, dfa.getPropertyValue("close")); } + + @Test + public void testConnClient1() { + assertTrue(client1 instanceof TcpNioClientConnectionFactory); + assertEquals("localhost", client1.getHost()); + assertEquals(9876, client1.getPort()); + assertEquals(54, client1.getSoLinger()); + assertEquals(1234, client1.getSoReceiveBufferSize()); + assertEquals(1235, client1.getSoSendBufferSize()); + assertEquals(1236, client1.getSoTimeout()); + assertEquals(12, client1.getSoTrafficClass()); + DirectFieldAccessor dfa = new DirectFieldAccessor(client1); + assertSame(converter, dfa.getPropertyValue("inputConverter")); + assertSame(converter, dfa.getPropertyValue("outputConverter")); + assertEquals(true, dfa.getPropertyValue("soTcpNoDelay")); + assertEquals(true, dfa.getPropertyValue("singleUse")); + assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor")); + assertEquals(321, dfa.getPropertyValue("poolSize")); + } + + @Test + public void testConnServer1() { + assertTrue(server1 instanceof TcpNioServerConnectionFactory); + assertEquals(9876, server1.getPort()); + assertEquals(55, server1.getSoLinger()); + assertEquals(1234, server1.getSoReceiveBufferSize()); + assertEquals(1235, server1.getSoSendBufferSize()); + assertEquals(1236, server1.getSoTimeout()); + assertEquals(12, server1.getSoTrafficClass()); + DirectFieldAccessor dfa = new DirectFieldAccessor(server1); + assertSame(converter, dfa.getPropertyValue("inputConverter")); + assertSame(converter, dfa.getPropertyValue("outputConverter")); + assertEquals(true, dfa.getPropertyValue("soTcpNoDelay")); + assertEquals(true, dfa.getPropertyValue("singleUse")); + assertSame(taskExecutor, dfa.getPropertyValue("taskExecutor")); + assertEquals(123, dfa.getPropertyValue("poolSize")); + } + + @Test + public void testNewOut1() { + DirectFieldAccessor dfa = new DirectFieldAccessor(tcpNewOut1); + assertSame(client1, dfa.getPropertyValue("clientConnectionFactory")); + } + + @Test + public void testNewOut2() { + DirectFieldAccessor dfa = new DirectFieldAccessor(tcpNewOut2); + assertSame(server1, dfa.getPropertyValue("serverConnectionFactory")); + } + + @Test + public void testNewIn1() { + DirectFieldAccessor dfa = new DirectFieldAccessor(tcpNewIn1); + assertSame(client1, dfa.getPropertyValue("clientConnectionFactory")); + } + + @Test + public void testNewIn2() { + DirectFieldAccessor dfa = new DirectFieldAccessor(tcpNewIn2); + assertSame(server1, dfa.getPropertyValue("serverConnectionFactory")); + } + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml new file mode 100644 index 0000000000..9356b51bb8 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests-context.xml @@ -0,0 +1,26 @@ + + + + + + + + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java new file mode 100644 index 0000000000..7c3d348d24 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/ConnectionToConnectionTests.java @@ -0,0 +1,81 @@ +/* + * Copyright 2002-2010 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.tcp; + +import static org.junit.Assert.assertNotNull; + +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.support.AbstractApplicationContext; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.Message; +import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter; +import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpConnection; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + + +/** + * @author Gary Russell + * @since 2.0 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class ConnectionToConnectionTests { + + @Autowired + AbstractApplicationContext ctx; + + @Autowired + private AbstractClientConnectionFactory client; + + @Autowired + private AbstractServerConnectionFactory server; + + private TcpReceivingChannelAdapter receiver; + + @Before + public void setup() { + receiver = new TcpReceivingChannelAdapter(); + server.registerListener(receiver); + ctx.start(); + } + + @Test + public void testConnect() throws Exception { + int n = 0; + while (!server.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + throw new Exception("Failed to listen"); + } + } + TcpConnection connection = client.getConnection(); + QueueChannel channel = new QueueChannel(); + receiver.setOutputChannel(channel); + connection.send(MessageBuilder.withPayload("Test").build()); + Message m = channel.receive(10000); + assertNotNull(m); + } + +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests-context.xml new file mode 100644 index 0000000000..8d1354c8e2 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests-context.xml @@ -0,0 +1,58 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests.java new file mode 100644 index 0000000000..9e126e417e --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/SharedConnectionTests.java @@ -0,0 +1,80 @@ +/* + * Copyright 2002-2010 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.tcp; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.fail; + +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.support.AbstractApplicationContext; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.ip.tcp.connection.TcpListener; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + + +/** + * @author Gary Russell + * @since 2.0 + * + */ +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class SharedConnectionTests { + + @Autowired + AbstractApplicationContext ctx; + + @Autowired + @Qualifier(value="inboundServer") + TcpListener listener; + + /** + * Tests a loopback. The client-side outbound adapter sends a message over + * a connection from the client connection factory; the server side + * receives the message, puts in on a channel which is the input channel + * for the outbound adapter that's sharing the connections. The response + * comes back to an inbound adapter that is sharing the client's + * connection and we verify we get the echo back as expected. + * + * @throws Exception + */ + @Test + public void test1() throws Exception { + int n = 0; + while (!listener.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + fail("Failed to listen"); + } + } + MessageChannel input = ctx.getBean("input", MessageChannel.class); + input.send(MessageBuilder.withPayload("Test").build()); + QueueChannel replies = ctx.getBean("replies", QueueChannel.class); + Message message = replies.receive(10000); + assertNotNull(message); + assertEquals("Test", message.getPayload()); + } + +} diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java index 2b6630efcf..1828bfa3f5 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpReceivingChannelAdapterTests.java @@ -17,13 +17,31 @@ package org.springframework.integration.ip.tcp; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import java.io.IOException; +import java.io.InputStream; +import java.net.Socket; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.LinkedList; +import java.util.List; +import java.util.Set; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; + +import javax.net.SocketFactory; import org.junit.Test; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; import org.springframework.integration.ip.AbstractInternetProtocolReceivingChannelAdapter; +import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory; +import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter; import org.springframework.integration.ip.util.SocketUtils; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; @@ -212,5 +230,354 @@ public class TcpReceivingChannelAdapterTests { new String((byte[])message.getPayload())); adapter.stop(); } + + @Test + public void newTestNet() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); + scf.setInputConverter(converter); + scf.setOutputConverter(converter); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(scf); + scf.start(); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + fail("Failed to start listening"); + } + } + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + socket.getOutputStream().write("Test1\r\n".getBytes()); + socket.getOutputStream().write("Test2\r\n".getBytes()); + Message message = channel.receive(10000); + assertNotNull(message); + assertEquals("Test1", new String((byte[]) message.getPayload())); + message = channel.receive(10000); + assertNotNull(message); + assertEquals("Test2", new String((byte[]) message.getPayload())); + } + + @Test + public void newTestNio() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); + scf.setInputConverter(converter); + scf.setOutputConverter(converter); + scf.setSoTimeout(5000); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(scf); + scf.start(); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + fail("Failed to start listening"); + } + } + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + for (int i = 0; i < 100; i++) { + socket.getOutputStream().write(("Test" + i + "\r\n").getBytes()); +// if (i % 10 == 0) { +// Thread.sleep(1000); +// } + } + for (int i = 0; i < 100; i++) { + Message message = channel.receive(10000); + assertNotNull(message); + assertEquals("Test" + i, new String((byte[]) message.getPayload())); + } + } + + @Test + public void newTestNetShared() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); + scf.setInputConverter(converter); + scf.setOutputConverter(converter); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(scf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(scf); + scf.start(); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + fail("Failed to listen"); + } + } + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + socket.setSoTimeout(2000); + socket.getOutputStream().write("Test\r\n".getBytes()); + socket.getOutputStream().write("Test\r\n".getBytes()); + Message message = channel.receive(10000); + assertNotNull(message); + handler.handleMessage(message); + message = channel.receive(10000); + assertNotNull(message); + handler.handleMessage(message); + byte[] b = new byte[6]; + readFully(socket.getInputStream(), b); + assertEquals("Test\r\n", new String(b)); + readFully(socket.getInputStream(), b); + assertEquals("Test\r\n", new String(b)); + } + + @Test + public void newTestNioShared() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); + scf.setInputConverter(converter); + scf.setOutputConverter(converter); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(scf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(scf); + scf.start(); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + fail("Failed to listen"); + } + } + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + socket.setSoTimeout(2000); + socket.getOutputStream().write("Test\r\n".getBytes()); + socket.getOutputStream().write("Test\r\n".getBytes()); + Message message = channel.receive(10000); + assertNotNull(message); + handler.handleMessage(message); + message = channel.receive(10000); + assertNotNull(message); + handler.handleMessage(message); + byte[] b = new byte[6]; + readFully(socket.getInputStream(), b); + assertEquals("Test\r\n", new String(b)); + readFully(socket.getInputStream(), b); + assertEquals("Test\r\n", new String(b)); + } + + @Test + public void newTestNetSingleNoOutbound() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); + scf.setInputConverter(converter); + scf.setOutputConverter(converter); + scf.setSingleUse(true); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(scf); + scf.start(); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + fail("Failed to start listening"); + } + } + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + socket.getOutputStream().write("Test1\r\n".getBytes()); + socket = SocketFactory.getDefault().createSocket("localhost", port); + socket.getOutputStream().write("Test2\r\n".getBytes()); + Message message = channel.receive(10000); + assertNotNull(message); + // with single use, results may come back in a different order + Set results = new HashSet(); + results.add(new String((byte[]) message.getPayload())); + message = channel.receive(10000); + assertNotNull(message); + results.add(new String((byte[]) message.getPayload())); + assertTrue(results.contains("Test1")); + assertTrue(results.contains("Test2")); + } + + @Test + public void newTestNioSingleNoOutbound() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); + scf.setInputConverter(converter); + scf.setOutputConverter(converter); + scf.setSingleUse(true); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(scf); + scf.start(); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + fail("Failed to start listening"); + } + } + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + Socket socket = SocketFactory.getDefault().createSocket("localhost", port); + socket.getOutputStream().write("Test1\r\n".getBytes()); + socket = SocketFactory.getDefault().createSocket("localhost", port); + socket.getOutputStream().write("Test2\r\n".getBytes()); + Message message = channel.receive(10000); + assertNotNull(message); + // with single use, results may come back in a different order + Set results = new HashSet(); + results.add(new String((byte[]) message.getPayload())); + message = channel.receive(10000); + assertNotNull(message); + results.add(new String((byte[]) message.getPayload())); + assertTrue(results.contains("Test1")); + assertTrue(results.contains("Test2")); + } + + /** + * @param is + * @param buff + */ + private void readFully(InputStream is, byte[] buff) throws IOException { + for (int i = 0; i < buff.length; i++) { + buff[i] = (byte) is.read(); + } + } + + @Test + public void newTestNetSingleShared() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + AbstractServerConnectionFactory scf = new TcpNetServerConnectionFactory(port); + ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); + scf.setInputConverter(converter); + scf.setOutputConverter(converter); + scf.setSingleUse(true); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(scf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(scf); + scf.start(); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + fail("Failed to listen"); + } + } + Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port); + socket1.setSoTimeout(2000); + socket1.getOutputStream().write("Test1\r\n".getBytes()); + Socket socket2 = SocketFactory.getDefault().createSocket("localhost", port); + socket2.setSoTimeout(2000); + socket2.getOutputStream().write("Test2\r\n".getBytes()); + Message message = channel.receive(10000); + assertNotNull(message); + handler.handleMessage(message); + message = channel.receive(10000); + assertNotNull(message); + handler.handleMessage(message); + byte[] b = new byte[7]; + readFully(socket1.getInputStream(), b); + assertEquals("Test1\r\n", new String(b)); + readFully(socket2.getInputStream(), b); + assertEquals("Test2\r\n", new String(b)); + } + + @Test + public void newTestNioSingleShared() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); + scf.setInputConverter(converter); + scf.setOutputConverter(converter); + scf.setSingleUse(true); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(scf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(scf); + scf.start(); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + fail("Failed to listen"); + } + } + Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port); + socket1.setSoTimeout(2000); + socket1.getOutputStream().write("Test1\r\n".getBytes()); + Socket socket2 = SocketFactory.getDefault().createSocket("localhost", port); + socket2.setSoTimeout(2000); + socket2.getOutputStream().write("Test2\r\n".getBytes()); + Message message = channel.receive(10000); + assertNotNull(message); + handler.handleMessage(message); + message = channel.receive(10000); + assertNotNull(message); + handler.handleMessage(message); + byte[] b = new byte[7]; + readFully(socket1.getInputStream(), b); + assertEquals("Test1\r\n", new String(b)); + readFully(socket2.getInputStream(), b); + assertEquals("Test2\r\n", new String(b)); + } + + @Test + public void newTestNioSingleSharedMany() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + TcpNioServerConnectionFactory scf = new TcpNioServerConnectionFactory(port); + ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); + scf.setInputConverter(converter); + scf.setOutputConverter(converter); + scf.setSingleUse(true); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(scf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(scf); + Executor te = Executors.newFixedThreadPool(100); + scf.setTaskExecutor(te); + scf.start(); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (n++ > 100) { + fail("Failed to listen"); + } + } + List sockets = new LinkedList(); + for (int i = 100; i < 200; i++) { + Socket socket1 = SocketFactory.getDefault().createSocket("localhost", port); + socket1.setSoTimeout(2000); + socket1.getOutputStream().write(("Test" + i + "\r\n").getBytes()); + sockets.add(socket1); + } + for (int i = 100; i < 200; i++) { + Message message = channel.receive(10000); + assertNotNull(message); + handler.handleMessage(message); + } + byte[] b = new byte[9]; + for (int i = 100; i < 200; i++) { + readFully(sockets.remove(0).getInputStream(), b); + assertEquals("Test" + i + "\r\n", new String(b)); + } + } + + } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java index b4d5ad80b7..f3e65093e0 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java @@ -16,22 +16,42 @@ package org.springframework.integration.ip.tcp; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; import java.io.IOException; import java.io.InputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; import java.net.ServerSocket; import java.net.Socket; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import javax.net.ServerSocketFactory; import org.junit.Test; +import org.springframework.commons.serializer.JavaSerializationConverter; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; +import org.springframework.integration.ip.tcp.connection.AbstractConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory; +import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory; +import org.springframework.integration.ip.tcp.converter.ByteArrayCrLfConverter; +import org.springframework.integration.ip.tcp.converter.ByteArrayLengthHeaderConverter; +import org.springframework.integration.ip.tcp.converter.ByteArrayStxEtxConverter; import org.springframework.integration.ip.util.SocketUtils; import org.springframework.integration.message.MessageBuilder; /** * @author Gary Russell + * @since 2.0 * */ public class TcpSendingMessageHandlerTests { @@ -197,13 +217,655 @@ public class TcpSendingMessageHandlerTests { server.close(); } - /** - * @param is - * @param buff - */ private void readFully(InputStream is, byte[] buff) throws IOException { for (int i = 0; i < buff.length; i++) { buff[i] = (byte) is.read(); } } + + @Test + public void newTestNetCrLf() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + Socket socket = server.accept(); + int i = 0; + while (true) { + byte[] b = new byte[6]; + readFully(socket.getInputStream(), b); + b = ("Reply" + (++i) + "\r\n").getBytes(); + socket.getOutputStream().write(b); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(ccf); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + Message mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply1", new String((byte[]) mOut.getPayload())); + mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply2", new String((byte[]) mOut.getPayload())); + done.set(true); + } + + @Test + public void newTestNio() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + Socket socket = server.accept(); + int i = 0; + while (true) { + byte[] b = new byte[6]; + readFully(socket.getInputStream(), b); + b = ("Reply" + (++i) + "\r\n").getBytes(); + socket.getOutputStream().write(b); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); +// ccf.setSoTimeout(10000); + ccf.start(); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(ccf); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + Message mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply1", new String((byte[]) mOut.getPayload())); + mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply2", new String((byte[]) mOut.getPayload())); + done.set(true); + } + + @Test + public void newTestNetStxEtx() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + Socket socket = server.accept(); + int i = 0; + while (true) { + byte[] b = new byte[6]; + readFully(socket.getInputStream(), b); + b = ("\u0002Reply" + (++i) + "\u0003").getBytes(); + socket.getOutputStream().write(b); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + ccf.start(); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(ccf); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + Message mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply1", new String((byte[]) mOut.getPayload())); + mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply2", new String((byte[]) mOut.getPayload())); + done.set(true); + } + + @Test + public void newTestNioStxEtx() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + Socket socket = server.accept(); + int i = 0; + while (true) { + byte[] b = new byte[6]; + readFully(socket.getInputStream(), b); + b = ("\u0002Reply" + (++i) + "\u0003").getBytes(); + socket.getOutputStream().write(b); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + ByteArrayStxEtxConverter converter = new ByteArrayStxEtxConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + ccf.start(); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(ccf); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + Message mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply1", new String((byte[]) mOut.getPayload())); + mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply2", new String((byte[]) mOut.getPayload())); + done.set(true); + } + + @Test + public void newTestNetLength() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + Socket socket = server.accept(); + int i = 0; + while (true) { + byte[] b = new byte[8]; + readFully(socket.getInputStream(), b); + if (!"\u0000\u0000\u0000\u0004Test".equals(new String(b))) { + throw new RuntimeException("Bad Data"); + } + b = ("\u0000\u0000\u0000\u0006Reply" + (++i)).getBytes(); + socket.getOutputStream().write(b); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + ccf.start(); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(ccf); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + Message mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply1", new String((byte[]) mOut.getPayload())); + mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply2", new String((byte[]) mOut.getPayload())); + done.set(true); + } + + @Test + public void newTestNioLength() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + Socket socket = server.accept(); + int i = 0; + while (true) { + byte[] b = new byte[8]; + readFully(socket.getInputStream(), b); + if (!"\u0000\u0000\u0000\u0004Test".equals(new String(b))) { + throw new RuntimeException("Bad Data"); + } + b = ("\u0000\u0000\u0000\u0006Reply" + (++i)).getBytes(); + socket.getOutputStream().write(b); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + ByteArrayLengthHeaderConverter converter = new ByteArrayLengthHeaderConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + ccf.start(); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(ccf); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + Message mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply1", new String((byte[]) mOut.getPayload())); + mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply2", new String((byte[]) mOut.getPayload())); + done.set(true); + } + + @Test + public void newTestNetSerial() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + Socket socket = server.accept(); + int i = 0; + while (true) { + ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); + ois.readObject(); + ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); + oos.writeObject("Reply" + (++i)); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + JavaSerializationConverter converter = new JavaSerializationConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + ccf.start(); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(ccf); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + Message mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply1", mOut.getPayload()); + mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply2", mOut.getPayload()); + done.set(true); + } + + @Test + public void newTestNioSerial() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + Socket socket = server.accept(); + int i = 0; + while (true) { + ObjectInputStream ois = new ObjectInputStream(socket.getInputStream()); + ois.readObject(); + ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream()); + oos.writeObject("Reply" + (++i)); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + JavaSerializationConverter converter = new JavaSerializationConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + ccf.start(); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(ccf); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + Message mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply1", mOut.getPayload()); + mOut = channel.receive(10000); + assertNotNull(mOut); + assertEquals("Reply2", mOut.getPayload()); + done.set(true); + } + + @Test + public void newTestNetSingleUseNoInbound() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final Semaphore semaphore = new Semaphore(0); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + while (true) { + Socket socket = server.accept(); + semaphore.release(); + byte[] b = new byte[6]; + readFully(socket.getInputStream(), b); + semaphore.release(); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + ccf.start(); + ccf.setSingleUse(true); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + assertTrue(semaphore.tryAcquire(4, 10000, TimeUnit.MILLISECONDS)); + done.set(true); + } + + @Test + public void newTestNioSingleUseNoInbound() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final Semaphore semaphore = new Semaphore(0); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + while (true) { + Socket socket = server.accept(); + semaphore.release(); + byte[] b = new byte[6]; + readFully(socket.getInputStream(), b); + semaphore.release(); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + ccf.start(); + ccf.setSingleUse(true); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + assertTrue(semaphore.tryAcquire(4, 10000, TimeUnit.MILLISECONDS)); + done.set(true); + } + + @Test + public void newTestNetSingleUseWithInbound() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final Semaphore semaphore = new Semaphore(0); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + int i = 0; + while (true) { + Socket socket = server.accept(); + semaphore.release(); + byte[] b = new byte[6]; + readFully(socket.getInputStream(), b); + b = ("Reply" + (++i) + "\r\n").getBytes(); + socket.getOutputStream().write(b); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNetClientConnectionFactory("localhost", port); + ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + ccf.start(); + ccf.setSingleUse(true); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(ccf); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + assertTrue(semaphore.tryAcquire(2, 10000, TimeUnit.MILLISECONDS)); + Set replies = new HashSet(); + for (int i = 0; i < 2; i++) { + Message mOut = channel.receive(10000); + assertNotNull(mOut); + replies.add(new String((byte[])mOut.getPayload())); + } + assertTrue(replies.remove("Reply1")); + assertTrue(replies.remove("Reply2")); + done.set(true); + } + + @Test + public void newTestNioSingleUseWithInbound() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final Semaphore semaphore = new Semaphore(0); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + int i = 0; + while (true) { + Socket socket = server.accept(); + semaphore.release(); + byte[] b = new byte[6]; + readFully(socket.getInputStream(), b); + b = ("Reply" + (++i) + "\r\n").getBytes(); + socket.getOutputStream().write(b); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + ccf.start(); + ccf.setSingleUse(true); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(ccf); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + handler.handleMessage(MessageBuilder.withPayload("Test").build()); + assertTrue(semaphore.tryAcquire(2, 10000, TimeUnit.MILLISECONDS)); + Set replies = new HashSet(); + for (int i = 0; i < 2; i++) { + Message mOut = channel.receive(10000); + assertNotNull(mOut); + replies.add(new String((byte[])mOut.getPayload())); + } + assertTrue(replies.remove("Reply1")); + assertTrue(replies.remove("Reply2")); + done.set(true); + } + + @Test + public void newTestNioSingleUseWithInboundMany() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + final CountDownLatch latch = new CountDownLatch(1); + final Semaphore semaphore = new Semaphore(0); + final AtomicBoolean done = new AtomicBoolean(); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + int i = 0; + while (true) { + Socket socket = server.accept(); + semaphore.release(); + byte[] b = new byte[8]; + readFully(socket.getInputStream(), b); + b = ("Reply" + (i++) + "\r\n").getBytes(); + socket.getOutputStream().write(b); + } + } catch (Exception e) { + if (!done.get()) { + e.printStackTrace(); + } + } + } + }); + AbstractConnectionFactory ccf = new TcpNioClientConnectionFactory("localhost", port); + ByteArrayCrLfConverter converter = new ByteArrayCrLfConverter(); + ccf.setInputConverter(converter); + ccf.setOutputConverter(converter); + ccf.setSoTimeout(10000); + ccf.start(); + ccf.setSingleUse(true); + TcpSendingMessageHandler handler = new TcpSendingMessageHandler(); + handler.setConnectionFactory(ccf); + TcpReceivingChannelAdapter adapter = new TcpReceivingChannelAdapter(); + adapter.setConnectionFactory(ccf); + QueueChannel channel = new QueueChannel(); + adapter.setOutputChannel(channel); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + for (int i = 100; i < 200; i++) { + handler.handleMessage(MessageBuilder.withPayload("Test" + i).build()); + } + assertTrue(semaphore.tryAcquire(10, 10000, TimeUnit.MILLISECONDS)); + Set replies = new HashSet(); + for (int i = 100; i < 200; i++) { + Message mOut = channel.receive(10000); + assertNotNull(mOut); + replies.add(new String((byte[])mOut.getPayload())); + } + for (int i = 0; i < 100; i++) { + assertTrue("Reply" + i + " missing", replies.remove("Reply" + i)); + } + done.set(true); + } } diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java new file mode 100644 index 0000000000..9831b12200 --- /dev/null +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionTests.java @@ -0,0 +1,121 @@ +/* + * Copyright 2002-2010 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.tcp.connection; + +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.io.IOException; +import java.io.InputStream; +import java.net.ServerSocket; +import java.net.Socket; +import java.net.SocketTimeoutException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import javax.net.ServerSocketFactory; + +import org.junit.Test; +import org.springframework.integration.core.Message; +import org.springframework.integration.ip.util.SocketUtils; +import org.springframework.integration.message.MessageBuilder; + + +/** + * @author Gary Russell + * @since 2.0 + * + */ +public class TcpNioConnectionTests { + + @Test + public void testWriteTimeout() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", port); + factory.setSoTimeout(1000); + factory.start(); + final CountDownLatch latch = new CountDownLatch(1); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + server.accept(); + // block so we fill the buffer + server.accept(); + } catch (Exception e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + } + }); + assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); + try { + TcpNioConnection connection = factory.getConnection(); + connection.send(MessageBuilder.withPayload(new byte[1000000]).build()); + } catch (Exception e) { + assertTrue("Expected SocketTimeoutException, got " + e.getClass().getSimpleName() + + ":" + e.getMessage(), e instanceof SocketTimeoutException); + } + } + + @Test + public void testReadTimeout() throws Exception { + final int port = SocketUtils.findAvailableServerSocket(); + TcpNioClientConnectionFactory factory = new TcpNioClientConnectionFactory("localhost", port); + factory.setSoTimeout(1000); + factory.start(); + final CountDownLatch latch = new CountDownLatch(1); + Executors.newSingleThreadExecutor().execute(new Runnable() { + public void run() { + try { + ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port); + latch.countDown(); + Socket socket = server.accept(); + byte[] b = new byte[6]; + readFully(socket.getInputStream(), b); + // block to cause timeout on read. + server.accept(); + } catch (Exception e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + } + }); + assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); + try { + TcpNioConnection connection = factory.getConnection(); + connection.send(MessageBuilder.withPayload("Test").build()); + Thread.sleep(2000); + assertTrue(!connection.isOpen()); + } catch (Exception e) { + fail("Unexptected exception " + e); + } + + } + + private void readFully(InputStream is, byte[] buff) throws IOException { + for (int i = 0; i < buff.length; i++) { + buff[i] = (byte) is.read(); + } + } + +} diff --git a/spring-integration-ip/template.mf b/spring-integration-ip/template.mf index a0ae1b519b..4a374d6c79 100644 --- a/spring-integration-ip/template.mf +++ b/spring-integration-ip/template.mf @@ -5,6 +5,7 @@ Bundle-ManifestVersion: 2 Import-Template: org.apache.commons.logging;version="[1.1.1, 2.0.0)", org.springframework.integration.*;version="[2.0.0, 2.0.1)", + org.springframework.commons.*;version="[2.0.0, 2.0.1)", org.springframework.beans.*;version="[3.0.0, 4.0.0)", org.springframework.context;version="[3.0.0, 4.0.0)", org.springframework.core.*;version="[3.0.0, 4.0.0)",