diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java index 286f6f36ea..73d87d42bf 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/AbstractInternetProtocolSendingMessageHandler.java @@ -82,4 +82,12 @@ public abstract class AbstractInternetProtocolSendingMessageHandler implements M this.soSendBufferSize = size; } + + /** + * @return the port + */ + public int getPort() { + return port; + } + } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java index afa5734521..7476ac003d 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java @@ -77,8 +77,15 @@ public abstract class IpAdapterParserUtils { static final String CUSTOM_SOCKET_READER_CLASS_NAME = "custom-socket-reader-class-name"; -// static final String + static final String CUSTOM_SOCKET_WRITER_CLASS_NAME = + "custom-socket-writer-class-name"; + static final String SO_LINGER = "so-linger"; + + static final String SO_TCP_NODELAY = "so-tcp-nodelay"; + + static final String SO_TRAFFIC_CLASS = "so-traffic-class"; + /** * Adds a constructor-arg to the bean definition with the value diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java index a2383bd33e..5173e1e0b5 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/config/IpOutboundChannelAdapterParser.java @@ -16,17 +16,19 @@ package org.springframework.integration.ip.config; -import org.w3c.dom.Element; - import org.springframework.beans.factory.BeanCreationException; import org.springframework.beans.factory.support.AbstractBeanDefinition; import org.springframework.beans.factory.support.BeanDefinitionBuilder; import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.core.Conventions; import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser; import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.integration.ip.tcp.TcpNetSendingMessageHandler; +import org.springframework.integration.ip.tcp.TcpNioSendingMessageHandler; import org.springframework.integration.ip.udp.MulticastSendingMessageHandler; import org.springframework.integration.ip.udp.UnicastSendingMessageHandler; import org.springframework.util.StringUtils; +import org.w3c.dom.Element; /** * @author Gary Russell @@ -38,25 +40,21 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt String protocol = IpAdapterParserUtils.getProtocol(element); BeanDefinitionBuilder builder = null; if (protocol.equals("tcp")) { - throw new BeanCreationException("tcp not yet supported"); + builder = parseTcp(element); } else if (protocol.equals("udp")) { - String multicast = IpAdapterParserUtils.getMulticast(element); - if (multicast.equals("true")) { - builder = BeanDefinitionBuilder - .genericBeanDefinition(MulticastSendingMessageHandler.class); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, - element, IpAdapterParserUtils.MIN_ACKS_SUCCESS, - "minAcksForSuccess"); - IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, - element, IpAdapterParserUtils.TIME_TO_LIVE, - "timeToLive"); - } - else { - builder = BeanDefinitionBuilder - .genericBeanDefinition(UnicastSendingMessageHandler.class); - } + builder = parseUdp(element); } + IpAdapterParserUtils.addCommonSocketOptions(builder, element); + return builder.getBeanDefinition(); + } + + /** + * @param element + * @param builder + */ + private void addHostAndPortToConstructor(Element element, + BeanDefinitionBuilder builder) { String host = element.getAttribute(IpAdapterParserUtils.HOST); if (!StringUtils.hasText(host)) { throw new BeanCreationException(IpAdapterParserUtils.HOST @@ -65,6 +63,30 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt builder.addConstructorArgValue(host); String port = IpAdapterParserUtils.getPort(element); builder.addConstructorArgValue(port); + } + + /** + * @param element + * @return + */ + private BeanDefinitionBuilder parseUdp(Element element) { + BeanDefinitionBuilder builder; + String multicast = IpAdapterParserUtils.getMulticast(element); + if (multicast.equals("true")) { + builder = BeanDefinitionBuilder + .genericBeanDefinition(MulticastSendingMessageHandler.class); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, + element, IpAdapterParserUtils.MIN_ACKS_SUCCESS, + "minAcksForSuccess"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, + element, IpAdapterParserUtils.TIME_TO_LIVE, + "timeToLive"); + } + else { + builder = BeanDefinitionBuilder + .genericBeanDefinition(UnicastSendingMessageHandler.class); + } + addHostAndPortToConstructor(element, builder); IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder, element, IpAdapterParserUtils.CHECK_LENGTH, true); IpAdapterParserUtils.addConstuctorValueIfAttributeDefined(builder, @@ -91,10 +113,43 @@ public class IpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapt + " must be supplied"); } } - IpAdapterParserUtils.addCommonSocketOptions(builder, element); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, IpAdapterParserUtils.RECEIVE_BUFFER_SIZE); - return builder.getBeanDefinition(); + return builder; + } + + /** + * @param element + * @return + */ + private BeanDefinitionBuilder parseTcp(Element element) { + BeanDefinitionBuilder builder; + String useNio = IpAdapterParserUtils.getUseNio(element); + if (useNio.equals("false")) { + builder = BeanDefinitionBuilder + .genericBeanDefinition(TcpNetSendingMessageHandler.class); + } + else { + builder = BeanDefinitionBuilder + .genericBeanDefinition(TcpNioSendingMessageHandler.class); + } + addHostAndPortToConstructor(element, builder); + builder.addPropertyValue( + Conventions.attributeNameToPropertyName(IpAdapterParserUtils.MESSAGE_FORMAT), + IpAdapterParserUtils.getMessageFormat(element)); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, + IpAdapterParserUtils.CUSTOM_SOCKET_WRITER_CLASS_NAME); + 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); + return builder; } } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java index b705835fb0..8e512a13b9 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpReceivingChannelAdapter.java @@ -57,7 +57,7 @@ public abstract class AbstractTcpReceivingChannelAdapter extends */ public void run() { if (logger.isDebugEnabled()) { - logger.debug(this.getClass().getSimpleName() + " running..."); + logger.debug(this.getClass().getSimpleName() + " running on port: " + port); } if (this.active && this.threadPoolTaskScheduler == null) { this.threadPoolTaskScheduler = new ThreadPoolTaskScheduler(); diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java index 33529a9e1a..bf7499f39c 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/AbstractTcpSendingMessageHandler.java @@ -50,7 +50,7 @@ public abstract class AbstractTcpSendingMessageHandler extends protected boolean soKeepAlive = false; - protected int messageFormat; + protected int messageFormat = MessageFormats.FORMAT_LENGTH_HEADER; protected boolean blockingWrite = false; @@ -111,7 +111,7 @@ public abstract class AbstractTcpSendingMessageHandler extends .newSingleThreadExecutor(new ThreadFactory() { public Thread newThread(Runnable runner) { Thread thread = new Thread(runner); - thread.setName("UDP-Ack-Handler"); + thread.setName("Tcp-NonBlocking-Handler-port-" + port); thread.setDaemon(true); return thread; } @@ -131,9 +131,16 @@ public abstract class AbstractTcpSendingMessageHandler extends protected void doWrite(Message message) { try { byte[] bytes = mapper.fromMessage(message); - this.getWriter().write(bytes); + SocketWriter writer = this.getWriter(); + if (writer == null) { + throw new MessageMappingException("Failed to create SocketWriter"); + } + writer.write(bytes); } catch (Exception e) { writer = null; + if (e instanceof MessageMappingException) { + throw (MessageMappingException) e; + } throw new MessageMappingException("Failed to map message", e); } } diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java index f705aa54cb..b770b17968 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/NioSocketReader.java @@ -230,6 +230,10 @@ public class NioSocketReader extends AbstractSocketReader { protected void readChannel(ByteBuffer buffer) throws IOException { try { int len = channel.read(buffer); + if (len < 0) { + logger.debug("Socket closed"); + throw new IOException("Socket closed"); + } if (logger.isDebugEnabled()) { logger.debug("Read " + len + " bytes, buffer is now at " + buffer.position() + " of " + diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java index cd5646a5a6..abfdd9e04f 100644 --- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java +++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/tcp/TcpNioReceivingChannelAdapter.java @@ -67,7 +67,7 @@ public class TcpNioReceivingChannelAdapter extends try { serverChannel = ServerSocketChannel.open(); serverChannel.configureBlocking(false); - serverChannel.socket().bind(new InetSocketAddress(port)); + serverChannel.socket().bind(new InetSocketAddress(port), 10); final Selector selector = Selector.open(); serverChannel.register(selector, SelectionKey.OP_ACCEPT); doSelect(serverChannel, selector); @@ -107,7 +107,7 @@ public class TcpNioReceivingChannelAdapter extends while (active) { int selectionCount = selector.select(); if (logger.isDebugEnabled()) - logger.debug("SelectionCount: " + selectionCount); + logger.debug("Port " + port + " SelectionCount: " + selectionCount); if (selectionCount > 0) { Set keys = selector.selectedKeys(); Iterator iterator = keys.iterator(); diff --git a/org.springframework.integration.ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd b/org.springframework.integration.ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd index 28343ce950..7c3824eed4 100644 --- a/org.springframework.integration.ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd +++ b/org.springframework.integration.ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.0.xsd @@ -66,6 +66,11 @@ the custom message format. See java docs for TcpNetReceivingChannelAdapter and T + + + + + diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/IpChannelAdapterParserTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/IpChannelAdapterParserTests.java index 6cb80aaf6a..4463172da2 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/IpChannelAdapterParserTests.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/IpChannelAdapterParserTests.java @@ -18,17 +18,29 @@ package org.springframework.integration.ip.config; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.fail; + +import java.lang.reflect.Field; +import java.net.InetSocketAddress; import org.junit.Test; import org.junit.runner.RunWith; +import org.springframework.beans.BeansException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; +import org.springframework.integration.ip.AbstractInternetProtocolReceivingChannelAdapter; +import org.springframework.integration.ip.AbstractInternetProtocolSendingMessageHandler; 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.Utils; import org.springframework.integration.ip.udp.UnicastReceivingChannelAdapter; +import org.springframework.integration.message.MessageBuilder; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -37,7 +49,9 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; * @author Gary Russell * */ -@ContextConfiguration(locations="inboundAdapters.xml") +@ContextConfiguration(locations={"inboundAdapters.xml" + ,"outboundAdapters.xml" + }) @RunWith(SpringJUnit4ClassRunner.class) public class IpChannelAdapterParserTests { @@ -72,11 +86,15 @@ public class IpChannelAdapterParserTests { @Qualifier(value="udp1") UnicastReceivingChannelAdapter udp1; + @Autowired + @Qualifier(value="org.springframework.integration.ip.tcp.TcpNioSendingMessageHandler#0") + TcpNioSendingMessageHandler tcpOut1; + @SuppressWarnings("unchecked") @Test public void testTcpInbound1() { Utils.testSendFragmented(tcp1.getPort()); - Message message = (Message) channel.receive(); + Message message = (Message) channel.receive(10000); assertNotNull(message); assertEquals("xx", new String(message.getPayload())); } @@ -85,7 +103,7 @@ public class IpChannelAdapterParserTests { @Test public void testTcpInbound2() { Utils.testSendFragmented(tcp2.getPort()); - Message message = (Message) channel.receive(); + Message message = (Message) channel.receive(10000); assertNotNull(message); assertEquals("xx", new String(message.getPayload())); } @@ -94,7 +112,7 @@ public class IpChannelAdapterParserTests { @Test public void testTcpInbound3() { Utils.testSendFragmented(tcp3.getPort()); - Message message = (Message) channel.receive(); + Message message = (Message) channel.receive(10000); assertNotNull(message); assertEquals("xx", new String(message.getPayload())); } @@ -103,10 +121,10 @@ public class IpChannelAdapterParserTests { @Test public void testTcpInbound4() { Utils.testSendStxEtx(tcp4.getPort(), null); - Message message = (Message) channel.receive(); + Message message = (Message) channel.receive(10000); assertNotNull(message); assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String(message.getPayload())); - message = (Message) channel.receive(); + message = (Message) channel.receive(10000); assertNotNull(message); assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String(message.getPayload())); } @@ -115,10 +133,10 @@ public class IpChannelAdapterParserTests { @Test public void testTcpInbound5() { Utils.testSendCrLf(tcp5.getPort(), null); - Message message = (Message) channel.receive(); + Message message = (Message) channel.receive(10000); assertNotNull(message); assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String(message.getPayload())); - message = (Message) channel.receive(); + message = (Message) channel.receive(10000); assertNotNull(message); assertEquals(Utils.TEST_STRING + Utils.TEST_STRING, new String(message.getPayload())); } @@ -127,11 +145,11 @@ public class IpChannelAdapterParserTests { @Test public void testTcpInbound6() { Utils.testSendStxEtx(tcp6.getPort(), null); - Message message = (Message) channel.receive(); + Message message = (Message) channel.receive(10000); assertNotNull(message); assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", new String(message.getPayload())); - message = (Message) channel.receive(); + message = (Message) channel.receive(10000); assertNotNull(message); assertEquals("\u0002" + Utils.TEST_STRING + Utils.TEST_STRING + "\u0003", new String(message.getPayload())); @@ -141,4 +159,34 @@ public class IpChannelAdapterParserTests { public void testUdpInbound1() { assertNotNull(udp1); } + + @SuppressWarnings("unchecked") + @Test + public void testTcpOutbound1() { + setPort(tcpOut1, tcp1); + Message message = MessageBuilder.withPayload("TESTING").build(); + tcpOut1.handleMessage(message); + Message mOut = (Message) channel.receive(10000); + assertNotNull(mOut); + assertEquals("TESTING", new String(mOut.getPayload())); + + } + + private void setPort(AbstractInternetProtocolSendingMessageHandler tcpSMA, + AbstractInternetProtocolReceivingChannelAdapter tcpRCA) { + try { + int port = tcpRCA.getPort(); + Field portField = tcpSMA.getClass().getSuperclass().getSuperclass().getDeclaredField("port"); + portField.setAccessible(true); + assertEquals(9999, portField.getInt(tcpSMA)); + portField.setInt(tcpSMA, port); + InetSocketAddress address = new InetSocketAddress("localhost", port); + Field addressField = tcpSMA.getClass().getSuperclass().getSuperclass().getDeclaredField("destinationAddress"); + addressField.setAccessible(true); + addressField.set(tcpSMA, address); + } catch (Exception e) { + fail("Couldn't fix port:" + e); + } + } + } diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/inboundAdapters.xml b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/inboundAdapters.xml index c18000503e..daa22359f3 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/inboundAdapters.xml +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/inboundAdapters.xml @@ -9,12 +9,14 @@ http://www.springframework.org/schema/integration/spring-integration.xsd http://www.springframework.org/schema/integration/ip http://www.springframework.org/schema/integration/ip/spring-integration-ip.xsd"> - + + + diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/outboundAdapters.xml b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/outboundAdapters.xml new file mode 100644 index 0000000000..6e333284f7 --- /dev/null +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/config/outboundAdapters.xml @@ -0,0 +1,33 @@ + + + + + + + + + + + + + + diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/Utils.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/Utils.java index 85fc214b60..90a1f3ad5a 100644 --- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/Utils.java +++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/tcp/Utils.java @@ -78,6 +78,7 @@ public class Utils { Thread thread = new Thread(new Runnable() { public void run() { try { + System.out.println("Connecting to " + port); Socket socket = new Socket(InetAddress.getByName("localhost"), port); OutputStream os = socket.getOutputStream(); writeByte(os, 0); @@ -168,8 +169,8 @@ public class Utils { thread.start(); } - public static int findAvailableServerSocket() { - for (int i = 5678; i < 5878; i++) { + public static int findAvailableServerSocket(int seed) { + for (int i = seed; i < seed+200; i++) { try { ServerSocket sock = ServerSocketFactory.getDefault().createServerSocket(i); sock.close(); @@ -178,4 +179,8 @@ public class Utils { } throw new RuntimeException("Cannot find a free server socket"); } + + public static int findAvailableServerSocket() { + return findAvailableServerSocket(5678); + } }